https://prefect.io logo
Join Slack
Powered by
I'm running into a funny problem using the prefect...
# ask-community
p
I'm running into a funny problem using the prefect client api. I'd like to programmatically query prefect to see the flow runs of mine that are running and tagged a certain way, but it doesn't seem like the filtering works on states or tags the way it works on ids. code-snippet below:
I create a flow programattically like so:
Copy code
deployment_filter = DeploymentFilter(
    name=DeploymentFilterName(any_=["deployment-name"])  
)

#get the deployment from the name, and then create a run! 
async with prefect.get_client() as client:

    response = await client.read_deployments(deployment_filter=deployment_filter)
    #print(response[0].id)
    run = await client.create_flow_run_from_deployment(response[0].id, tags=["my-tag"])

print(run)
I can manually check the state of this run successfully like so:
Copy code
this_run_filter = FlowRunFilter(
    id=FlowRunFilterId(any_=[run.id])
)
async with prefect.get_client() as client:

    run_info = await client.read_flow_runs(flow_run_filter=this_run_filter)
    print(run_info[0].state_type == StateType.RUNNING or 
          run_info[0].state_type == StateType.SCHEDULED or 
          run_info[0].state_type == StateType.PENDING)
But, trying to query all of my flow runs that are scheduled/running/pending doesn't quite work- it just seems as though zero filters are applied.
Copy code
#Get running and tagged flow runs
running_filter = FlowRunFilter(
    state=FlowRunFilterStateType(any_=[StateType.RUNNING, StateType.SCHEDULED,StateType.PENDING])
)

async with prefect.get_client() as client:
    run_info = await client.read_flow_runs(flow_run_filter=running_filter)

print(len(run_info)) #I always get a value of 200 here (mostly a big backlog of completed runs) 
print(run_info[0]) #a random old completed run.
any insight?
Ooohkay, idk what it is about cracking a problem the minute I reach out for help, but I think I sorted it.
Copy code
running_tagged_filter = FlowRunFilter(
    state=FlowRunFilterState(
        type=FlowRunFilterStateType(any_=[StateType.RUNNING, StateType.SCHEDULED,StateType.PENDING])))
this did the trick. I wouldn't have known to have expected the the triple-nested type declaration, except for that the double-nested declaration was driving me in circles!
5 Views
Open in Slack
PreviousNext
Powered by