Okay, follow-up question. Is there any way to create a subflow by flow name, id, deployment_id or something like that? I've tried using create_flow_run_from_deployment but the flow it creates is not a subflow, so it's not suitable for a flow-of-flows scenario. I'm trying to avoid using direct flow-object references since my flows will be deployed individually, using separate docker containers etc. If it's not possible, that's alright, I'll just have to rethink my strategy in that case. I know I've asked a similar question before but the response to that was "use create_flow_run_from_deployment", which I've finally had time to attempt and it's not what I need. 🙂

Hi all, the Prefect 2.0 Kubernetes agent auto-generated manifest appears to be a lot smaller than the 1.0 manifest. There are far fewer command-line switches and envvar options (just

PREFECT_API_URL

?) Are there additional agent settings in 2.0 I should be aware of for running a production k8s deployment, like the work queue name? If so, do they have envvars, and where in the docs can I read about them?

Hi, Do you have example of create_flow_run_from_deployment, wait for completion of flow run before continue to next line of code execution in 2.0? I was using create_flow_run and wait_for_flow_run in 1.0

How can I query "Service Accounts" from GraphQL? Just want to query which ones are already created or not

hey Prefect. I am looking to build a prefect v2 client with golang. This would then be used to build a terraform provider or kubernetes controller to create resources using infra as code. Maybe I just missed this in the docs, but how am I able to get a token to authenticate to prefect cloud v2

PrefectV2 appears to prefer throwing an exception to abort a flow early based on a condition (https://discourse.prefect.io/t/how-can-i-stop-a-flow-run-execution-based-on-a-condition/105). However, this results in failed run. Is there any way to end a flow early conditionally without a failed state. For example, we would like to run a daily dbt rollup when our source table has 24 hours of data, if not we should skip the dbt task. In this case, skipping the task is expected and show not results in a failed state.

Also anything more informative on GraphQL about why I cannot delete a specific user, would be nice 🙏 What do you think is the problem of the following query?

Hello all, Is there a way to set the path for the

manifest.json

file that is created from the

deployment build

cli command. I see you can set the

--output

for the

deployment.yaml

file. I have a number of flows in a single repo and would like to keep the structure clean, I know I can run the command from within the directory that I have the flow in but am looking to run the commands from the project root. This is also so I can simplify CI/CD jobs in the future. Any help would be appreciated.

I'm running latest prefect version 2.0.3 and getting the error below. Although the error isn't a show stopper since it didn't stop running after this point. Also I noticed the 'update-record' flow failed with validation error but the UI log doesn't show anything; only the terminal running the agent shows the error. I've seen this "Orion logging error" in version 2.0.2 and I think 2.0.1.

00:35:13.071 | DEBUG   | Flow run 'logical-lemming' - Resolving inputs to 'update-record'
00:35:14.896 | DEBUG   | prefect.client - Connecting to API at <http://127.0.0.1:4200/api/>
00:35:17.563 | DEBUG   | prefect.agent - Checking for flow runs...
00:35:22.591 | DEBUG   | prefect.agent - Checking for flow runs...
--- Orion logging error ---
The log worker encountered a fatal error.
Traceback (most recent call last):
  File "/Applications/anaconda3/envs/datapipeline/lib/python3.9/site-packages/prefect/logging/handlers.py", line 82, in _send_logs_loop
    anyio.run(self.send_logs)
  File "/Applications/anaconda3/envs/datapipeline/lib/python3.9/site-packages/anyio/_core/_eventloop.py", line 70, in run
    return asynclib.run(func, *args, **backend_options)
  File "/Applications/anaconda3/envs/datapipeline/lib/python3.9/site-packages/anyio/_backends/_asyncio.py", line 292, in run
    return native_run(wrapper(), debug=debug)
  File "/Applications/anaconda3/envs/datapipeline/lib/python3.9/asyncio/runners.py", line 47, in run
    _cancel_all_tasks(loop)
  File "/Applications/anaconda3/envs/datapipeline/lib/python3.9/asyncio/runners.py", line 56, in _cancel_all_tasks
    to_cancel = tasks.all_tasks(loop)
  File "/Applications/anaconda3/envs/datapipeline/lib/python3.9/asyncio/tasks.py", line 53, in all_tasks
    tasks = list(_all_tasks)
  File "/Applications/anaconda3/envs/datapipeline/lib/python3.9/_weakrefset.py", line 65, in __iter__
    for itemref in self.data:
RuntimeError: Set changed size during iteration
Worker information:
    Approximate queue length: 0
    Pending log batch length: 0
    Pending log batch size: 0
00:35:27.618 | DEBUG   | prefect.agent - Checking for flow runs...
00:35:32.655 | DEBUG   | prefect.agent - Checking for flow runs...
00:35:37.680 | DEBUG   | prefect.agent - Checking for flow runs...
00:35:42.713 | DEBUG   | prefect.agent - Checking for flow runs...
00:35:46.427 | DEBUG   | prefect.flows - Parameter 'target_collection' for flow 'update-record' is of unserializable type 'Collection' and will not be stored in the backend.
00:35:46.427 | DEBUG   | prefect.flows - Parameter 'sys_collection' for flow 'update-record' is of unserializable type 'Collection' and will not be stored in the backend.
00:35:46.427 | DEBUG   | prefect.flows - Parameter 'metadata_collection' for flow 'update-record' is of unserializable type 'Collection' and will not be stored in the backend.
00:35:47.751 | DEBUG   | prefect.agent - Checking for flow runs...
00:35:48.177 | INFO    | Flow run 'logical-lemming' - Created subflow run 'knowing-avocet' for flow 'update-record'
00:35:48.395 | ERROR   | Flow run 'knowing-avocet' - Received invalid parameters
Traceback (most recent call last):
  File "/Applications/anaconda3/envs/datapipeline/lib/python3.9/site-packages/prefect/engine.py", line 433, in create_and_begin_subflow_run
    parameters = flow.validate_parameters(parameters)
  File "/Applications/anaconda3/envs/datapipeline/lib/python3.9/site-packages/prefect/flows.py", line 275, in validate_parameters
    raise validation_err
prefect.exceptions.ParameterTypeError: 1 validation error for UpdateRecord
end_date
  str type expected (type=type_error.str)
00:35:48.435 | INFO    | Flow run 'logical-lemming' - Created task run 'update_record_in_db-a12634ec-0' for task 'update_record_in_db'
00:35:48.435 | INFO    | Flow run 'logical-

Hi. Do you have any idea why I receive such warnings on the k8s agent?

WARNING - agent | Job 'prefect-job-623be8f6' is for flow run '958c388f-f155-4ee8-a279-40c546b99808' which does not exist. It will be ignored.

Hey everyone, I have a little question regarding the block storage (and specifically the azure block storage). My understanding is we use the block as primary unit for prefect. There is two main use for them : 1. Store persistent logs 2. Store the flow I undertsand that with this python code I can define (and create the block) where I want to store the flow :

block = Azure(azure_storage_connection_string="paste_the_string_here")
block.save("dev")

and with that commande line I finalize the storage information for the flow :

prefect deployment build flow.py:flowname \
 --name deploy_name --tag dev -sb azure/dev

Can I do the same things from the UI ? In this case, I just need to use the command line and reference the block define in the UI. Am I right with all this assumptions ? And so, how I specify the storage for the persistent logs ? Thank you ! 🙂

Hi! I'm having a problem with that's been impacting our system in production for months now. In an apparently random fashion, from once a month to once every few days, a request from a flow runner to the server API times out, and that's the start of a major sh*t storm which usually ends up in having to manually restart the machine. (Details in the next messages...)

Hi! I'm facing error with using the for loop condition while using it in 'with Flow("flow_name") as flow' syntax and facing error as TypeError: object of type 'GetItem' has no len() How can I use the for loop inside the flow (Prefect v1.2.2). Any suggestions?

Hi Prefect Community, i am facing the following issue:

(begin_task_run pid=81126)     from prefect.packaging.docker import DockerPackager

(begin_task_run pid=81126) ImportError: cannot import name 'DockerPackager' from partially initialized module 'prefect.packaging.docker' (most likely due to a circular import) (/home/christian/Documents/ray_and_prefect/temp/temp-venv/lib/python3.9/site-packages/prefect/packaging/docker.py)

Hi all I’m attempting to implement a flow where a task only runs if any of the direct upstreams fail so I’m using

trigger=any_failed

which works fine when one of those does fail. But when those upstream tasks are all successful, I seem to get a

TriggerFailed

on that task and therefore the flow run is

FAILED

. Is that expected behaviour and if so, how do I instead get it to just skip that task instead of marking it as Failed and therefore the whole flow as failed? Thanks in advance

Hi! Now that 2.0 is GA, is there a Prefect 2.0 published roadmap somewhere of future plans for next few months (or beyond)?

Hi Prefect team Regarding Schedule, is it possible to build the yaml file with a

--schedule

option (like we do for storage block for instance) ? In my automated process I can’t edit the deployment.yaml file as I re-build this file every time in case of something has changed

Hi, I am fairly new to Prefect so any help (resources / tips&tricks etc) is appreciated. I want to setup a Prefect Server and a Local Agent. I got a

docker-compose.yaml

from running

prefect server config

(posted below). I should now add a new service at the end that runs a Local Agent and a service that registers my flow at startup - is that correct? For now if I add a service that runs

prefect agent local start

I get

ValueError: You have not set an API key for authentication.

when running

docker-compose up

.

What's the default value of "heartbeat_mode" ? process / thread or off? https://docs-v1.prefect.io/orchestration/concepts/services.html#heartbeat-configuration

Hi, am migrating my pipelines from v1 to v2. Each of the tasks returns large amount of data, so movement of data from task to task is quite slow. I noticed that the results are being stored in the local storage n if I am right thats why its so slow, i removed that setting in v1 (dont really remember how) n was wondering if such an option is available in v2

pickle issue in prefect 2

The migration guide doesn’t mention this, but did GitHub flow storage in Prefect 1.0 go away in Prefect 2.0 in favour of the generic cloud block storage providers?

👋 I am having trouble logging into Prefect cloud with the CLI. When I login in I get:

✗ poetry run prefect cloud login -k my-key
Traceback (most recent call last):
  File "/Users/username/Library/Caches/pypoetry/virtualenvs/prefect-eval-ZjHwTZGp-py3.9/lib/python3.9/site-packages/prefect/cli/_utilities.py", line 41, in wrapper
    return fn(*args, **kwargs)
  File "/Users/username/Library/Caches/pypoetry/virtualenvs/prefect-eval-ZjHwTZGp-py3.9/lib/python3.9/site-packages/prefect/utilities/asyncutils.py", line 193, in wrapper
    return run_async_in_new_loop(async_fn, *args, **kwargs)
  File "/Users/username/Library/Caches/pypoetry/virtualenvs/prefect-eval-ZjHwTZGp-py3.9/lib/python3.9/site-packages/prefect/utilities/asyncutils.py", line 140, in run_async_in_new_loop
    return anyio.run(partial(__fn, *args, **kwargs))
  File "/Users/username/Library/Caches/pypoetry/virtualenvs/prefect-eval-ZjHwTZGp-py3.9/lib/python3.9/site-packages/anyio/_core/_eventloop.py", line 70, in run
    return asynclib.run(func, *args, **backend_options)
  File "/Users/username/Library/Caches/pypoetry/virtualenvs/prefect-eval-ZjHwTZGp-py3.9/lib/python3.9/site-packages/anyio/_backends/_asyncio.py", line 292, in run
    return native_run(wrapper(), debug=debug)
  File "/Users/username/.pyenv/versions/3.9.13/lib/python3.9/asyncio/runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "/Users/username/.pyenv/versions/3.9.13/lib/python3.9/asyncio/base_events.py", line 647, in run_until_complete
    return future.result()
  File "/Users/username/Library/Caches/pypoetry/virtualenvs/prefect-eval-ZjHwTZGp-py3.9/lib/python3.9/site-packages/anyio/_backends/_asyncio.py", line 287, in wrapper
    return await func(*args)
  File "/Users/username/Library/Caches/pypoetry/virtualenvs/prefect-eval-ZjHwTZGp-py3.9/lib/python3.9/site-packages/prefect/cli/cloud.py", line 264, in login
    current_workspace = get_current_workspace(workspaces)
  File "/Users/username/Library/Caches/pypoetry/virtualenvs/prefect-eval-ZjHwTZGp-py3.9/lib/python3.9/site-packages/prefect/cli/cloud.py", line 64, in get_current_workspace
    current_workspace_id = re.match(
AttributeError: 'NoneType' object has no attribute 'groups'
An exception occurred.

I also get an error if I manually set the key via:

prefect config set PREFECT_API_URL="<https://api.prefect.cloud/api/accounts/[our-account-id]/workspace/[our-workspace-id]>"
prefect config set PREFECT_API_KEY="mykey"

And then use

prefect profile use my-profile

to login. How do I get my CLI setup to work against our cloud workspace?

Hey all, we’re hosting @Christopher Boyd on PrefectLive today at 3p Eastern where he’s going to show how to set up Microsoft Azure with Prefect on Twitch. Come join us if this is the topic you are interested in

Hello prefect community, I wonder what is the best way to share data within flow (prefect 1.x). What I’m trying to achieve is running a bunch of tasks using map (bit of a fire and forget approach - it just generates data someplace). These tasks run computations and report status. In certain situations it breaks and log would be produced. When there are many of these tasks, scrolling through the log is a bit cumbersome. My idea would be to use artifacts API and have a single page with all the errors summarized. I’d build that page incrementally by updating a single artifact with the info of what has failed, much like it is described here: https://docs-v1.prefect.io/orchestration/concepts/artifacts.html#creating-markdown-artifacts, but I’m not too sure how I could sync the content of the error list and the initial markdown_id across all of the mapped tasks and perhaps even with other tasks other than creating hell of passed parameters and extra tasks that would update the artifact page. My idea was something along the lines:

tag = prefect.context.get("artifacts_tag_id")
        msgs: list = prefect.context.get("artifacts_content") or []
        msgs.append(str(err))
        content = create_err_artifact_from_list_of_errs(msgs)

        if tag:
            update_markdown_artifact(tag, content)
        else:
            tag = create_markdown_artifact(content)
            prefect.context["artifacts_tag_id"] = tag

        prefect.context["artifacts_content"] = msgs

but I found out that mutating the context from within a task is not recommended. So I wonder if there is any other prefect way of sharing these two ‘variables’ throughout the life of a prefect flow. Thank you very much!

Hi, I’m using Prefect on ECS and have two separate clusters for test and prod. As suggested in one of the other threads here, I’m trying to use environment variables to specify the cluster name: However, when I run the flow in ECS it tells me the cluster is of type ‘None’ Here’s the run config I’m trying to use: Should using os.getenv work here?

Hello Guys! I got a quick question. Is there a way to track Prefect's CPU and Memory Utilization or is it better done through a third party tool, If so, do you have any suggestion on a good way to do it? Thanks a lot!

Hey peeps, have you guys been able to deploy to an AWS S3 block that requires MFA?

Hi, How can I programmatically retrieve deployment id from deployment name. Eg, retrieve id 'b86c7264-beae-4485-a68c-290286e92f00' by providing 'LiveLots-ETL/dev', is there Python API to achieve this?

Hey all, we’re hosting @Christopher Boyd on PrefectLive today at 3p EST (starting in 15') where he’s going to show how to set up Microsoft Azure with Prefect on Twitch. Come join us if this is the topic you are interested in