prefect register --project yourproject -p yourflow.py
prefect register --project yourproject -p tests/test_yourflow.py

import cdpcommon.prefect.flow_helpers as helpers
from typing import List, Optional
from cdpprefect.flows import minimal_test as minimal_test1
from typing import List, Optional
from unittest import mock


def flow(
    register_settings: Optional[helpers.Configuration],
    state_handlers: Optional[List],
):
    print("state_handler: ", state_handlers)
    minimal_test1.dummy2 = mock.Mock(return_value=None)
    minimal_test1.flow(register_settings, state_handlers)


flows = helpers.initialise(__file__, flow)

helpers.Configuration

"""
This module provides helpers for using flows in the cloud data platform
"""
import importlib
import os
from collections import namedtuple
from pathlib import Path
from typing import Any, Callable, Dict, Tuple, Optional

from cdpsettings.settings import settings
from config import Configuration
from prefect import Flow
from prefect.executors import LocalDaskExecutor
from prefect.run_configs import UniversalRun
from prefect.storage import Module

RUN_TIME_FORMAT = "%Y-%m-%dT%H:%M:%S.%f%z"


def _get_state_handler_func(handler_path: str):
    module_name, func_name = handler_path.rsplit(".", 1)
    return getattr(importlib.import_module(f"cdpprefect.{module_name}"), func_name)


def _extract_relative_path(path: str, parent_level: int = 0) -> str:
    rel_path_with_os_sep = os.path.relpath(
        path, start=Path(os.path.dirname(path)).parents[parent_level]
    )
    rel_path_with_os_sep_no_suffix = rel_path_with_os_sep.split(".")[0]
    return rel_path_with_os_sep_no_suffix.replace(os.sep, ".")


def initialise(module_file: str, flow_function: Callable) -> Tuple[Flow, ...]:
    """Initialise a flow module with settings defined in the CDP settings file.
    In almost all cases invoke using
    `flows = helpers.initialise(__file__, flow)`

    :param module_file: The module file containing the flow
    :type module_file: str
    :param flow_function: \
        A function defining the Flow according to Prefect's functional API
    :type flow_function: Callable
    :return: A named tuple with a named flow for each configuration defined in settings
    :rtype: Tuple[Flow, ...]
    """
    prefect_settings: Configuration = settings["prefect"]
    print("prefect_settings: ", prefect_settings)
    module_name: str = _extract_relative_path(module_file, parent_level=1)
    flow_key: str = _extract_relative_path(module_file)
    registrations: Configuration = prefect_settings[flow_key]
    number_of_workers: int = int(
        prefect_settings["local_dask_executor.number_of_workers"]
    )

    flows_dict: Dict[str, Any] = {}

    for r in registrations:
        registration = str(r)
        flow_name = f"{module_name}:flows.{registration}"
        flow_registrations = registrations[registration]
        state_handlers = registrations[registration]["state_handlers"]
        optional_flow_args = {}
        if state_handlers is not None:
            optional_flow_args["state_handlers"] = [
                _get_state_handler_func(handler_path) for handler_path in state_handlers
            ]
        with Flow(
            flow_name,
            run_config=UniversalRun(labels=[settings["pipeline.environment_prefix"]]),
            storage=Module(flow_name),
            executor=LocalDaskExecutor(
                scheduler="threads", num_workers=number_of_workers
            ),
            **optional_flow_args,
        ) as f:
            flow_function(
                register_settings=flow_registrations["register_settings"],
                state_handlers=state_handlers,
            )
            print("register_settings: ", flow_registrations["register_settings"])
            print("state_handlers: ", state_handlers)
            flows_dict[registration] = f.copy()
            print("flows_dict: ", flows_dict)

    print(namedtuple("flows", flows_dict.keys())(*flows_dict.values()))
    return namedtuple("flows", flows_dict.keys())(*flows_dict.values())

from prefect import task, Flow
import prefect

def my_state_handler(Flow, old_state, new_state):
    if new_state.is_finished():
        <http://prefect.context.logger.info|prefect.context.logger.info>("it ran")
    return

@task
def task1():
    return 1

with Flow("skip") as flow:
    task1()


from unittest.mock import MagicMock
dummy = MagicMock()
flow.state_handlers = [dummy]
flow.run()

minimal_test1

cdpprefect.flows

with Flow(
            flow_name,
            run_config=UniversalRun(labels=[settings["pipeline.environment_prefix"]]),
            storage=Module(flow_name),
            executor=LocalDaskExecutor(
                scheduler="threads", num_workers=number_of_workers
            ),
            **optional_flow_args,
        ) as f:
            flow_function(
                register_settings=flow_registrations["register_settings"],
                state_handlers=state_handlers,
            )
            print("register_settings: ", flow_registrations["register_settings"])
            print("state_handlers: ", state_handlers)
            flows_dict[registration] = f.copy()
            print("flows_dict: ", flows_dict)