Hieu Tran
10/11/2022, 9:53 AMimport asyncio
from prefect.client import get_client
from prefect.orion.schemas.states import StateType
from prefect.orion.schemas.filters import FlowRunFilter, FlowRunFilterState, FlowRunFilterStateType
async def main():
client = get_client()
state_type_filter = FlowRunFilterStateType(any_=[StateType.RUNNING])
state_filter = FlowRunFilterState(state_type_filter)
flow_filter = FlowRunFilter(state_filter)
flow_runs = await client.read_flow_runs(flow_filter)
print(flow_runs)
if __name__ == '__main__':
asyncio.run(main())
Traceback (most recent call last):
File "get_client_context.py", line 21, in <module>
asyncio.run(main())
File "/usr/lib/python3.8/asyncio/runners.py", line 44, in run
return loop.run_until_complete(main)
File "/usr/lib/python3.8/asyncio/base_events.py", line 616, in run_until_complete
return future.result()
File "get_client_context.py", line 9, in main
state_filter = FlowRunFilterState(state_type_filter)
File "pydantic/main.py", line 333, in pydantic.main.BaseModel.__init__
TypeError: __init__() takes exactly 1 positional argument (2 given)
state_filter = FlowRunFilterState(type=state_type_filter)
flow_filter = FlowRunFilter(state=state_filter)
flow_runs = await client.read_flow_runs(flow_run_filter=flow_filter)