Hi, just wanted to drop this query here as it might have been missed .

Hi, is there a way to pull an image from AWS ECR and run the image? the only one I found is to pull from Docker server which doesn’t apply to my case.

I found this library particularly helpful for deferring flow configuration until it is actually needed. This lets me import the

flow

object without having

MY_BUCKET

secret available (e.g. for generating documentation), while still materializing the config correctly when the flow is run or registered. Curious if anyone has seen similar use cases or solutions?

from prefect.client import Secret
from prefect.storage import S3
from lazy_object_proxy import Proxy

flow = Flow(
    ...,
    storage=Proxy(lambda: S3(Secret('MY_BUCKET').get())),
)

Hey Prefect community 😎 had a question about local flow execution that I'm hoping someone can help with. Before the local prefect runner, we were previously running/debugging flows locally like this:

import os

from flows.wherever import flow

from prefect.executors import LocalDaskExecutor
from prefect.storage import Local

context = {
# ...
}

flow.executor = LocalDaskExecutor(scheduler="threads", num_workers=8)
flow.storage = Local()
state = flow.run(
    param="something"
    context=context,
)

This works fine, but doesn't seem to fly with flows that call

StartFlowRun

, since it's trying to reach out to a server. Anyone have a way to do dependent flows locally? Not sure if the new local runner can do it, I haven't had the chance to try it yet 🙂

Hi all, I’m having some issues building + registering my flows via Jenkins. I am using

prefect build …

to first build the flow and output a

flow.json

to then use in my

register

command, but the build is failing. with the error message

docker.errors.DockerException: Error while fetching server API version: ('Connection aborted.', FileNotFoundError(2, 'No such file or directory'))

I believe this is a side effect of trying to build a docker image whilst running in a docker container, has anyone ever run into this issue?

Hi guys. Anybody had success making the Agent/Job recognize

imagePullSecrets

using the Helm chart? I can make it work when I manually write the

deployment

and pass the

--image-pull-secrets

argument, but I can’t make it work passing the secret argument on the helm chart. Any ideas what I might be doing wrong? I doing something like

imagePullSecrets: mysecretname

in the Helm (inside the “job”, inside the “agent” part)

when running flows with the Kubernetes Agent, is prefect able to recover/restart a flow that has the dask-scheduler pod killed by kubernetes? (for oom reasons or due to a spot node dying etc)

Hi what is best way to create task that fetches data from regular server usually I would do this with scp, rsynch or something similar

Is there a nice way to map with tasks that return tuples? Here’s what we had without mapping:

part1, part2 = my_tuple_task(input)
otherresult1 = othertask1(part1)
otherresult2 = othertask2(part2)

However now we want to map over multiple inputs to have multiple parallel pipelines of the above:

whole_tuple_result = my_tuple_task.map(inputs)
otherresult1 = othertask1.map(whole_tuple_result) # tasks must break up tuple inside the function
otherresult2 = othertask2.map(whole_tuple_result)

Is there a way to maintain the elegant tuple-destructuring while still mapping over the result? Just trying to do it directly gives the error

TypeError: Task is not iterable. If your task returns multiple results, pass nout to the task decorator/constructor, or provide a Tuple return-type annotation to your task.

(we’ve already set that which is why the first example works)

Hello all, how can i deploy the same flow for sevral projects with some different predefined variables (not in input)? thx

Hi everyone, I'm facing some issues running two agents in different conda environments. I have a LocalDastExecutor agent in a conda environment with uses Python 3.6 (let's name env_1) and runs on a Windows Working DIR C:\Project_1, and another (LocalRun) in env_2 with uses Python 3.9. The first one is running okay, but when I try to run the second agent on Prefect Cloud I receive the following error:

Failed to load and execute Flow's environment: FlowStorageError("An error occurred while unpickling the flow:\n TypeError('code() takes at most 15 arguments (16 given)',)\nThis may be due to one of the following version mismatches between the flow build and execution environments:\n - prefect: (flow built with '0.15.6', currently running with '0.15.3')\n - python: (flow built with '3.9.7', currently running with '3.6.12')",)

I already tried to use LocalRun(working_dir="C:\Project_2") but doesn't seem to work either. How can I specify to the second agent that he needs to use the desired "env_2"?

Hi, I wonder whether it is possible to set dynamic context / environment variables during flow runs? (This context is only known at run time)

Hi. Is it normal act, that running container doesn’t stop after flow was stopped via UI?

Hi there! I’d like to use the

RenameFlowRun

prefect task to rename my flow based on a value in the context. I noticed that if I try something like:

rename_flow = RenameFlowRun(flow_name=f"{prefect.context.today}")

It tries to evaluate the context at deploy time, instead of at runtime. Is there a reasonable way to do this? @Rob Douglas ended up just pulling the guts out of the

run()

method of

RenameFlowRun

and putting it in a task. That works, but it seems a bit dodgy. PS - the documentation for this is wrong, as the example shows importing

FlowRenameTask

instead of

RenameFlowRun

Hi everyone! Is there any way not to fail the entire mapped task if one of the results is Fail?

Hey all 👋 First time here - I’m learning prefect at my company. Is this the right place to go to ask questions or potentially troubleshoot? Or is there a better channel for that?

Has anyone had success using Git storage with an Azure DevOps repo? Is this supported?

Hello, I am trying to test “flow of flows” using StartFlowRun task as provided in docs…but while running the flow getting this error ‘StartFlowRun’ object has no attribute ‘poll_interval’…. has anyone run into same issue before ?

Hi all, tried using the slack notifier as described in the docs (installed the prefect app, added the webhook url in the secrets) but getting this error Any ideas?

hello! I have a question re: cloud pricing. one of the suggested solutions for sensor-type workflows is to have a task that raises a pause or retry signal (eg. as described at https://github.com/PrefectHQ/prefect/discussions/4156), and only progress to success when the task finds an appropriate file/event/whatever. am I right thinking that these retry+pause runs would not count as success for pricing purposes? So I’d only be paying when the polling task actually finds work to do?

Hi, I keep having a problem of

The secret KUBERNETES_API_KEY was not found

when trying to run a Kubernetes Task. I thought I’m using in-cluster connection so it shouldn’t be required? Otherwise is there any recommendation on creating a ``KUBERNETES_API_KEY` ? according to: https://docs.prefect.io/api/latest/tasks/kubernetes.html#createnamespaceddeployment

Attempt to use a Prefect Secret that contains a Kubernetes API Key. If 

kubernetes_api_key_secret

 = 

None

 then it will attempt the next two connection methods. By default the value is 

KUBERNETES_API_KEY

 so providing 

None

 acts as an override for the remote connection. 2. Attempt in-cluster connection (will only work when running on a Pod in a cluster) 3. Attempt out-of-cluster connection using the default location for a kube config file

Wanted to ask this here before raising an issue on Github. Is there a reason the EmailTask cannot send emails via an internal SMTP server, without SSL?

So I'm looking at creating multiple flows with some shared configuration and utilities. I'm going to use the docker storage for this, but I'm curious how others have structured their repositories where there are many different flows. Do you keep a flat directory of flow files (with no external dependencies or shared code) and use S3 / git storage? Or do you build a separate docker storage for each flow in CI / build a single storage and include all the flows (as described here); if you do this what does your project structure look like? I'm looking to run the flow build and registration steps in CI; I'm assuming this is a common approach. The documentation seems to suggest that it isn't, as most of the examples point towards the other storage types, and running the flow setup locally.

I have a strange issue where changes I make to Flows are not being reflected in the execution environment. I am running a docker agent on an AWS EC2 instance, and when I make updates to a Flow, those updates are not reflected at runtime. All of the storage building and registration of the Flow appear to happen successfully, and when i look at the terminal output on the EC2 instance i see

Successfully pulled image XXXXX

and

agent | Completed deployment of flow run XXXXX

, but the flows content is not being updated . When I change the runtime Labels on Prefect cloud and execute the same (updated) flow in another environment (docker agent running on mac osx) the updates i made to the flow are reflected…

Hi all, a couple questions about state handlers. 1. I want to run a function f() when a task finishes successfully on retry. In other words, a task goes from failure —> retrying —> running —> success. How can I capture that scenario in a state handler if I only have new_state and old_state? 2. Where can I find a task’s start and end time from within a state handler? For context, I want a state handler to log this information in the event of a task failure. Thanks so much!

Hi! I'm getting the following error when I am trying to run a flow locally through Prefect UI. I am launching both the

prefect server

and

prefect agent

from a directory where version

3.8.9

is active (I am using

pyenv

), so I don't know where this version

3.7.10

comes from. How can I tell which python version my agent uses? I assume, according to the error message, that the faulty version

3.7.10

is tied to the agent.

Failed to load and execute Flow's environment: StorageError("An error occurred while unpickling the flow:\n  TypeError('code() takes at most 15 arguments (16 given)')\nThis may be due to one of the following version mismatches between the flow build and execution environments:\n  - python: (flow built with '3.8.9', currently running with '3.7.10')")

Hi! This might have been raised before. I’m running the latest version (0.15.6) on a virtual machine. It works on localhost, but is not reachable via external IP. If I try the same with version 0.15.0 everything works as expected. Is this new behaviour and we need to do more to make it work like before? Thank you!

I’m new to Prefect and have been attempting to run a flow using ECS. Unfortunately, my containers keep stopping immediately and the only info I see says 

Essential container in task exited

 . I’ve been trying to debug this for a bit and am not sure how to proceed. Any advice?

How do we pass multiple parameters in Flow, is below syntax good (second parameter is 2d array)

flow.run(parameters=dict(studentId=student, data=data))

Hi, I am trying to set up task orchestration, where all the tasks will be AWS Batch jobs or similar (no actual processing happens within a task). What's the recommended configuration here (e.g. what agent to use, how to configure)? The setup should accommodate some concurrency (say up 10 flows run concurrently, and up to 40 tasks run concurrently), with minimum resource usage - as these should only be API calls.