Hi, I had a couple questions as my team is exploring V2. If there is a better place to ask these questions, let me know! 1) Is there a way to run tasks downstream from a specific point in a flow? For example, suppose a user has the following flow

a >> b >> c >> d

. After the flow has run, what is the best way to run tasks

b

to

d

? 2) On a related note, in the following thread, it seems that there was a PR for flow restarts. Has there been any follow-up there by any chance? Can you also clarify the difference between a flow retry and a flow restart? https://discourse.prefect.io/t/how-can-i-restart-a-flow-in-prefect-2-0/1332/9 3) After a flow run is canceled, what's the best way to resume the flow run? When we cancel a flow run, it seems that the flow run is indeed canceled. On the other hand, any tasks that were previously running are interrupted but left in the running state. So, when we attempt to retry the flow, those tasks that are still in the running state cannot be restarted. 4) What is the purpose behind deleting a task run? How should we use this feature? Thanks for your help!

Hey Y'all, I'm not sure if anyone else has noticed this, but even though we didn't change anything on the flow runs, I got

Client Error '404 Not Found' for the url

, but this happened only once before my flows resumed normal operation.

hey I was thinking shouldn’t automation triggers have

deployment run state

in addition to

flow run state

and

work queue health

? I have a flow

trigger_dbt

in multiple deployments and I only want to trigger for specific conditions (i.e. the deployment conditions.. specific params etc).

Hi 😄 I have a quick question is it a good practice to use the prefect api to launch flows ? If you liked answering this question feel free to answer this wonderful question it will help me a lot ! Thanks prefecters 💌

Hi there! Have more questions regarding the orion REST API, specifically the one for the cloud. We are trying to create flow runs from deployments using this cloud api endpoint https://app.prefect.cloud/api/docs#tag/Deployments/operation/create_flow_run_from_deployme[…]es__workspace_id__deployments__id__create_flow_run_post I can see the flow runs appearing in our cloud UI and kubernetes work queue. However, our kubernetes agent is not picking up these jobs from the queue. Any ideas?

Hi there! Does anybody know, what environment variables are needed to be switched on in Prefect OSS in order to use SSL/HTTPS with organization CERTS? If I understand, I should set HTTPS_PROXY and point mounted SSL_CERT_FILE and that's all?

Hi! We're in the midst of just migrating to self hosting prefect 2 (

2.7.8

) and I'm having some trouble with a 'permission denied' error when my pod (run via

KubernetesJob

) tries to run my deployment (using s3 storage block) The short version of the error is:

Flow could not be retrieved from deployment.
Traceback (most recent call last):
...
File "/opt/env/lib/python3.9/site-packages/s3fs/core.py", line 1141, in _get_file
    with open(lpath, "wb") as f0:
PermissionError: [Errno 13] Permission denied: '/workdir/./src/hvac/__init__.py'

I can get it to go away if I give that

rw

permissions for every user before it gets uploaded to s3, but it seems strange that I should have to do this, so I feel like I must be missing something... Any help is appreciated, but also I'm just looking to understand better what prefect is actually doing when it spins up my pod. I understand that it downloads code from the storage block (S3 in my case), but I'm fuzzy on a few details: • What determines where the code gets downloaded to on the pod? Is it the

.path

of the Deployment? • Why does prefect need to attempt to write to the file in my error? My Deployment's

.entrypoint

is a completely different file (containing only my simple test flow) which has no outside dependencies beyond prefect. My understanding was that other files would only be read if they were imported in the entrypoint file? • What user on the pod is executing this code that is downloading from Storage, retrieving the flow, etc? Thanks for any info you can give!

Hello, We're running on Prefect 2.7.8. Up until today our Prefect deployment CI/CD job was working just fine. We haven't performed any recent version upgrades to our environment or our Python dependencies. This error began to occur a earlier today:

httpx.HTTPStatusError: Client error '422 Unprocessable Entity' for url '***/deployments/'
For more information check: <https://httpstatuses.com/422>

A version upgrade to Prefect 2.7.9 appears to have solved the issue. However, it's strange that our CI/CD for deployments started failing seemingly out of no where. Does anyone know the cause of this issue?

With my own endpoints using FastApi, would it be possible to call a flow locally, return the flow id immediately and have the flow run in the background? I tried with BackgroundTasks but I get a weird error. Thank you.

In Prefect v1 is there a way to detect which user kicked off the flow run?

How do you handle different conda/poetry environments? How do you specify which python env to use in the deployment?

Hi! Sorry if this is not the correct place to ask, I would like to know more about Marvin (I tried to ask him but I understand the code is not public). I'm looking for a slack bot to export a message thread to a git issue, and here is the only place I saw a bot having this exact feature (I'm not looking for all the all-knowing super bot features that offers Marvin, just this), is this a something you are able to share ?

Hello guys would like to ask how can i get the flow schedule using python api or prefect cli? The main idea is that at the moment I have to check in the UI of each flow its schedule in the ui and it takes a lot of time to check several flow schedules, if we also assume that the number of flows will only increase, I decided to find an optimization of this process. I did not find information on how to do this, only how to set a schedule

Hey all! Is there some syntax similar to airflow to set the task execution order? For example:

task_a >> task_b
task_b >> [task_c, task_d] >> task_e

Hey all, I have the following question I have a class with

class TestClass:
    def __init__(self, atr_a, atr_b):
        self.atr_a=atr_a
        self.atr_b=atr_b

@task()
def test_method(self, text):
    texts = self.atr_a(text)
    ...
    ...

Than i have a other script whicht defines my flow

@flow()
def my_test_flow():
    test_class = TestClass(atr_a=..., atr_b=...)
    test.class.test_method(text=text)

If I run this example I get the error: Function 'test_method' has signature 'self, text' but received args: (....... ) Is there a way, I can call a Task inside a class?

Hi all ! We're currently fighting with mapping features, which overflow the backend when mapping over many inputs (which is a requirement for our flows - typically over 50) and cause the flow to crash. When running locally, the flow crashes with a message that the database is locked (on the example below) - when running on the server, we do get an error 500 back after many creations in a row (

Crash detected! Execution was cancelled by the runtime environment.

and then

Crash detected! Execution was interrupted by an unexpected exception: httpx.HTTPStatusError: Server error '500 Internal Server Error' for url 'http://[our instance url]/api/task_runs/'

For more information check: <https://httpstatuses.com/500>

) A minimum running example is (in Python 3.10 with prefect 2.7.9) :

import time
from prefect import unmapped, flow, task

@task(name="Sleeper {id}")
def sleeper(id, seconds_to_wait: int):
    print(f"Start Sleeper {id}")
    time.sleep(seconds_to_wait)
    print(f"End Sleeper {id}")

@flow(
    name="Prefect 2 concurrency checker",
)
def flow(n_runs: int = 128, seconds_to_wait: int = 2):
    res = sleeper.map(list(range(n_runs)), seconds_to_wait=unmapped(seconds_to_wait))
    return res


if __name__ == "__main__":
    res = flow()
    print("done")

In Prefect 1, we had a way to limit the number of task run creations, but that's apparently not available in Prefect 2 AFAIK (note: we've tried tag-limiting, but that's not helping as it's the creation of the many tasks that kills it, not their actual run ; we've also tried an async approach, which fails in the same way). My questions to you: 1- Have you ever experienced this too? 2- Did you find a walk-around? 3- If not, do you have an idea of how to manage this?

seeing a lot of api errors today. have seen the following error message in 4 different flows overnight. have never seen it before. any idea on how to fix?

Failed to set task state with error: ClientError([{'path': ['set_task_run_states'], 'message': 'Operation timed out', 'extensions': {'code': 'API_ERROR'}}]) Traceback (most recent call last):   File "/usr/local/lib/python3.9/site-packages/prefect/engine/cloud/task_runner.py", line 91, in call_runner_target_handlers     state = self.client.set_task_run_state(   File "/usr/local/lib/python3.9/site-packages/prefect/client/client.py", line 1622, in set_task_run_state     result = self.graphql(   File "/usr/local/lib/python3.9/site-packages/prefect/client/client.py", line 465, in graphql     raise ClientError(result["errors"]) prefect.exceptions.ClientError: [{'path': ['set_task_run_states'], 'message': 'Operation timed out', 'extensions': {'code': 'API_ERROR'}}]

Hi All. I’m trying to understand how I should structure the code in my project. Obviously, I’ll have flows and probably separate files for tasks. What I’m struggling with is where should I store common code, such as custom libraries and database models?

project
  flows
    flow1.py
    task1.py
  common
    db.py
    utils.py

I can get that to work if I have the proper PYTHONPATH in place, but that seems wrong because the flow/task code is run from the

/tmp

directory, but the common code would be running in the original source location. Any hints? Thanks!

Hello, I have a question about notifications of task status. I can write a task to send status of a given task as a Teams message. However, if a tasked is mapped 10 times, I would receive 10 notifications. It would be great to either: 1) wait until all mapped tasks are complete and send notifications of all mapped tasks 2) post all notifications in the same Teams thread Does anyone have any suggestions about either of those route?

Quick one RE the cloud API (2.0) - I want to get the count of task runs for a given flow run ID. Am I right to hit the

read_task_runs()

endpoint and count the number of task runs returned? And RE the max limit of 200 task runs per API call, am I right to loop over it in the way that I’m doing in the thread, or is there a neater pattern? Thanks!

Hi, I'm excited to learn more about Prefect V2! I had a couple questions as we're exploring some specific use cases. Is there a way to run tasks downstream from a specific point in a flow? For example, suppose a user has the following flow

a >> b >> c >> d

. After the flow has run, what is the best way to run tasks

b

to

d

?

Hi I’m trying to run prefect orion on EKS with gitlab storage block. I have prefect-gitlab installed in the prefect agent and registered the blocks as well that I can confirm in the UI and from the console. However, for some reason, when I try to run the flow i get this error:

KeyError: "No class found for dispatch key 'gitlab-repository' in registry for type 'Block'."

Am I missing a step somewhere?

Hi team, I am new to prefect and GCP. In bigquery_create_table() from

prefect_bigquery

lib, what should I fill in the

location

param if I’ve specified the region for my GCP project as asia-southeast1? I’ve tried to leave it as default (“US”),fill it as

None

,

Asia

,

ASIA

, and

asia-southeast1

,but it’s kept showing the same error as below

File <my_path_to_venv>/lib/python3.10/site-packages/prefect_gcp/bigquery.py", line 223, in bigquery_create_table
    client = gcp_credentials.get_bigquery_client(project=project, location=location)
AttributeError: 'SecretDict' object has no attribute 'get_bigquery_client'

Hi, does anyone know how to set environment variables for flow runs? I can’t find anything on how to do it with prefect 2.

Hey all, I have an event trigger azure function that kicks off a prefect deployment run using create_flow_run_from_deployment() whenever a blob is uploaded, however when there are multiple files uploaded at the same time, the az function crashes with this traceback msg. I would like the az function to concurrently create flow runs from a deployment. Is there a way to accomplish this? the img is the error msg of one of the crashed flows/az function.

Hey Community! Is there a way to wait for another deployment's flow to finish before doing something in our main flow? i.e. I have a deployment/pipeline with three sub-flows A, B, and C, I want to run sub-flow A first, then it should check if another deployment's main flow has finished or not if it has it should trigger sub-flow B. More like a wait_for functionality but instead of tasks or flows within the same deployment, it should wait for another deployment's main flow to finish first. I am able to come up with a solution using Prefect Orion API where my deployment A takes the name of the second deployment's flow on which sub-flow B depends and then it checks the last execution of the second deployment's flow's state for the current day, if the state is finished, it would go on to execute the sub-flow B, if it's in any of the pending states, it would just hold on and retry for an hour and wait until it completes. it's working okay for now but I think this is a naive approach. I was just wondering if there is any prefect native approach to do this?

Hi everyone, does anyone have experience dealing with this or similar issues?

prefect.exceptions.UpstreamTaskError: Upstream task run 'None' did not reach a 'COMPLETED' state.

It doesn't appear to happen consistently, so it has been difficult to debug. We are currently on Prefect

2.1.1

.

Hello! Anyone else receiving this error when they try to log into Prefect Cloud with Google Sign In 😿 : It's working now, disregard!

Hi all 👋! Is there some way to interact with the data produced by

persist_result=True

programmatically in Prefect 2? e.g. initialize a local flow run with persisted state/results from some remote flow run? Usecase: Flow fails in remote environment a few hours into a run. It'd be nice if a developer could run that flow locally in their debugger, but initialize the flow using the persisted results from the remote run to skip straight to the tasks that failed.

Hello, we are in the process of porting our Prefect 1 flows to 2. There are several places that we used to get the flow name from context. Based on some articles like this. It seems like it may still be possible. However, when I call the suggested methods, I don't see flow name available. Looking at the porting guide I don't see that flow name is available in the "Example Output". Also a community member seems to be having the same problem in the comments below. Is there any way to get the flow name contextually during a run in Prefect 2? I'm running my flows locally in a docker container that matches our production environment with a fresh pull of

prefecthq/prefect:2-latest

Thanks!