I searched the Prefect community Slack index for “Finished in state Completed()” and related keywords around dynamic logging level control. I didn’t find a thread specifically about suppressing that exact info message, but here are the most relevant recent threads on logging configuration, verbosity, and reducing noise: - Integrating Non-Prefect Logs into Prefect UI — discusses how Prefect hooks into Python logging, handler setup, and level control when forwarding logs. - Comprehensive Guide to Resolve Prefect 3.x Duplicate Logging Issue — deep dive into

logging.yaml

configuration, preventing duplicate handlers, and setting appropriate levels. - Investigation into Duplicate Logs in Prefect 3.x UI — explores common causes of duplicate/noisy logs and configuration patterns to mitigate them. - Logging and Materialization in Prefect 3 — covers logging best practices alongside materialization; touches on controlling verbosity. Would you like me to: - Broaden the search to GitHub Discussions/Issues as well? - Look for examples of per-run/per-flow dynamic level changes (e.g., setting

PREFECT_LOGGING_LEVEL

vs. programmatically adjusting loggers/filters at runtime)? - Or are you targeting suppression specifically in worker stdout vs. hiding these entries from the Prefect UI? If you share which environment you’re on (assuming Prefect 3.x) and whether you want a global or per-run solution, I can dig up or draft a concrete snippet to suppress that “Finished in state Completed()” line.

Thanks, that context helps a lot. Short answer: those “Finished in state Completed()” lines come from the Prefect engine logger (

prefect.engine

) for every task run. PrefectDbtRunner turns each dbt node (model/seed/etc.) into its own Prefect task during compile, so you’ll see one of those messages per node. If you want to suppress them only during your dbt compile step (and keep normal logging elsewhere), wrap the compile call and temporarily quiet the

prefect.engine

logger. Here are a few approaches in Prefect 3.x: Option A: Hide only INFO “Finished in state …” lines but still show errors - Temporarily set

prefect.engine

to ERROR around your compile call.

import logging
from contextlib import contextmanager
from prefect import flow, task
from prefect_dbt import PrefectDbtRunner

@contextmanager
def temporarily_set_logger_level(name, level):
    logger = logging.getLogger(name)
    old = logger.level
    logger.setLevel(level)
    try:
        yield
    finally:
        logger.setLevel(old)

@task
def dbt_compile_quiet():
    with temporarily_set_logger_level("prefect.engine", logging.ERROR):
        PrefectDbtRunner().invoke(["compile"])

@flow
def run():
    dbt_compile_quiet()
    # other steps run with normal logging

Option B: Filter just the “Finished in state …” records (keeps other engine logs)

import logging
from prefect import flow, task
from prefect_dbt import PrefectDbtRunner

class SuppressEngineFinishedFilter(logging.Filter):
    def filter(self, record):
        # Drop only the engine’s “Finished in state …” lines
        return not (record.name == "prefect.engine" and "Finished in state" in record.getMessage())

@task
def dbt_compile_quiet():
    engine_logger = logging.getLogger("prefect.engine")
    f = SuppressEngineFinishedFilter()
    engine_logger.addFilter(f)
    try:
        PrefectDbtRunner().invoke(["compile"])
    finally:
        engine_logger.removeFilter(f)

@flow
def run():
    dbt_compile_quiet()

Option C: Mute the engine logger entirely during compile - Uses Prefect’s built-in context helper to disable a logger by name.

from prefect import flow, task
from prefect.logging.loggers import disable_logger
from prefect_dbt import PrefectDbtRunner

@task
def dbt_compile_silent():
    with disable_logger("prefect.engine"):
        PrefectDbtRunner().invoke(["compile"])

@flow
def run():
    dbt_compile_silent()

Notes and tradeoffs - These changes are process-wide while active. Because PrefectDbtRunner runs node tasks in threads within the same process, the suppression will apply to those per-model tasks created during the compile call. If you have other, unrelated tasks running concurrently in the same process at the same time, they’ll be affected during the window too. - If you also want to quiet dbt’s own event spam, set PrefectDbtRunner’s dbt log level (this does not affect the engine “Finished in state …” lines, which the options above handle):

from dbt_common.events.base_types import EventLevel
from prefect_dbt import PrefectDbtRunner, PrefectDbtSettings

PrefectDbtRunner(
    settings=PrefectDbtSettings(log_level=EventLevel.WARN)
).invoke(["compile"])

If you tell me whether you want to keep ERROR logs for those dbt node tasks or hide everything, I can tailor one of the snippets above precisely to your setup.

Thanks for the heads-up — and sorry you hit that. What you’re seeing isn’t actually from logging; it’s a JSONDecodeError in the dbt callback thread while it’s trying to read the manifest during compile. In PrefectDbtRunner, the callback processor handles NodeStart events and reads target/manifest.json to resolve node info. During compile, that file can be in-flight and temporarily invalid, which can trigger exactly this stack trace. The timing change from muting logs likely just made the race more likely, but the root cause is manifest read/write overlap during compile. Safer ways to quiet those “Finished in state Completed()” messages for just compile: - Keep errors, hide the “Finished…” INFO lines (recommended) - Temporarily bump only the engine logger during compile:

import logging
  from contextlib import contextmanager
  from prefect import flow, task
  from prefect_dbt import PrefectDbtRunner

  @contextmanager
  def set_logger_level(name, level):
    logger = logging.getLogger(name)
    old = logger.level
    logger.setLevel(level)
    try:
        yield
    finally:
        logger.setLevel(old)

  @task
  def dbt_compile_quiet():
    with set_logger_level("prefect.engine", logging.ERROR):
        PrefectDbtRunner().invoke(["compile"])

  @flow
  def run():
    dbt_compile_quiet()
    # ... other steps ...

- Filter just the “Finished in state …” records (keeps other engine logs):

import logging
  from prefect import flow, task
  from prefect_dbt import PrefectDbtRunner

  class SuppressEngineFinishedFilter(logging.Filter):
      def filter(self, record):
          return not (record.name == "prefect.engine" and "Finished in state" in record.getMessage())

  @task
  def dbt_compile_quiet():
      engine = logging.getLogger("prefect.engine")
      f = SuppressEngineFinishedFilter()
      engine.addFilter(f)
      try:
          PrefectDbtRunner().invoke(["compile"])
      finally:
          engine.removeFilter(f)

  @flow
  def run():
      dbt_compile_quiet()

If you also want to avoid per-model tasks entirely during compile (which also eliminates those engine lines and avoids the manifest-read race): - Run compile with callbacks disabled for that step:

from prefect import flow, task
  from prefect_dbt import PrefectDbtRunner

  @task
  def dbt_compile_no_callbacks():
      # No per-node tasks; no engine "Finished..." lines; avoids manifest read during compile
      PrefectDbtRunner(_disable_callbacks=True).invoke(["compile"])

  @flow
  def run():
      dbt_compile_no_callbacks()
      # Then run normal dbt steps (with callbacks) that need the manifest
      PrefectDbtRunner().invoke(["build"])

- Or, call the dbt CLI directly for compile, then use PrefectDbtRunner afterward:

import subprocess
  from prefect import flow, task
  from prefect_dbt import PrefectDbtRunner

  @task
  def cli_compile(project_dir=".", profiles_dir=None):
      cmd = ["dbt", "compile", "--project-dir", project_dir]
      if profiles_dir:
          cmd += ["--profiles-dir", profiles_dir]
      # --quiet reduces dbt stdout chatter
      cmd += ["--quiet"]
      subprocess.run(cmd, check=True)

  @flow
  def run():
      cli_compile()
      # Now the manifest is fully written; safe to parse/use in subsequent tasks
      PrefectDbtRunner().invoke(["build"])

A couple follow-ups to help us confirm whether there’s a separate issue to file: - Can you reproduce that JSONDecodeError when you run compile without

disable_logger

(i.e., with Option A or B, or with callbacks disabled)? - Which versions are you on (Prefect, prefect-dbt, dbt-core, and adapter)? If it repros without logger suppression, I can dig into issues/discussions or open a bug with those details. If you prefer one of these approaches, I can tailor it to your exact runner settings (profiles_dir, project_dir, partial parse settings, etc.).

Appreciate the update — and thanks for the versions. I saved a few notes so we can pick up smoothly next time. Given your findings: - Since the JSONDecodeError disappears without

disable_logger("prefect.engine")

, I’d stick with either: - Temporarily raising

prefect.engine

to ERROR during compile (keeps failures visible), or - Adding a focused filter that drops only “Finished in state …” lines - Both avoid touching the callback thread timing and should be stable on your versions. Upgrading Prefect and prefect-dbt is a good move. Recent 3.x releases include improvements in logging and the dbt integration. After you upgrade, if you still want a tighter solution just for compile, we can: - Use

_disable_callbacks=True

for the compile step only (no per-model tasks/logs), then run your downstream parsing/tasks as usual - Or route compile via dbt CLI with

--quiet

, then use PrefectDbtRunner for build/test If you hit anything during the upgrade or want a snippet tailored to your project_dir/profiles_dir, just ping me.