https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> can you explain to me how async works...
# ask-marvin
a
@Marvin can you explain to me how async works within a flow and tasks..specifically regarding example inside an async function why would we not use await for futures. task_future = send_audit_log_query.submit( _report_generator_=report_generator, client=client, _display_name_=display_name, _request_body_=request_body, ) await task_futures.append((idx, date_range, task_future)) # <- Instead the documentation is mentioning to not use await, but I think that's causing a block in code. 212642.672 | ERROR | asyncio - Critical error in *: Event loop is closed 212642.672 | ERROR | Flow run 'busy-mastodon' - Encountered exception during execution: RuntimeError('Event loop is closed') Traceback (most recent call last): File "/usr/local/lib/python3.13/site-packages/prefect/flow_engine.py", line 1299, in run_context yield self File "/usr/local/lib/python3.13/site-packages/prefect/flow_engine.py", line 1357, in run_flow_async await engine.call_flow_fn() File "/usr/local/lib/python3.13/site-packages/prefect/flow_engine.py", line 1316, in call_flow_fn result = await call_with_parameters(self.flow.fn, self.parameters) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/app/***/src/main.py", line 146, in ** async with AsyncClient() as client: ~~~~~~~~~~~^^
m
Your message was too long by 35 tokens...
In Prefect 3.x, when dealing with asynchronous flows and tasks, it's important to understand how futures are managed. Here's the gist: 1. Synchronous Nature of 
.submit
and `.map`: Even if the underlying user function is asynchronous, Prefect's 
.submit
and 
.map
methods are synchronous. They return futures that represent the eventual completion of the task, but they don't need to be awaited themselves. Instead, you handle them in a non-blocking way. 2. Handling Futures: The futures returned by 
.submit
or 
.map
can be resolved by: - Passing them to another task. - Returning them from a task or flow. - Manually calling 
.result()
or 
.wait()
on them. 3. Event Loop Issue: The error you're encountering ("Event loop is closed") might arise if there's a conflict in how the event loop is being managed. Prefect manages its own event loop, and mixing it with other async libraries, like `httpx`'s 
AsyncClient
, can sometimes cause issues. 4. Avoiding 
await
on Futures: You should not use 
await
on the futures returned by 
.submit
or 
.map
directly because they are not asyncio-compatible futures. Instead, use 
.result()
or 
.wait()
to handle them. For more detailed information, you can refer to the Prefect task documentation and flow engine documentation. If you have a specific setup or stack configuration, let me know, and I can provide more tailored advice.
10 Views
Open in Slack
PreviousNext
Powered by