Yes — you can work with dbt results programmatically, but there’s an important distinction: PrefectDbtRunner sends dbt events into Prefect logging, which is why you see them in the UI, but if you want to build a report in Python, the most reliable thing to consume is usually the dbt run result object, not the emitted log lines. For

dbt test

, that’s often enough because failed/warned tests are reflected in the returned result data. A good pattern is: 1. run

dbt test

with

raise_on_failure=False

2. inspect the returned

dbtRunnerResult

3. extract failed/error test results 4. build a report dict/string 5. pass that report to another task/flow Example:

python
from prefect import flow, task
from prefect_dbt import PrefectDbtRunner

@task
def run_dbt_tests():
    runner = PrefectDbtRunner(raise_on_failure=False)
    result = runner.invoke(["test"])

    report_items = []

    if result.result and hasattr(result.result, "results"):
        for test_result in result.result.results:
            status = str(test_result.status).lower()
            message = getattr(test_result, "message", "") or ""

            # Keep only warning-ish / failure-ish outcomes
            if "warn" in status or "fail" in status or "error" in status:
                report_items.append(
                    {
                        "node_id": test_result.unique_id,
                        "status": status,
                        "message": message,
                        "execution_time": getattr(test_result, "execution_time", None),
                    }
                )

    return {
        "success": result.success,
        "items": report_items,
    }

@task
def send_report(report: dict):
    if not report["items"]:
        return

    lines = ["dbt test report:"]
    for item in report["items"]:
        lines.append(
            f"- {item['node_id']} | {item['status']} | {item['message']}"
        )

    rendered_report = "\n".join(lines)

    # here you can pass it to another flow/task/email sender
    print(rendered_report)

@flow
def main():
    report = run_dbt_tests()
    send_report(report)

A few important notes: *

raise_on_failure=False

is important, otherwise

invoke(["test"])

may raise before you can inspect the results. *

result.success

tells you whether dbt considered the command successful overall. *

result.result.results

contains per-test/per-node results that are much easier to process than raw log messages. If you specifically need the actual log messages with severity

WARNING

or higher, you can also attach a Python logging handler and capture Prefect-emitted log records during the run:

python
import logging
from prefect import flow
from prefect_dbt import PrefectDbtRunner

class WarningCaptureHandler(logging.Handler):
    def __init__(self):
        super().__init__(level=logging.WARNING)
        self.records = []

    def emit(self, record):
        if record.levelno >= logging.WARNING:
            self.records.append(
                {
                    "level": record.levelname,
                    "message": record.getMessage(),
                }
            )

@flow
def main():
    handler = WarningCaptureHandler()
    logger = logging.getLogger("prefect")
    logger.addHandler(handler)

    try:
        runner = PrefectDbtRunner(raise_on_failure=False)
        result = runner.invoke(["test"])

        warning_report = handler.records
        print(warning_report)
        return {
            "dbt_success": result.success,
            "warnings_and_errors": warning_report,
        }
    finally:
        logger.removeHandler(handler)

In practice: * If you need a quality report about dbt test outcomes, use

result.result.results

. * If you need the exact rendered log lines shown in the UI, capture Python log records with a handler. If you want, I can give you a complete example that:

separates WARNING vs ERROR, formats an email-friendly report, and passes it into another flow with

run_deployment

or a subflow call.

Yes — the simplest way is to attach your own Python

logging.Handler

while the flow runs and collect the log records. Since Prefect flow/task logs are standard Python logging records underneath, you can capture them in memory and then process them however you want. Example:

python
import logging
from prefect import flow, get_run_logger
from prefect_dbt import PrefectDbtRunner

class ListHandler(logging.Handler):
    def __init__(self):
        super().__init__()
        self.records = []

    def emit(self, record):
        self.records.append(
            {
                "level": record.levelname,
                "message": record.getMessage(),
                "logger": record.name,
            }
        )

@flow
def my_flow():
    handler = ListHandler()

    # Attach to root logger to capture everything emitted during the flow
    root_logger = logging.getLogger()
    root_logger.addHandler(handler)

    try:
        logger = get_run_logger()
        <http://logger.info|logger.info>("Flow started")

        runner = PrefectDbtRunner(raise_on_failure=False)
        result = runner.invoke(["test"])

        <http://logger.info|logger.info>(f"dbt success: {result.success}")

        # Now you have all log messages in memory
        return handler.records

    finally:
        root_logger.removeHandler(handler)

That will give you a list like:

python
[
    {"level": "INFO", "message": "Flow started", "logger": "prefect.flow_runs"},
    {"level": "WARNING", "message": "...", "logger": "prefect.task_runs"},
]

A couple of caveats: * This captures Python log records emitted in the current process. * It’s the easiest approach if your dbt run happens in the same flow process. * If logs are emitted from other infrastructure/processes/workers, this won’t automatically fetch them back from the Prefect API. * If you only want Prefect flow/task logs, attach to a more specific logger like

prefect

instead of the root logger. Slightly narrower version:

python
prefect_logger = logging.getLogger("prefect")
prefect_logger.addHandler(handler)

If your real goal is “get exactly the logs from this flow run after execution”, Prefect does not provide a simple in-flow API like

flow_run.get_logs()

for local programmatic access during the run. So using a custom logging handler is usually the most straightforward solution. Docs: * prefect-dbt integration docs * Prefect logging guide If you want, I can also show you a small reusable utility/context manager like

capture_logs()

so you can do:

python
with capture_logs() as logs:
    runner.invoke(["test"])