update prefect version

[xiaoyachong]

I upgrade prefect 2.14.21 to prefect 3.4.2, which is consistent with David's splash project.

I add mlflow utils to this repo.

[kkamdin]

This looks good to me! I'm not very educated (yet!) on writing prefect tests so I may have missed something. I just had one question and a note about building common-use prefect code in in-line comments.

[kkamdin]

just so I understand -- can you explain why you no longer need the asyncio?

[xiaoyachong]

Thanks for your comments. Previously, in run_flow(), we invoked the async function create_then_begin_flow_run. Since this function is now deprecated, we instead call parent_flow inline. As a result, we no longer need to use asyncio.run when calling run_flow.

[kkamdin]

if we want to move towards commonly used prefect code, we could try renaming MLexchange-specific parameters like "model_name" to something more generic. Since parent flow is basically a test fixture here (I think?), maybe that's less useful. I think in prefect_utils/core.py there are some candidates, but that can be a job for future us.

[xiaoyachong]

run_flow() is a helper function created to simplify testing. It runs a flow inline and returns a flow ID, which we can then use to test operations such as delete, cancel, or fetching logs.

In real applications, we do not run flows inline this way. Instead, only schedule_prefect_flow (defined in core.py) is used. This function uses more general parameter names:

def schedule_prefect_flow(
    deployment_name: str,
    parameters: Optional[dict] = None,
    flow_run_name: Optional[str] = None,
    tags: Optional[list] = [],
)

It also returns a flow ID.

[davramov]

This assert statement is commented out now. Is there a reason to keep it, or is the next assertion comprehensive enough?

[xiaoyachong]

Thanks for your comments.
The assert len(flow_run_logs) > 0 check is important to ensure we successfully retrieve logs. However, in Prefect 3 with pytest, len(flow_run_logs) is always 0.

It looks like pytest changed between Prefect 2 and Prefect 3.
Prefect 2:

Prefect 3:

From the above images, you may notice that Prefect 3 uses a temporary server with no persistent log storage, whereas Prefect 2 saved logs in a local database by default.

When I tested against a real Prefect server, I was able to retrieve logs using the get_flow_runs_logs() API, but this no longer works in pytest. Since I couldn’t find a way to test it reliably, I delete the assertion for now.

Have you run into a similar issue?

[davramov]

In the other tests for flow_run_id, it is checking that it is a str. Here, it is checking if flow_run_id is a uuid.UUID. Super nitpicky, but maybe these should be consistent? I am not sure if other code assumes the id to be a str or uuid.

[xiaoyachong]

That sounds good. I’ve updated them to use the same type: uuid.UUID.

[davramov]

I left a couple of comments with minor suggestions; otherwise, it looks good!

[xiaoyachong]

I left a couple of comments with minor suggestions; otherwise, it looks good!

Thank you, David.
Would you like any additional functions to be added to prefect/core.py?
Also, would it be feasible to import this package in the Splash project so that you can retire your own prefect.py (https://github.com/als-computing/splash_flows/blob/main/orchestration/prefect.py)?