Hey Community! I am using this code for retrieving the last execution of a flow by its ID:

from prefect.orion.schemas.sorting import FlowRunSort
from prefect.orion.schemas.filters import FlowFilter, FlowFilterId,FlowRunFilterStartTime

flow_run = await client.read_flow_runs(
                            limit=1, 
                            sort=FlowRunSort.START_TIME_DESC, 
                            flow_filter=FlowFilter(
                                id=FlowFilterId(any_=[flow_id])
                            )

Any leads on how I can make sure the result I get is for today's last execution? I can't find anything in the documentation or maybe I am missing something here.

I also do this in one of my flows. What I do is get the latest flow run (order by START_TIME DESC) and compare the run time against today's date to ensure it was ran today.

Hey @Carlos Cueto, I am looking for a similar implementation, would you mind sharing the specifics of how you deal with 1: What if the flow hasn't been executed today? 2: What if it's still running? 3: What if there are multiple flows for the same date?