I use aws ecr for prefect docker storage. I log in to ECR periodically with the following command "`aws ecr get-login-password --region yourregion | docker login --username AWS --password-stdin yourregistryid.dkr.ecr.yourregion.amazonaws.com`" because the ECR token is expired after 12 hours. But In prefect agent, docker client is not updated when the token is expired. So every 12 hours, I have got 404 error and have to restart my agent.

I tried to register a flow to Prefect cloud and store it on S3. AWS_CREDENTIALS is set as secret within Prefect Cloud. If I register the flow i receive the following error:

botocore.exceptions.NoCredentialsError: Unable to locate credentials

I register the flow with the following method:

flow.storage = S3(
    bucket="prefect-flows",
    secrets=["AWS_CREDENTIALS"],
)
flow.register(project_name="abc")

The backend is set to cloud and I logged in successfully (

prefect auth login ...

) How could I use the AWS_CREDENTIALS secret which is set in the cloud?

Hello prefecteers! I'm having some trouble with what appears to be a deadlock. I've gotten a minimum reproducible example that may or may not be the cause for my issue, which centers around tasks that submit additional work to the dask client which serves as the executor. I am using prefect core with the dask executor. Here is the example:

Hello, I’d like to schedule multiple runs of a parametrized flow at the same time, with each run having a different parameter input. I’m thinking I can either build a parent flow that uses the python client api to initiate each child flow run, or maybe its possible to have a parentless approach where multiple runs of the flow are on the same schedule, each with a different input. Another possible solution using a parentless approach could involve an external data source that stores the set of parameters and keeps track of whether each one has been picked up by a flow run. I’m looking for best practice as I don’t think these are great approaches. Would appreciate any help here! Thank you.

hi @nicholas, following up on this thread: we still get logged out every day and have to log back in in the morning, which causes all of our open tabs to redirect to 404s. is this daily logout the intended behavior, and is there a way to avoid the 404'ing?

So I'm digging into the prefect repo and see that AWS credentials are handed off as a credential dict to a MagicMock() function. The reason I'm looking is because passing CodeCommit(secrets=) does not work outright whereas S3(secrets=) does work, and our flows run remotely. While trying to pass in client_options={'credentials':['AWS_Credentials']} yields "TypeError("get_boto_client() got multiple values for keyword argument 'credentials'",)". This makes sense as this seems to be how prefect calls boto:

@property
    def _boto3_client(self):  # type: ignore
        from prefect.utilities.aws import get_boto_client

        kwargs = self.client_options or {}
        return get_boto_client(resource="codecommit", credentials=None, **kwargs)

and then the get_boto_client func:

def get_boto_client(
    resource: str,
    credentials: Optional[dict] = None,
    region_name: Optional[str] = None,
    profile_name: Optional[str] = None,
    **kwargs: Any
) -> "botocore.client.BaseClient":

but it seems it holds onto None and doesn't let anything else pass. Is there anyone else here that has successfully connected to CodeCommit through Prefect?

I'm suddenly locked out from being able to update my flows with the following error: My flow has a registration statement at the very end and it shows up on the front end, but I can't run any of my flows anymore. "CRED_secrets" does not have a -1 next to it in my flow and is also titled CRED_secrets on the front end secrets module. Context: this is a simple flow that has run consistently prior to this error. CRED_secrets is being called with client_secrets=PrefectSecret("CRED_secrets").

Traceback (most recent call last):
  File "/usr/local/lib/python3.6/site-packages/prefect/engine/flow_runner.py", line 245, in run
    parameters=parameters,
  File "/usr/local/lib/python3.6/site-packages/prefect/engine/cloud/flow_runner.py", line 402, in initialize_run
    raise KeyError(msg) from exc
KeyError: 'Task slug CRED_secrets-1 not found in the current Flow; this is usually caused by changing the Flow without reregistering it with the Prefect API.'

Upgrading from 0.13.19 to the latest version rocked! One flow in particular went from 35 minutes to 6 minutes. Just out of curiosity, what might have caused that huge speedup? Curious as to what went on under the hood between 0.13.19 and 0.14.12!

Hi - I am running into an issue with

PrefectSecret

and

coiled

. Posted the question on stack in case somebody else runs into this. https://stackoverflow.com/questions/66664086/how-does-one-use-prefect-cloud-secrets-with-coiled

Dear all, trying to figure out the difference between using a task class and a task decorator

from prefect import task, Task

# Example 1

class Example(Task):
  def run(self, **kwargs):
    do something

# Example 2

@task
def example():
  do something

In particular, how do I force a rebuild of the second task (using decorator)? Running into some issues running flows using ECS / ECR, where the task would not get rebuilt if using decorator format

Team, 1. Please advise on how to get notified when the prefect agent stops due to ec2 restart...we are using backend as the server and not the cloud version 2. How to check in the shell script to see if the agent is already running, if not start the agent

Hello. We are using prefect with its free plan, and yesterday when I tried to deploy a new flow, I got this error message: 'prefect.utilities.exceptions.ClientError: [{'path': ['create_flow_from_compressed_string'], 'message': 'This tenant already has the maximum number of flows.', 'extensions': {'code': 'INTERNAL_SERVER_ERROR'}}]' I looked for this message in this slack, and the two related topics are using paid plans. Could you explain me if it is related to free vs paid plans please, because I cannot find this information in your website plans comparison page? And how could we solve this limitation? (notice than I checked, and we currently have 25 flows deployed in total amongst our different prefect projects). Best !

I have a question about Prefect Cloud and FlowRunners. It looks like, if I want to loop a flow based on state, I need to run it through a FlowRunner. How would I do that through Prefect Cloud?

Is there any documentation regarding cleaning the postgres in a standalone cluster? Is the database growing forever? Can I trigger deletion of e.g. "more than 30d old" state (flow runs, logs)? Mayby using a mutation or api call? (I'm not talking about STORAGE or RESULT) Edit: found an old answer: no automatic process available. 😞

Hi all, where I can find details in docs about this?

Realese 0.14.11:

Added 

project_name

 /`project_id` to flow run execution context in orchestrated flow runs

Hello all, I have what I think is a simple question: I have a flow that creates a container, waits on the container, and then gets the container logs. How can I have it throw an error and bail out on things if the WaitOnContainer has a failure in it? I have a slack notifier set up to only output failure, but it is not notifying upon failure of WaitonContainer. Flow code for reference in the thread

map throttling with cloud task concurrency and dask cluster with multiple workers! 🔥 🔥 🔥

interesting issue I created here in case anyone has ever run into issues using case and skip_on_upstream_skip=False https://github.com/PrefectHQ/prefect/issues/4264 there is a work around that is helpful in there

Hi all, what kind of permission does the Postgres "prefect_user" in config.toml file needs...I am seeing that the prefect server is starting only if the postgres "prefect_user" is part of rds_superadmin, otherwise getting error "Could not upgrade the Database!". All I am trying to do is to avoid assigning elevated access to postgres "prefect_user" id, because for prefect we are using postgres that is used by our OLTP applications. Also advise what kind of database upgrade happens every time the prefer server is started.

How do I save the results of a task to the local filesystem? Suppose I have a workflow

test.py

from prefect import task, Flow
from prefect.engine.results import LocalResult

OUTPUT_DIR = './outputs'


@task(target='x.txt', checkpoint=True, result=LocalResult(dir=OUTPUT_DIR))
def saver():
    return 10


with Flow('example') as flow:
    saved = saver()

flow.run()

when I run

python test.py

I expect a file tree like

tree

.
├── outputs
│   └── x.txt
└── test.py

but when I run

python test.py

I don't see an

x.txt

file

.
├── outputs
└── test.py

I am using

prefect==0.14.12

What do I need to do to get

prefect

to save (and persist) file results?

I have a flow that I want to use to backfill data for about 9 years which is going to equal around 3000 jobs. I am curious if Prefect Cloud will have any issues queuing up so many jobs and then just running them sequentially.

I'm trying to register a new flow version (ie: same name) and kick off a run, but the updated labels don't seem to be getting used so it gets run on the wrong agent. Is there a way to set label per-flow or per-flow-run that override the flow_group?

Is there a good way to log to Prefect from a jupyter notebook running in

ExecuteNotebook

?

Hey all, @Adithya Ramanathan and I are running into some issues with Prefect’s mapping functionality and would appreciate some more information on the mechanics of mapping in Prefect! My understanding is that currently, having two consecutive mapped components allows you to run the second component on the same level of mapping as the first component, like in one of the examples in the docs:

from prefect import Flow, task

@task
def add_ten(x):
    return x + 10

with Flow('iterated map') as flow:
    mapped_result = add_ten.map([1, 2, 3])
    mapped_result_2 = add_ten.map(mapped_result)

This allows you to map the first add_ten call to the three elements in the list, and then to map the second add_ten call to the three outputs from the first call. Instead of returning a single int from the add_ten function, we want it to return an list of ints and continue to fan-out (to 6 elements in the example below):

@task
def add_ten(x):
    return [x + 10, x]

with Flow('iterated map') as flow:
    mapped_result = add_ten.map([1, 2, 3])
    mapped_result_2 = add_ten.map(mapped_result)

Is this something that is supported by Prefect, or can you only have 1 fan-out per map/reduce?

Have just come across this error when trying to use my own custom decorator;

ValueError: Tasks with variable positional arguments (*args) are not supported, because all Prefect arguments are stored as keywords. As a workaround, consider modifying the run() method to accept **kwargs and feeding the values to *args.

Has anyone had any luck using their own? The one I want to use is this;

def jitter(func: Callable) -> Callable:
    """
    For use with short running mapped tasks
    to avoid overwhelming the api server

    calls function sleeps for random period
    """

    def wrapped(*args, **kwargs):
        res = func(*args, **kwargs)

        sleep(random.gauss(1.2, 0.2))
        return res

    return wrapped

Basically just waits for a while and does so randomly so lots of short running mapped tasks don't all hit my api server (on k8s) - I much prefer this than having sleeps in my actual task code

Are there any code examples using the

files

arg for the Docker storage? I'm trying to use some common scripts across flows packaged into separate Docker images. I think the

files

arg will serve my purpose, but the Prefect API docs suggest to use absolute paths, while relative paths would make more sense.

Hi all, I was looking for a feature.. For a given flow run I want to restart the flow from taskA to taskB only. Where task A is an ancestor of taskB in the flow. All the tasks have a target path. Any possible way to simply achieve this..?

I’ve read through the docs and cannot figure out why my use of LocalResult on a task is not persisting results. There are definitely results, because I can print them out like so:

print(state.result[baseline_fractions]._result)

. All that happens is that my

dir

provided location produces the directories and then an empty folder. What am I missing here?

Hi folks, I am trying to understand how to make use of the

not_all_skipped

trigger, it seems the naming might be a bit not intuitive for me, or I might be missing something - please see the example in the thread (any help would be appreciated)

Hi Prefect Team! I am wondering if there is a way to set static

run_config

parameters on a Kubernetes Agent for subsequent flows to consume rather than duplicating similar

run_configs

across various flows. We currently make use of the

PREFECT__CLOUD__AGENT__ENV_VARS

to pass along

storage

configurations, but not having much success attempting to update the execution environment to use an internal custom image. I have tried passing the

IMAGE

and the

PREFECT__CONTEXT__IMAGE

environment variables to my job spec thus far to no avail.