Empty flow run graph and no logs in the Prefect UI when decorating @flow

[Moortiii]

Bug summary

Similar results as in #15153, but the cause seems to be unrelated.

Decorating a @flow in Prefect >= 3 results in an empty flow run graph, and no logs are shown in the Prefect UI. The contents of the decorator do not matter, and the bug occurs using both @flow and @flow() syntax.

Minimal reproducible example:

1. Start a local Prefect server with prefect server start
2. Run the file below using PREFECT_API_URL=http://127.0.0.1:4200/api python3 example.py

import functools
from prefect import task, flow, serve

handler_ids = []


def dummy_decorator(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)

    return wrapper


@task(log_prints=True)
@dummy_decorator
def task_1():
    print("This was logged in Task 1")


@task(log_prints=True)
@dummy_decorator
def task_2():
    print("This was logged in Task 2")


@flow(log_prints=True)
@dummy_decorator
async def decorated_flow():
    task_1()
    task_2()


@flow(log_prints=True)
async def example_flow():
    task_1()
    task_2()


if __name__ == "__main__":
    serve(
        decorated_flow.to_deployment(name="decorated-flow"),
        example_flow.to_deployment(name="example-flow"),
    )

Running the flow decorated-flow yields an empty flow graph and no logs. Running the flow example-flow yields the expected flow graph and the expected logs. This issue was not present in Prefect 2.*, see additional context.

Version info

Version:             3.1.5
API version:         0.8.4
Python version:      3.13.0
Git commit:          3c06654e
Built:               Mon, Dec 2, 2024 6:57 PM
OS/Arch:             linux/x86_64
Profile:             local
Server type:         ephemeral
Pydantic version:    2.10.2
Server:
  Database:          sqlite
  SQLite version:    3.46.0

Additional context

Decorated flow run example run:

Non-decorated flow run example run:

The issue was discovered while trying to configure loguru and Prefect, using the following decorator, which worked in Prefect 2.*, but results in the mentioned issues in Prefect 3.*:

handler_ids: list[int] = []
 
def configure_logger():
    global handler_ids

    try:
        # Passthrough in case we are outside of a task / flow context, e.g. `.fn`
        run_logger = get_run_logger()
    except Exception:
        return

    for handler_id in handler_ids:
        try:
            logger.remove(handler_id)
        except ValueError:
            pass

    log_format = "{message}"

    handler_ids.append(
        logger.add(
            run_logger.debug,
            filter=lambda record: record["level"].name == "DEBUG",
            level="TRACE",
            format=log_format,
        )
    )

    handler_ids.append(
        logger.add(
            run_logger.warning,
            filter=lambda record: record["level"].name == "WARNING",
            level="TRACE",
            format=log_format,
        )
    )

    handler_ids.append(
        logger.add(
            run_logger.error,
            filter=lambda record: record["level"].name == "ERROR",
            level="TRACE",
            format=log_format,
        )
    )

    handler_ids.append(
        logger.add(
            run_logger.critical,
            filter=lambda record: record["level"].name == "CRITICAL",
            level="TRACE",
            format=log_format,
        )
    )

    handler_ids.append(
        logger.add(
            run_logger.info,
            filter=lambda record: record["level"].name
            not in ["DEBUG", "WARNING", "ERROR", "CRITICAL"],
            level="TRACE",
            format=log_format,
        )
    )


def forward_logs(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        configure_logger()
        return func(*args, **kwargs)

    return wrapper

[desertaxle]

Thanks for the bug report @Moortiii! This is quite surprising. Does changing the order of the decorators have any effect?

[Moortiii]

Sadly, switching the decorators around doesn't work, since @flow returns an instance of <class 'prefect.flows.Flow'>, and not a plain function that we can further decorate. I agree, the behavior is really strange!