Custom flow decorator broken starting with Prefect 2.19

[LeeMendelowitz]

Bug summary

We use a custom flow decorator to automatically adjust flow names, descriptions, and versions, and deploy using a prefect.yaml. We found this method very helpful for standardizing how we name flows and ensures our repo URL's are added to every flow description. Our custom decorator no longer works starting with Prefect 2.19.

Digging into the prefect code base, it looks like the load_flow_argument_from_entrypoint function introduced in prefect 2.19 breaks a behavior we were relying on. This function load_flow_argument_from_entrypoint does not properly extract attributes that are manipulated by our custom flow decorator.

We use our customflow decorator just like prefect.flow:

from internal_package import flow

@flow
def my_flow():
    pass

where internal_package defines the flow decorator as:

import prefect.flows
from prefect.flows import Flow
from prefect import task, get_run_logger
import os
from functools import wraps, partial
import inspect

project_name = os.environ["PROJECT_NAME"]
git_url = os.environ["GIT_URL"]

@task
def init_task():
    logger = get_run_logger()
    logger.info("this task always runs at the start of every flow")

def get_git_hash_version():
    return 'abc'


def annotate_flow(flow: Flow):
    """
     - Prepend the project name to the flow name
     - Add the git url to the flow description
     - Set the flow version to the git hash
    """

    if not isinstance(flow, Flow):
        raise TypeError("flow must be a prefect Flow")

    if project_name is not None:
        # Prepend the project name to the flow name.
        pfx = f'{project_name}: '
        if not flow.name.startswith(pfx):
            flow.name = pfx + flow.name

    if git_url is not None:
        description_suffix = f"\nGit Repo: {git_url}"
        description = (flow.description or "")
        if description_suffix not in description:
            flow.description = description + description_suffix

    # Set the version to the Git Hash.
    git_hash = get_git_hash_version()
    if git_hash:
        flow.version = f'git:{git_hash}'

    return flow


def flow(fn = None, **kwargs):
    """
    A substitute for prefect's @flow decorator which:
     - Calls ecs_container_info() as the first task.
     - Annotates the flow with `annotate_flow`

    Any keyword arguments are passed to prefect.flows.Flow
    """

    if fn is None:
        return partial(flow, **kwargs)

    @wraps(fn)
    def wrapped_flow(*args, **kwargs):

        # Call the init_task as the first task in the flow.
        init_task()

        return fn(*args, **kwargs)

    wrapped_flow.__signature__ = inspect.signature(fn)

    # Create a flow with the provided function
    my_flow = Flow(wrapped_flow, **kwargs)
    my_flow = annotate_flow(my_flow)

    return my_flow

Version info (prefect version output)

Version:             2.19.4
API version:         0.8.4
Python version:      3.11.6
Git commit:          867543a8
Built:               Tue, Jun 4, 2024 3:14 PM
OS/Arch:             darwin/arm64
Profile:             test
Server type:         cloud

Additional context

No response

[desertaxle]

Hey @LeeMendelowitz! I'm pretty sure this bug was fixed in #14782 which was released in 2.20.0. Could you try upgrading to 2.20.0 and see if the issue persists?

[LeeMendelowitz]

Thanks for the quick response @desertaxle! From the quick testing I've done it looks like prefect 2.20 solves our issue. I'll close the issue.