is there any known issue with

create_flow_run

that will cause two flows to spawn from the one call?

hi folks, i have a question about agent-specific flow config. is there a way to make config available in a flow where the value will depend on the agent it's running in?

Can a templated task run name access attributes of objects? Like task_run_name="{my_object.name}"

This message was deleted.

This message was deleted.

The error is simply sqlalchemy.exec.OperationalError: (sqlite3.OperationalError) table flow already exists

Prefect Cloud 1.0 gives me the error message `Something went wrong. Please wait a few moments and try again.`when I try to enable a schedule. My flow was registered with a simple IntervalSchedule. How would I debug that?

This message was deleted.

Hey, for prefect 2.0 can I use *def my_flow(*kwargs) to get my flow parameters? Or do I need to specify each parameter individually as named parameter in my flow function?

Hello, i was wondering if there was a way to retrieve current flow run metadata such as id or time elapsed inside the code ?

Hi Folks! I am facing an issue in running a flow on schedule after registering it on prefect cloud. I am getting the following error in logs in prefect cloud

Failed to load and execute flow run: FlowStorageError('An error occurred while unpickling the flow:\n  ModuleNotFoundError("No module named \'google.api\'")\nThis may be due to a missing Python module in your current environment. Please ensure you have all required flow dependencies installed.')

I tried running the flow by putting the

flow.run

instead of

flow.register

and then ran the flow using

python3 workflow.py

. And its running fine. The agent is installed on the same machine from where I ran the flow manually. So registering and running are done from the same machine. So I cannot understand why such error is thrown. Can anybody help?

Has anybody come across an error when using

prefect cloud login --key {key} --workspace {workspace}

related to timeouts? I'm trying to run the command in a CI/CD pipeline through Google Cloud Build but it fails due to a timeout and I have very little information in the error.

Is it possible to set max_retries on Tasks from a Parameter, i.e. runtime?

Hi all loving the new look for prefect 2! I had a question about the storage options for prefect 2, it is possible for me to configure storage for flows and flow results separately? For instance if i wanted to store flow results in cloud storage and flows on local storage can i do that currently? Also curious about the roadmap for storage options. In prefect 1 you allowed for github and module storage just curious if that is coming back in 2?

Curious if there is a best practice on something I'm struggling with. I have a task that creates a sql alchemy engine from a parameter, and this is passed on to the next set of tasks. However, you can't pickle a sql alchemy engine, which means I have checkpointing turned off for the Get Engine task. This means that if one of the queries fails and retries it will not have the input it needs. Do I need to just create the engine inside the task? Or is there another way people have used?

Is there a plan inbetween starter and enterprise to have more than 3 users?

Hi all, we have named volumes to which the agents are mounted to in prefect. We are facing an issue whenever a Flow Fails/Task fails involving a docker mount that the full server error message is returned with the full server values including themount location, user, address, and password. Obviously this is a massive security issue. Is there anyway to circumvent this output? SCRIPT ALERT: Intraday Strategy Greeks entered state Failed with message 500 Server Error for `http+docker://localhost/v1.41/containers/416e8db53c290dc44f8b1621c027e1930f07606101c82ed0ed059a367bcd8c13/start`: Internal Server Error ("error while mounting volume '/var/lib/docker/volumes/output/_data': failed to mount local volume: mount //XXXXXX.XXX.XXX/XXX:/XXX/lib/docker/volumes/output/_data, data: addr=XX.XXX.X.XX, username=XXXX, password=XXXXXX: no such file or directory"). (edited)

Hi I have the same issue as in this thread with loosing value of “task_loop_result” on retries https://prefect-community.slack.com/archives/CL09KU1K7/p1650958214772249 More code inside

how do I add an additional slack notification action to my Automation Actions. I added a new webhook today but idk how to get it to show up in the UI. Do I need to reinstall the slack app?

Hello. Very new to Prefect, and thinking about how to structure my project(s). Is there any way to create a master flow composed of flows that exist across different git repositories?

Hello, I am new at Prefect, trying to run flow with dbt execution and in part of initializing GitHub Storage I got error in Cloud (I renamed secret here in error):

Failed to load and execute flow run: KeyError('The secret ghp_pZdHbjkfhkjkefefiejfefnPd1VJikfn was not found.  Please ensure that it was set correctly in your tenant: <https://docs.prefect.io/orchestration/concepts/secrets.html>')

I suppose issue in access_token_secret, I stored it in Cloud Secrets

STORAGE = GitHub(
    repo="",
    path=f"flows/{FLOW_NAME}.py",
    access_token_secret=Secret('GITHUB_ACCESS_TOKEN').get()
)

I checked, github token is valid. Any ideas about this error? Thank you

quick question on the Prefect-maintained Orion collections — are there plans to port the 1.0 Jupyter task to Orion?

Hey, quick question, prefect 2.0: I can't figure out how to set time to live or successful/failed job history limits for k8s jobs which were spawned from a kubernetes orion agent. Everything works fine, but I've got hundreds of completed/failed jobs (and corresponding pods) lying around. Is there a way to configure job history limits, or configure automatic cleanup? (Now that I think about it, I could also use a prefect flow to clean up, but maybe that's too much 😄 )

Prefect 2.0 beginner here- Is there a way to retry a subflow in prefect 2.0 upon failure? Or are retries only reserved for tasks?

Hi, its possible naming a schedule of a flow in cloud? I run many schedules of the flow with different parameters but I would like named it to difference each other.

Hello, we’ve been running Prefect for some our jobs for a while now (3 months) and usually runs go through successfully, in our PROD environment every job has been getting this error since yesterday: “No heartbeat detected from the remote task; marking the run as failed.” Any thoughts on how I should go about debugging this?

Hi, trying to run DbtShellTask using command "dbt run" but get such kind of error:

[2022-06-11 08:35:49+0300] ERROR - prefect.CloudTaskRunner | Task 'DbtShellTask': Exception encountered during task execution!
Traceback (most recent call last):
  File "C:\Users\User\PycharmProjects\demo\venv\lib\site-packages\prefect\engine\task_runner.py", line 880, in get_task_run_state
    value = prefect.utilities.executors.run_task_with_timeout(
  File "C:\Users\User\PycharmProjects\demo\venv\lib\site-packages\prefect\utilities\executors.py", line 468, in run_task_with_timeout
    return task.run(*args, **kwargs)  # type: ignore
  File "C:\Users\User\PycharmProjects\demo\venv\lib\site-packages\prefect\utilities\tasks.py", line 456, in method
    return run_method(self, *args, **kwargs)
  File "C:\Users\User\PycharmProjects\demo\venv\lib\site-packages\prefect\tasks\dbt\dbt.py", line 192, in run
    return super(DbtShellTask, self).run(
  File "C:\Users\User\PycharmProjects\demo\venv\lib\site-packages\prefect\utilities\tasks.py", line 456, in method
    return run_method(self, *args, **kwargs)
  File "C:\Users\User\PycharmProjects\demo\venv\lib\site-packages\prefect\tasks\shell.py", line 145, in run
    line = raw_line.decode("utf-8").rstrip()
UnicodeDecodeError: 'utf-8' codec can't decode byte 0xad in position 7: invalid start byte

Any ideas what is the issue?

Hello, i'm stuck in the following issue setting up Prefect + Dask cluster. When the lfow is executed multiple tasks are run on the Dask Worker. When it run i get an exception (see thread)

Hello, I have an async function, which fetches some data from an REST API (using aiohttp). The task returns a pandas DataFrame. A second function saves the pandas DataFrame as a parquet file into a s3 bucket. Running this functions in python is really fast (<1s). However, if I wrap both functions into prefect tasks and place them into a prefect flow, the flow execution takes up to 5 minutes. I do not understand, why the pure python solution os so much fast, compared to when using prefect. I have posted some sample code inside of the thread.

hello -- i have installed prefect-2 but I got this error

Traceback (most recent call last):
  File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/prefect/cli/base.py", line 59, in wrapper
    return fn(*args, **kwargs)
  File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/prefect/utilities/asyncio.py", line 120, in wrapper
    return run_async_in_new_loop(async_fn, *args, **kwargs)
  File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/prefect/utilities/asyncio.py", line 67, in run_async_in_new_loop
    return anyio.run(partial(__fn, *args, **kwargs))
  File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/anyio/_core/_eventloop.py", line 56, in run
    return asynclib.run(func, *args, **backend_options)
  File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/anyio/_backends/_asyncio.py", line 233, in run
    return native_run(wrapper(), debug=debug)
  File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/asyncio/runners.py", line 43, in run
    return loop.run_until_complete(main)
  File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/asyncio/base_events.py", line 584, in run_until_complete
    return future.result()
  File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/anyio/_backends/_asyncio.py", line 228, in wrapper
    return await func(*args)
  File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/prefect/cli/base.py", line 162, in version
    if is_ephemeral:
UnboundLocalError: local variable 'is_ephemeral' referenced before assignment
An exception occurred.