Hi there. I am working on a flow that needs to do the following: • Start on a given day of the month (for example, the 6th) • Call an API to see if the data is ready. If the data is not ready, it should continue to check every few hours. If the data is not ready by the 15th of the month, it should fail. • Once the data is ready, it should do additional work I am thinking that the prefect 2 pattern would be: • Have a monthly deployment with a schedule that starts this flow on the 6th of the month. • The flow needs to have retries specified (a sufficiently large number) • Have a

check_data_availability

task that calls the API. If the data is not ready, it should raise an exception. • If the data is ready, it will continue Can you think of a better pattern? If I go with this approach, I was thinking that it would be good to define a new state

waiting for data availability

or something along those lines to differentiate it from a failure. Does this make sense? Are custom states possible? Can Prefect Cloud render custom states? How would I specify a custom state? Thanks!

Hello community. I would like to have my agent continue to run flows up to the queue concurrency limit. Will a typical agent run flows asynchronously, regardless of whether the flow deployment is a separate ECSTask, vs. a simple (non-ECSTask) deployment that runs one of the flow tasks as a separate ECSTask? I am mainly curious what are the differences from an agent-asynchronous-concurrency perspective between a deployment as ECSTask vs a portion of the flow as ECSTask. Thanks!

Hi all! I am working to transition to Prefect 2.x and am looking for guidance on deployment. Formerly, I used a python script to deploy flows with the following general pattern:

from prefect.storage import Docker
from prefect.run_configs import DockerRun
import os
from prefect.utilities.storage import extract_flow_from_file

# locate all flows and extract from files
flow_directory = "..."
flow_file_names = [...]
all_flows = [
    extract_flow_from_file(os.path.join(flow_directory, file_name))
    for file_name in flow_file_names
]

# define flow storage/config
STORAGE = Docker(...)
RUN_CONFIG = DockerRun(...)


def build_and_register_all_flows(
    flow_list,
    storage=STORAGE,
    run_config=RUN_CONFIG,
):
    # assign run config and storage to all flows
    for flow in flow_list:
        flow.run_config = run_config
        storage.add_flow(flow)
    # build the Docker Image
    storage = storage.build()

    for flow in flow_list:
        # Reassign the new storage object to each Flow
        flow.storage = storage
        # Register each flow without building a second time
        flow.register(...)


if __name__ == "__main__":
    build_and_register_all_flows(all_flows)

This pattern relied heavily on the

extract_flow_from_file

utility from prefect 1.x which I am not sure exists in prefect 2.x I have found this deployment tutorial on the prefect site that shows a reasonably familiar pattern in the “Deployment creation with Python” section that I hope to follow. However, it only applies to one flow. Can someone help me understand how to apply this to multiple flows? is there a utility similar to the

extract_flow_from_file

utility from prefect 1.x that can help me to bundle up a bunch of flows and deploy them together?

fyi, there is a broken link on https://www.prefect.io/opensource/. The learn more button routes you to a 404 v1 doc page. https://docs-v1.prefect.io/core/concepts/failures.html

I want to write a flow which executes some tasks and then waits for user to do some manual QA before proceeding to run the remaining tasks. I was wondering if there’s a way to do this in prefect? Basically I want a Task that will wait for a user’s input in order to succeed.

Is there a simple way to register every single flow in a directory? The equivalent of

prefect register <filepath>

in Prefect 1? Assume it's a bunch of

py

files with Deployments defined in Python code.

Running

prefect server start

results in server starting then immediately stopping after prefect reinstall in new conda environment

We are using WaitOnContainer from prefect.tasks.docker in Prefect 1, to wait for container until it finish its work. In Prefect 2 we use prefect_docker.containers but, it doesn't have function to wait for container to finish its work. Is this functionality not implemented in Prefect 2.0? Please help!!!!

Hi Community, I have a task that is making over 100 http api requests in the loop. What is the prefect way to implement such task?

Hey - we are trying to migrate to github storage. I have set up the storage for a private repo and all looks good, but when the flow starts to run and the

storage.get_directory()

is triggered I get the following error in 🧵

Hi, I think there is an issue with editing Variables. when a new Variable is created, the value is limited to 5000 chars. but when editing an existing value, the limitation is set to 256 chars

Hi, Is there anything I can do to facilitate the review of https://github.com/PrefectHQ/prefect/pull/9567?

Hello! I would like to ask, is it possible to modify the WorkQueueHealthPolicy of specific queues in Prefect, or change the global default values? Is there a setting for it? This is helpful when calling the api/work_queues/{id}/status API call

Hi All, I have a question regarding accessing Prefect server UI, with Prefect hosted on AWS EC2 Ubuntu 22.04 in private subnet. I have trouble accessing it directly via private IP of instance, while connected to corporate network/VPN, which allows communucating with AWS VPC using private IPs. If anyone has similar experience, I'd appreciate help. Thank you! 🙂 Let me explain the issue in details: Assumptions: • Prefect is hosted on EC2 in private subnet of VPC. It is hosted directly within conda environment, and uses Postgres DB backend, running on Docker container on same instance. This all works fine • I was able to access Prefect UI using a proxy EC2 instance with public IP, which is allowed to send HTTP requests to private IP of the instance Prefect is hosted on. I used following configuration, and everything works great: ◦ prefect config set PREFECT_API_URL='<http//14200/api|http//&lt;private ip of instance&gt;4200/api>' ◦ prefect config set PREFECT_SERVER_API_HOST='0.0.0.0' ◦ prefect config set PREFECT_API_DATABASE_CONNECTION_URL="<postgresql+asyncpg//pyourTopSecretPassword@localhost5432/prefect|postgresql+asyncpg//<user>:<pw>@localhost:5432/prefect>" • However, when trying to access '<http//14200/api|http//&lt;private ip of instance&gt;4200/api>' directly (without proxy) from corporate network/VPN with the same settings as above, Prefect UI complains it can't communicate with API • If I delete PREFECT_API_URL and PREFECT_SERVER_API_HOST from config, and run 'prefect server start --host 0.0.0.0', I'm able to access the UI only if I expose a port in VS Code, and click on attached 'Open in Browser' option. Again everything works fine • However, I want to be able to access UI directly on '<http//14200/api|http//&lt;private ip of instance&gt;4200/api>' from corporate network/VPN. When I try to do that, I only get a blank page titled 'Prefect Server', like in attached image • Using cURL on <http//14200/api|http//&lt;private ip of instance&gt;4200> returns HTML response from Prefect • Nothing in network settings, security groups etc changed meanwhile If someone has experience with problems like this, I would really appreciate help. Thank you! 🙂

Hi everyone, I have a question regarding crashed tasks and task retries on Prefect 2.10.10. I noticed that task retry only works for tasks that go in FAILED state, whereas it doesn't for tasks in CRASHED, I was just wondering why is that and if I have any option to make it retry in case of a crash. Thanks!

Just something silly but I always type

-h

instead of

--help

because every other CLI allows this as well. It would be nice if

prefect

would also just understand

-h

Hi everyone, i have created prefect v2 flows and it was running fine. Then, i have created a new prefect account and tried to run the same flows with zero difference, But the flows are not running in that account. I again switched to the old account and tried running the flows which worked fine. Do anyone faced this similar issue or anyone have any ideas to sort this out?

Do I need a separate deployment for the same file/flow using different values for a given parameter? I want to schedule a flow run using Parameter 1 = A and flow run using Parameter 1 = B.

Hey peeps, I’m having some trouble wrapping my head around the new Projects and Workers. I can see how it would be useful with my current deployment with GitHub Storage and Docker infrastructure blocks. I can be debugging an issue and if I forget to push the latest commit or latest Docker image I run into the same errors. When I run `prefect deployment build`… I’d love for it to build and push the Docker image, sync the script to GitHub, and apply the deployment in Prefect. Is there a sample tutorial for something like this available?

Hello team! I recently migrated from prefect 1 to 2

i have a questions, everythings its ok execp with an single script with troubles, i need a hand 👋

i have the script :

from prefect import flow from prefect_shell import shell_run_command @flow(name_=_"Run Meltano Google") def run_meltano_google(): command = "meltano --log-level=debug elt tap-google-ads target-postgres--google --state-id=google-ads-to-postgres" return shell_run_command(command_=_command, helper_command_=_"cd ~/agendapro-data-integration/meltano/;source .venv/bin/activate;", return_all_=_True) if name == "__main__": run_meltano_google().run()

on prefect 1 it runs ok, but when a migrated to prefect 2 the flow keep run infinite, never end, its only a shell command but i dont know why it stuck

if i run the comment inside cd ~/agendapro-data-integration/meltano/;source .venv/bin/activate;" and then run it manually and it works but from prefect2 orquestator dont

Hi Prefect team, We recently reconfigured our deployments to run multiple times from a master flow which uses an ETL template. We use AWS infrastructure and found that this reconfiguration works fine on Fargate instances but mysteriously crashes on EC2 instances. These flows were previously working just fine on EC2 (and Fargate) when instantiated as one-deployment-per-flow.

21:53:15.511 | INFO    | prefect.agent - Submitting flow run '7bec67b1-ddba-429a-a3a6-dc071401433b'
21:53:17.265 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'pretty-mussel': Retrieving task definition 'arn:aws:ecs:us-east-1:421396523132:task-definition/zarr-etl-ec2-efs:44'...
21:53:17.517 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'pretty-mussel': Retrieving task definition 'zarr-etl-ec2-efs'...
21:53:17.540 | WARNING | prefect.infrastructure.ecs-task - ECSTask 'pretty-mussel': Settings require changes to the linked task definition. A new task definition will be registered.
21:53:17.603 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'pretty-mussel': Creating task run...
21:53:17.726 | ERROR   | prefect.agent - Failed to submit flow run '7bec67b1-ddba-429a-a3a6-dc071401433b' to infrastructure.
Traceback (most recent call last):
  File "/root/.local/lib/python3.10/site-packages/prefect/agent.py", line 490, in _submit_run_and_capture_errors
    result = await infrastructure.run(task_status=task_status)
  File "/root/.local/lib/python3.10/site-packages/prefect_aws/ecs.py", line 621, in run
    ) = await run_sync_in_worker_thread(
  File "/root/.local/lib/python3.10/site-packages/prefect/utilities/asyncutils.py", line 91, in run_sync_in_worker_thread
    return await anyio.to_thread.run_sync(
  File "/root/.local/lib/python3.10/site-packages/anyio/to_thread.py", line 31, in run_sync
    return await get_asynclib().run_sync_in_worker_thread(
  File "/root/.local/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 937, in run_sync_in_worker_thread
    return await future
  File "/root/.local/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 867, in run
    result = context.run(func, *args)
  File "/root/.local/lib/python3.10/site-packages/prefect_aws/ecs.py", line 814, in _create_task_and_wait_for_start
    self._report_task_run_creation_failure(task_run, exc)
  File "/root/.local/lib/python3.10/site-packages/prefect_aws/ecs.py", line 810, in _create_task_and_wait_for_start
    task = self._run_task(ecs_client, task_run)
  File "/root/.local/lib/python3.10/site-packages/prefect_aws/ecs.py", line 1487, in _run_task
    return ecs_client.run_task(**task_run)["tasks"][0]
IndexError: list index out of range

We would like to avoid a one-deployment-per-flow setup if at all possible as it’s considerably clunkier code-wise and on the Cloud interface. We must run some of our ETLs on EC2 however. We encountered this error running on prefect

2.10.10

with flow runs using ECSTasks. Any ideas? Note that this appears a repeat of a previous, unanswered issue in this channel https://linen.prefect.io/t/10301091/hi-i-am-running-a-deployment-multiple-times-from-a-master-fl. @Tibs did you ever find a solution?

Is it okay to set only a single environment variable related to logging, or do I need to supply ALL the logging config? After setting this environment variable on the agent, no logs show up on the server when we kick off a flow run from the UI

SET "PREFECT_LOGGING_FORMATTERS_STANDARD_TASK_RUN_FMT=%(asctime)s.%(msecs)03d | %(levelname)-7s | %(task_name)s | %(message)s"

We've started to get frequent "429 Too many Requests" errors in the past few days causing flows to fail and our work queue to go into an unhealthy state. We have not changed the number of flows or tasks that are running. has something changed on the prefect cloud side? Have the api limits changed? Are there other api request limits besides the per-minute limits

Hi everyone! I wanted to highlight that due to this issue we cannot use Prefect >= 2.10.0

is it possible using the client read all blocks? I just need to retrieve all blocks name-id pairs