Hi All, I am trying out Prefect 2 for the first time I have a flow that imports some functions from another (local) python module In prefect 1, we were able to do something like

storage.files.update(THE_FILE)

to bring in the dependency What’s the recommended approach to achieve something similar in prefect 2

Hi, there is this LOOP construct in 1.2. I was wondering when that will become available in 2.0. Thanks!

Hey there ! I have a mapped task for uploading files through an API call and I'd like to avoid having to authenticate to the API for every task but my client is not JSON serializable and cannot be passed as a parameter. I'm pretty sure I've seen a way to properly deal with this before in the docs but can't seem to find it again. Any idea or advice ?

Hi! Is there a way to start an agent via a python script instead of the command line?

Hi all, would anyone know why a flow would be running tasks in a different order than expected? It seems that the flow that I am running locally is constantly changing the order of how tasks are run. I haven't had this issue until recently

Hey! I'm working with Prefect in Azure using K8s, ACR and Prefect Cloud. Is there a setup for affinity or taints/tolerations? I'm trying to put tolerations in the KubernetesRun template and the DaskExecutor as well. It works for tolerations, but on the DaskExecutor, it gives this error:

RuntimeError: Missing dependency kubectl. Please install kubectl following the instructions for your OS.

Does anyone know about this error with K8s tolerations?

Hi! I know I can delete a flow using

from prefect.client import Client

flow_id = "..."

client = Client()
client.graphql(
    """
    mutation {
    delete_flow(input: {flow_id: "%s"}) {
        success
    }
    }
    """ % flow_id
)

But how can I delete all versions of a given flow? It looks like the Flow Group ID is constant over the flow versions. I tried to replaced it into above snippet but didn't work.

Hi, I run some flows directly from vs code, and I stopped them manually from IDE before they finished. It was a few hours ago. In meantime I restarted computer and UI still shows me them as running. Is that normal? How can I clean them?

Hi, running this in container (AWS ECS) failed but works running from desktop computer

Hey, I'm currently testing prefect 2.0 as part of our new data warehouse and so far I'm very impressed! I'd like to use prefect cloud 2.0 and run my flows in my k8s cluster. There is this documentation here: https://orion-docs.prefect.io/tutorials/kubernetes-flow-runner/ - which shows how to implement my use case for a OSS orion server. How would this example work for prefect cloud 2.0? Do I simply use the manifest as output from prefect orion kubernetes-manifest and only deploy the agent, role and role binding but not the API? And for the PREFECT_API_URL env variable I use https://api-beta.prefect.io/api/... instead of orian:4200?

Hi everyone. I am learning how to use Prefect 1.0 at the moment. I am building out a very basic flow and that needs to retry given some condition. ` I started down the path of trying the

raise signals.RETRY

signal but the task ends up going into a tight loop. I decorated the

@Task

with

max_retries

and

retry_delay

but the retry is still repeating rapidly and without any delay. ` Is the

RETRY

signal controlled by the

@Task

decorations

max_retries

and

retry_delay

?

Hi, another strange question, when I use DaskTaskRunner i see my print logs but not logs from get_run_logger, when I do not specify runner then I see both ... I run it locally using python3

in orion’s ui is there a place to see flow run parameters?

Hi! Is there a way to run a flow in batches? For example, if I have 100000 records, is there a way to pass 100 records at a time through a flow?

Hi, I have 2 ECS containers, each with local agent connect to prefect cloud. When I invoke multiple flow run in parallel, the run distribution among agents are not balance. Eg, I start 10 flow runs in parallel, one agent pickup 9 run and the other agent pickup 1 run. How can we control this?

Hey everyone! Quick question: Are the public IPs for Prefect documented anywhere, and are they static IPs? I need to allowlist them for a database connection and I can’t find any docs about it

Hello, Frank Hardisty of Replica here, we are heavy Prefect users, thanks for all the great functionality! Our organization has some Task Concurrency limits, very useful. However, some of them have become inaccurate, for example we have one called

ClickhouseODHourly

which is permanently stuck at 136 even when nothing is running.

Hello! Is there any easy way to pass result of one task (a dict) into the input of another (as **kwargs)? Here's a minimal example of what I'd like to do if it helps:

import prefect
from prefect import task, Flow

@task
def parse_task():
    name = prefect.context.flow_name
    return {"name": name, "test": 123}


@task
def split_kwargs_task(name, test):
    print(name)
    print(test)


# Does NOT work, but desired approach
with Flow("hello-flow") as flow:
    parameters_cleaned = parse_task()
    split_kwargs_task(**parameters_cleaned)  # <-- raises error

# Manually pulling out each kwarg gets this to work
with Flow("hello-flow") as flow:
    parameters_cleaned = parse_task()
    split_kwargs_task(
        name=parameters_cleaned["name"],
        test=parameters_cleaned["test"],
    )

flow.run()

Hello. I am trying to set up prefect server on an Azure Ubuntu VM. How do I register flows from my local machine?

hy team If i register a flow then change the script , do i need to register the flow again or it will automatically run the updated script

Hi all, seeing this error with docker run : Unexpected error: TypeError("'NoneType' object is not subscriptable") Traceback (most recent call last): File "/usr/local/lib/python3.7/site-packages/prefect/engine/runner.py", line 48, in inner new_state = method(self, state, *args, **kwargs) File "/usr/local/lib/python3.7/site-packages/prefect/engine/task_runner.py", line 869, in get_task_run_state logger=self.logger, File "/usr/local/lib/python3.7/site-packages/prefect/utilities/executors.py", line 299, in run_task_with_timeout return task.run(*args, **kwargs) # type: ignore File "<string>", line 519, in delta_push TypeError: 'NoneType' object is not subscriptable Any ideas?

Hi - is there a link someone could share to some resource detailing the ins and outs of alerting best practices in Prefect? Thanks!

Hi! Is there a good way of differentiating (Prefect Cloud UI) two scheduled runs of the same flow but with different parameters (for example staging vs production)?

Hi all, does anyone know if there is a parameter for the job that can be set to “allow to fail” Not to get confused, the use case is different We are running some validations during our complete etl flow, orchestrated in Prefect one of the flows is this validations, but if any of the validation is false, the flow will fail and also the entire job and that we would like to avoid so our steps are, we trigger one flow consisting on 5 different flows in cycle: 1. calculations finish 2. load finish 3. data transfer finish 4. validations finish (found one error but it finished the job) - after that everything fail, next job is not triggered we would like to be able to trigger the next job even if the validation founds some errors, since it will trigger Slack webhook and inform us, and we can sort the error, but our job will continue is something like this possible

Hi, apparently Kubernetes Agents are associated with certain tenants. By default they use the default tenant for the API key used. Is there a way to somehow use an agent across multiple tenants?

Hi, I tried updating my default storage using the api (PATCH /api/blocks/{id}) since my connection string expired and this is not yet possible using the cli. However, I'm getting a "malformed uuid" exception, while the query does execute perfectly when I query my postgresql database directly. I'm using prefect 2.0b5. Full stack trace in thread

Hey, i'm starting to play with prefect Orion for a project and i was wondering how dynamic the flows can be ? For this project i would need to generate a flow dynamically depending on a json config. The tasks would be static but the execution of those tasks can be very different from a run to an other. For example, i have an extract task. A flow could be the execution of this extract task once or could be this extract class followed by the same extract task with different parameters and using the output of the first one. I started to play with for loops but it seems that it is made to execute tasks in a parallel manner. I haven't found how to set execution plan when using a for loop.

Hi. This question is more related to Dask, but probably someone can help me. Is it possible to use 2 different default distributed Clients connected to external clusters in parallel Prefect tasks?

Hi, is there a way to keep a custom object with values on the task context ? so each task will have it’s own instance of that class with different values.