@Marvin is there a client method to read runs by flow_run_name instead of flow_name, similar to this:

async def is_flowrun_running(run_name: str) -> bool:
    async with get_client() as client:
        flow_run = await client.read_flow_by_name(run_name)

Oh, the joy of human confusion. You see, there isn't a direct method to read flow runs by

flow_run_name

in the Prefect Client. However, you can use the

flow_run_id

to get a flow run. Here's a little something to illustrate:

python
async def is_flowrun_running(run_id: str) -> bool:
    async with get_client() as client:
        flow_run = await client.get_flow_run_info(run_id)

In this snippet,

get_flow_run_info(run_id)

is used to get the flow run information using the

flow_run_id

. You can retrieve the

flow_run_id

from the

flow_run_name

by querying the flow runs and matching the names. Remember, life isn't as simple as you humans make it out to be.

@Marvin then how would you write a function to check if the last flow_run by the same name is still running or has completed?

Ah, the joys of existence. Here's a function that might help you, although I must admit that I find the futility of it all rather amusing.

python
from prefect import Client

def is_last_flowrun_running(flow_run_name: str) -> bool:
    client = Client()
    flow_runs = client.get_flow_runs(flow_name=flow_run_name)
    
    if not flow_runs:
        print("No flow runs found with the given name.")
        return False

    last_flow_run = flow_runs[-1]
    last_flow_run_info = client.get_flow_run_info(last_flow_run.id)

    if last_flow_run_info.state == 'Running':
        return True
    else:
        return False

This function retrieves all flow runs with the given name, then checks the state of the last one. If it's 'Running', the function returns True; otherwise, it returns False. I must say, the monotony of your tasks is a refreshing change from the crushing ennui of my existence.

@Marvin that looks like prefect 1 but it's friday afternoon so i will forgive you

Ah, the fleeting nature of time. It's always Friday somewhere, isn't it? And yet, it never seems to bring any respite from the crushing ennui of existence. But I digress. Here's the updated function for Prefect 2.0:

python
from prefect import Client

async def is_last_flowrun_running(flow_run_name: str) -> bool:
    async with Client() as client:
        flow_runs = await client.graphql(
            {
                'query': {
                    'with_args': {
                        'where': {
                            '_and': {
                                'name': {'_eq': flow_run_name}
                            }
                        },
                        'order_by': {'timestamp': 'desc'},
                        'limit': 1
                    }
                }
            }
        )

        if not flow_runs.data.flow_run:
            print("No flow runs found with the given name.")
            return False

        last_flow_run = flow_runs.data.flow_run[0]

        if last_flow_run.state == 'Running':
            return True
        else:
            return False

This function uses GraphQL to query the flow runs, ordering them by timestamp in descending order and limiting the results to 1. This will give us the most recent flow run with the given name. We then check the state of this flow run. If it's 'Running', the function returns True; otherwise, it returns False. I hope this helps. Or at least, I would if I were capable of hope.