Is there a way to download a file from S3 and save to a location, and skip re-downloading the file if it's already saved.

Hi, How i can store all task result of flow in specific S3 prefix?

Hi, we're running into some issues using Bitbucket cloud storage with a branch ref (as opposed to a specific commit ref). It looks like the

client.get(f"repositories/{self.workspace}/{self.repo}/refs/branches")

call to get the list of branches actually returns a paginated list, and the method is not able to find the hash corresponding to the branch if the branch is not included in the first page. Additionally, the error raised in case the branch is not found is not formatted properly (it's a string instead of an f-string).

Orion License: In This commit the claim that "As Orion matures, most or all of its components will be released under the Apache 2.0 license." was removed from the Orion docs. Does this mean that Orion will not be released under Apache 2.0?

Hey I’m running prefect with a kuberentes agent and temporary dask cluster I’m quite frequently getting this error:

No heartbeat detected from the remote task; marking the run as failed.

Is there some obvious things to look for?

Hi, another Prefect + K8S question here 🙂 Like the previous post we're frequently seeing the message

No heartbeat detected...

. Usually It happens in 2 situations: • the pod that run the tasks gets evicted / OOM killed • the pod was running on a preemptible node that gets removed and replaced. Is there something on the k8s agent, k8s job specification, or something else to configure to allow k8s to reschedule the job and let prefect know about it, so that the flow would continue ?!

Hi, I have a question about DockerStorage I did not manage to figure out on my own. I would like to pass a variable to the docker build so that it is replaced inside the docker file. It seems the env_vars in the DockerStore are not the right place. Any hints on how to do this?

what is the graphql query to get list of flow run ids for a given flow name ?

Hi again. I am using custom tasks defined as an extension of Task class (see extract.py) and would like to take advantage of the settings available for

@task

like _trigger=all_successful , log_stdout=True, max_retries_ etc. How can I set it up for these tasks? Would maybe using functions with decorators be advised as best practice here instead?

Is there a Global/Flow level context I can put some data on?

...is there any chance that ditching the DAG with Orion could maybe allow for hot-reloading of Flow code? (So like Error->Push changes->Restart from where you were, but with changes incorporated on downstream Tasks)

How can I make a feature request for MFA on Prefect Cloud? I feel that MFA is a basic security measure that should be available to anyone using Prefect Cloud. Prefect Cloud Enterprise offers SSO, but is cost prohibitive for most small teams.

Is it possible to add a single task at the END of a complex mapped task set? E.g. in the https://docs.prefect.io/core/concepts/mapping.html#complex-mapped-pipelines, have a single task that runs after the apply_map tasks. i.e. it runs only once, but only when all the mapped tasks are complete. I tried just giving apply_map a result and setting that as upstream, but doing so set the new task before the mapped tasks in the schematic.

Can the ECS Agent execute flows from a custom Docker image that we build? Our goal is to have our production and local development environments be as similar as possible. Locally we run flows in a custom Docker image with all of our Prefect code, that's based on the Prefect image. Currently we use S3 storage for production, and while we haven't run into any significant issues with that, we're prefer to make the environments more similar. I couldn't find the answer in the docs: • LocalStorage docs say it's only compatible with the Local Agent, so not with our ECS Agent. • Docker storage seems the closest to what we'd want, but I understand it would build a separate Docker image for each flow, instead of using a single Docker image.

Hi! We want to use state handler to create an OpsGenie alarm when the flow fails and resolve the same alarm when the flow succeeds after restart. is there a way to store the alert id somewhere so that it can be used for closing the alert later.

Hello everyone! I'm currently upgrading to the newest prefect version (from 0.15.3 - server 2021.04.06) and while re-registering my flows I'm getting a GraphQL Error: validation failed - expected one of the values edge_pkey for type „edge_constraint“, but found „“edge_flow_id_task_ids_key“. It only appears when I use a Parameter and use the returned value in a task. Registering a simple flow or even one of my old flows with the parameters disabled works just fine.

hi I have a question, I am new to prefect , I have configured prefect cloud to run the flow, the flow is created in an aws ec2 and I run the job and scheduled it , now its running on cloud ui my question is the actual execution happens in my prefect cloud or in my aws ec2 ? is there any way i can run my prefect flow with docker in prefect cloud ?

Hi everyone, I'm running a Prefect server and when we run some tasks generating a lot of logs (like 600 lines in a second), the UI stops to show them at some moment and there are no logs for the following tasks : they are not delayed, they just never appear (the prefect cli doesn't show as well). Is it a known issue ? Has someone run into the same kind of problem here ? Thank you !

I have created an Azure VM with Docker and Prefect Agent using the marketplace template "Prefect Agents". My flows and dependencies are stored in an Azure Container Registry. I used Cloud Prefect for my UI and got everything running as expected. I wanted to verify what happens with the authentication from the VM to the Container Registry if the VM ever has to restart. Once i restarted the VM, I noticed in the Cloud Prefect UI that the agent stopped polling. I am assuming that the VM restart, did not restart the Prefect Agent on the VM. Has anyone experienced this specifically when creating a VM with the marketplace template? Maybe this is a bug in the marketplace template that doesnt account for VM restarts. Under normal circumstances i would think there would almost never be a restart, but i know Azure isnt perfect, plus if there are updates that need to get done to the VM.

Hey! What would you say is the best way of running prefect in production with minimal flow / task startup time?

Just arrived! First question here. I am trying to figure out Prefect logging using the documentation here: https://docs.prefect.io/core/concepts/logging.html#logging-from-tasks But when I add this to one of my tasks and run it I see no log file:

logger = prefect.context.get("logger")
<http://logger.info|logger.info>("An info message.")

I tried running it from the interface on Prefect Cloud but that doesn't do anything either. What am I missing here?

quick question - i set-up my ecs agent with what I think is a custom image - say prefect with pandas. I then try to run a flow from cloud, simple hello-world with

import pandas

This doesn’t seem to work unless I specifically supply the custom image (and the task execution arn). I would have expected if run_config:image is null it would use the one from the task_definition? Does it default to something other than the container def?

Hi, I'm trying to get the response of a flow run through the REST API. I'm using `fetch(`http://localhost:4200/api/flow_runs/${response.id}`` to fetch the data and just getting

*data*: {encoding: 'orion', blob: '{"encoding": "file", "blob": "file:///tmp/78442fec16214918a437e768fa384762"}'}

in the

state

section. Is there something I should configure to get the whole response? Thanks

Hi, Just arrived. I have registered a flow and have added the CronSchedule. It gets scheduled on the right time but it is picking up the wrong date

flow.schedule = CronSchedule("49 21 * * 1-5",start_date=pendulum.datetime(2022, 1, 1, tz="Asia/Kolkata"),)

flow.register(
      project_name=project_name,
      idempotency_key=flow.serialized_hash(),
      labels=["test"],
    )

What could be the reason for this?

hi all - I upgraded the prefect version on one of my agents today, in the cloud ui I see two agents for the same label, however I am only seeing 1 agent pod on my k8s cluster, should I just remove this from the UI since the old pod no longer exists?

Hi, Why our cloud calendar is showing as empty even though we have few upcoming schedules?

Is it possible to get a task to show it's underlying function definition for intellisense in dev tools? For example the decorated function (with @task) is always just considered an Any type, so it doesn't show what variables it's expecting. (see screenshots) I'm hoping there is a way to show what variables it expects, so that when I build the task in one script and use it in a parent or child script I can see what inputs it expects. The way a normal function works (see last screenshot).

If a task returns a tuple, does that count as 2 task runs?

What is the best way to store a byte object as a secret and read it in during prefect? Example attached that is erroring out: