Hi All I m getting an error when using `with options` for an Prefect Community #ask-community

Hi All, I’m getting an error when using `with_opt...

Avinash Santhanagopalan

02/01/2024, 4:59 PM

Hi All, I’m getting an error when using

with_options

for an async flow. It says the below error. Has anyone else faced the same issue? Any workarounds?

Copy code

AttributeError: 'coroutine' object has no attribute 'with_options'

Nate

02/01/2024, 5:52 PM

hi @Avinash Santhanagopalan - can you show your code? it sounds like you're not awaiting something that is supposed to return a flow

Avinash Santhanagopalan

02/01/2024, 6:05 PM

I can’t share the exact code but it would be something like

Copy code

jobs = [
        refresh_data_for_plugin(
            plugin_name,
            environment,
            *additional_args,
            **batch_kwargs,
        ).with_options(name=f"refresh-{plugin_name}-{environment}")
        for plugin_name in plugins
    ]

asyncio.gather(*jobs)

That

refresh_data_for_plugin

is the flow I’m trying to rename using

with_options

Copy code

@flow
async def refresh_data_for_plugin(
    plugin_name: str,
    environment: str,
    *additional_args,
    **batch_kwargs,
) -> None:

Nate

02/01/2024, 6:09 PM

hm so you should be calling with_options on the flow object

refresh_data_for_plugin

instead of the coro you get when you call it so as is you'd want

Copy code

jobs = [
        refresh_data_for_plugin.with_options(name=f"refresh-{plugin_name}-{environment}")(
            plugin_name,
            environment,
            *additional_args,
            **batch_kwargs,
        )
        for plugin_name in plugins
    ]

asyncio.gather(*jobs)

but what I would do instead is this

Copy code

@flow(name="refresh-{plugin_name}-{environment}")
async def refresh_data_for_plugin(
    plugin_name: str,
    environment: str,
    *additional_args,
    **batch_kwargs,
) -> None:

and then

Copy code

jobs = [
        refresh_data_for_plugin(
            plugin_name,
            environment,
            *additional_args,
            **batch_kwargs,
        )
        for plugin_name in plugins
    ]

asyncio.gather(*jobs)

thank you 1

Avinash Santhanagopalan

02/01/2024, 6:10 PM

Ah I see. I will try that out!

👍 1

Avinash Santhanagopalan

02/02/2024, 3:37 PM

Hey @Nate, I tested this out but the values for

{plugin_name}

and

{environment}

do not get applied for the flow if used in a decorator. f-strings also don’t work. I guess that substitution only works for

flow_run_name

. But

refresh_data_for_plugin.with_options(name=f"refresh-{plugin_name}-{environment}")

did work so thanks for that!

12 Views

Open in Slack

Previous Next