Git/GitHub would be nice to clone whole repo @Kevin Kho @Michael Duncan @Benjamin R. Everett We have a bunch of core logic in it's own package that we are building into a core base image to run our specific flows in using the ECS agent. Our specific flows have a fair amount of logic in a couple of modules. We see in the code that both Git and GitHub storage classes only wind up with the flow.py file getting loaded into the image at runtime. Granted they get there by two different. Git clones the repo, extracts the flow.py and then deletes the repo. GitHub downloads just the flow.py file. It seems like it wouldn't be all that difficult to modify the Git storage class to not delete the repo after it extracts the flow.py file and have the root of the modules on PYTHONPATH in the base image. Then when the flow is executed it would be able to find the modules and everything would be Prefect. While we are all noodling on this - is there anything obvious that keeps us from: 1. Creating a new GitRepo storage class that keeps the repo around 2. Adding it to the core package and building it into the core image 3. Adding where that repo's root will be to PYTHONPATH 4. running with the specific models loaded from Git.

Hello. I would like to understand my options for task output caching. Are my only options for caching (which can be invalidate when inputs/parameters change): • Memory, when running Prefect Core locally • PrefectResult, when running in Server or Cloud Or is it possible to use a different backend such as S3. Ideally I would like to use a combination of

target

and

cache_for

. So far I’ve not got this to work*, is this supported? *the target file is written and used by subsequent flow runs but it is never invalidated

result_location_and_target = "cache/{project_name}/{flow_name}/{task_name}.prefect_result"

s3result = S3Result(bucket="bucket-name", location=result_location_and_target)

@task(
    cache_for=datetime.timedelta(minutes=10),
    cache_validator=all_parameters,
    checkpoint=True,
    target=result_location_and_target,
    result=s3result,
)

Hi there! Can Prefect's map functionality work with dicts or named tuples? Thank you!

Hi, I am having a problem with prefect UI. Sometime the I change flow and register it or change flow label from UI. But the Metadata is not modified. I have to delete the flow and then register with a different name to make it work. I want to mention again, it is happening with some flows. Not a regular occurring. Also, if i delete the flow, it is kept in the archieve, if I register flow with same name the re-register does not work. I think metadata from archieved flows is messing up with the new flow Meta-data. Is there a way to delete flow from archieve. (permantly delete it from the database) @Kevin Kho posted in new Thread. Please refer it to your team.

Hello, I am trying to set up an ECS agent running on an ECS cluster. However, when I run a test flow, the ECS task definition is created but it appears that it is set to [INACTIVE] immediately, so the task doesn't actually get started (it shows as running in the AWS console, but with the [INACTIVE] flag) and my flow doesn't get executed. I don't see any logs other than

Submitted for execution

and eventually after three retries it marks the flow as failed

A Lazarus process attempted to reschedule this run 3 times without success. Marking as failed.

, on agent side, the only log i see is

Deploying flow run

. It appears that the agent is creating the task definition, running the task and then immediately de-registering it (as seen here), which could be the problem as I don't see how the task can run if the definition is immediately de-registered before the task run is complete. I have looked through this issue, but I don't think this is the same problem because I am using prefect cloud. Any help is much appreciated.

Hey is there any way to create an artifact of an image?

Hi all! I'm looking to leverage Prefect Parameters in order to sometimes manually enter run parameters if a pipeline needs to be re-run. I'm looking to clarify what exactly the difference between Context and Parameters is and how to use them correctly as I'm a little confused by the documentation.

Hello Prefect Family! I’m running a flow using mapped tasks, the LocalDaskExecutor and ECS/Fargate. There are 8 mapped tasks and when I run locally all 8 run in parallel. However, when I run in Prefect cloud, it runs 2 mapped tasks at a time until all 8 have completed. Since this works locally it’s not a restriction on ECS/Fargate. Any ideas on what this might be? (thinking out loud…) Is it possible that the “threads” default between my laptop (M1 Mac) and the Fargate nodes is different such that I need to default it in the flow?

We're seeing some weird behavior in Prefect Cloud with newly deployed flows. The UI keeps going back and forth saying that the flow is archived, to unarchived here is the flow: https://cloud.prefect.io/clearcover/flow/f3ee4527-3488-4d0a-8402-b998a974ed27

I am writing a Ruby Prefect client that’ll eventually allow our Rails app to schedule flows in Prefect. When exploring the GraphQL schema I am noticing that there are 3 uuid types defined. They are called:

uuid

,

_uuid

and

UUID

. Their definitions look identical, all are

SCALAR

values. What is the purpose of having 3 different UUID types defined in the GraphQL schema?

Hi everone. Are there any best practices for running flows locally that will ultimately run on Kubernetes? I was running them using the docker agent but I am starting to see a bunch of differences between the local environment and the remote kubernetes cluster. One of of these differences is how secrets are handled. The approach the devops person asked me to use is mounting the secrets as a volume in a custom job template that I pass to the

KubernetesRun

run config. I wrote a task that reads the secrets from the file system. However, locally, I was using environment variables for secrets. I'll probably have to see a way to mount the secrets locally on the docker container but before I go too far down this path, I'd like to get some input from the community. I suppose I could try to set up Kubernetes locally but I know very little about Kubernetes. This is my first experience with it.

Has anyone run into any problems reading a result written with a serializer other than the default pickle serializer? I set the result for a task to use GCSResult with the JSONSerializer. When I try to read the result from task runs obtained from

Client.get_flow_run_info(..)

, the read function is using the default pickle serializer. I used prefect 0.15.1 in both the flow environment and to run the read code. Am I missing something? Thank you in advance!

result = GCSResult(
    bucket="prefect-cloud-results",
    location="{flow_name}/{flow_run_id}/provenance.json",
    serializer=JSONSerializer()
)
@task(result=result)
def set_provenance_data(flow_run_id: str, prefect_cloud_client: prefect.client.client.Client = None):
...

Hello, I am using Docker storage to store my flows, but I'm on an ARM mac and our production environment is x86. From the command line I can build the docker image for the correct architecture using

docker buildx build --platform linux/amd64

. Is there any way to do this in the

prefect.storage.docker.Docker

class or can I build the image locally first then add flows to a pre-built image?

I have a flow that invokes a set of other flows, including the same flow twice with different parameters. I’m seeing an error consistently where only one of the invocations success and the other fails (and then disappears from the schematic). Error:

Failed to retrieve task state with error: ClientError([{'path': ['get_or_create_task_run_info'], 'message': 'Expected type UUID!, found ""; Could not parse UUID:

Hi prefect devs, is there a way to use a custom made Secrets provider (as explained in docs)  or the recent 

AWSSecretsManager

 in 

prefect.task

 library tasks? I don't figure out how to force 

slack_notifier

 task to use that Secret provider. for example this code throws an error because Slack try to load PrefectSecrets before Flow init:

Hello, I am trying to run a version of the bestpractices.py example from the documents on a windows server. Can anyone tell me why the shellcommand task never returns?

Hey folks, just wanted to that the README functionality in the UI is a huge boost to data scientists being able to do stuff self-service, and it would be really helpful to see the feature developed. I’m not a frontend person otherwise I’d help. Some ideas: • Better formatting. Right now, no offense, it’s a bit janky. Code snippets in particular. • I’d like to be able to set the documentation when registering the flow so that we can have the same flow deployed in staging and production as well as ensure that changes to code and docs are co-located. • I’d like to be able to setup examples of flow run parameters in the docs and have users of flows run the example directly from the docs.

Hey prefect devs, are default parameters (in this case dates) decided at the time a flow is registered or when the flow is executed following a daily schedule? I have a job that is registered to the prefect-server and takes in a date range wrt the date of execution. For example:

with Flow("my_daily_flow", schedule) as flow:
    start_date = Parameter("start_date", default=datetime.utcnow().date() - timedelta(days=5))
    end_date = Parameter("start_date", default=datetime.utcnow().date())
    daily_job.set_dependencies(keyword_tasks={"start_date": start_date, "end_date": end_date})

flow.register("test")

This question might have been asked before, but I would like some confirmation. Thank you!

Hi all, I'm struggling in finding where I can configure the format for the graphql server logs ( like the "INFO: ... - "POST /graphql/ HTTP/1.1" 200 OK" one ) , I need to change them to json format in order to have my infrastructure catch them.

Task notification in state_handlers is still near useless without task inputs. Just saying. Half my flow code is dealing with failed tasks. Even the example on the page on notifications is called sending-a-simple-notification. If you have mapped tasks posting Task failed on slack would be little use when you have 10000 of them.

Hello! I just upgraded to prefect 0.15 and started trying to use the kv store. My flow is giving this error:

field "key_value" not found in type: "query_root"

. But my query works in the interactive API console. (I'm using cloud prefect) Have folks seen this error before?

Hi! I'm new to prefect and am so keen to learn more! I have a use case that wants to have a flow that runs 2 different Docker images on k8s. Is there a good example I can follow?

Hi everyone, I've run into what looks like a bug in Prefect Cloud. No tasks are running, yet this State Message is being reported:

Queued due to concurrency limits. The local process will attempt to run the task for the next 10 minutes, after which time it will be made available to other agents.

That string does not appear in the open-source part of Prefect, so it must be part of Prefect Cloud. The concurrency limit on that task is 10, and things were working until I changed some of Dask's configuration parameters to try to resolve an issue with it. The most likely cause of the above message is that some error happened and it didn't get handled correctly. https://cloud.prefect.io/stockwell/flow-run/4460703d-3c91-4573-b85c-a4b001048999

Hey Prefect community, Wondering if anyone has a nice implementation or if prefect would potentially add this is a feature to the ECSRun config down the line. Looking to add Fargate spot constraints to my flow runs from my fargate service. What do you guys think? So far I haven't been able to find anything in the docs about it and assume I have to manually include it in the task def, which always seems to cause me issues. Cheers

extremely basic question, but i'm trying to get set up with prefect and am trying to figure out how to set up my secret auth keys to use boto3. In the Secrets documentation page it says to add

export PREFECT__CONTEXT__SECRETS__AWS_CREDENTIALS='{"ACCESS_KEY": "abcdef", "SECRET_ACCESS_KEY": "ghijklmn"}'

to the secrets context--could I do this in the prefect config toml file? That's where I was storing my other secret keys using prefect.config.api.XXX. This probably isn't the 'best' way to do this in production, but I'm used to using dotenv and os to just use a .env to pass in secret values.

Hey Team, are there any hard concurrency limits for prefect cloud standard membership? Eg. Could I schedule 50+ flows at a given minute schedule?

Dependency best practices question: I have an existing repo

feature-generator

which contains both worker/orchestration logic and the code for doing the work. I added task and flow definitions to it, but with github storage, the flow can't find the other modules in that repo (I've seen https://github.com/PrefectHQ/prefect/discussions/4776 and understand this is intentional). My question is how best to structure things so that my flow can use that repo's code, but also execute a parameterized run from

feature-generator

, on commit, through CI (because that's how we start the job right now). Obviously, I can make

feature-generator

a package and depend on it from a new

flows

repo, but to have

feature-generator

start the run would create a circular dependency. Would you split it into three repos, with one of them just being responsible for executing the flow? I don't love that idea, but maybe that's best practices?

Scaling question: forgive me if I missed this in the docs/github-issues (I have looked), but do flows executed with ECSRun auto-scale in some way? Ie, do mapped tasks turn into ECS tasks? Or does some ECS-service/task setting do the job based on resource consumption? Or do I need a Dask cluster for that? Does a Dask cluster even do this directly?

KubernetesSecret question: currently it seems that it's designed such that

KubernetesSecret

should not rely on any upstream task/parameters. However, I think it might not be a great assumption. I'm using an

env

parameter that denotes

"DEV"

or

"PROD"

and then would decide what secret I want to retrieve from k8s later (usually a database secret). Any plan to allow

KubernetesSecret

to be a dynamic task and interpreted naturally in the flow without having to manually use

flow.set_dependencies

?

Is there a way to use a

DictCursor

with the existing

SnowflakeQuery

task? The source shows

try:
            with conn:
                with conn.cursor() as cursor:
                    executed = cursor.execute(query, params=data).fetchall()
            conn.close()
            return executed

while what I would want to use is:

try:
            with conn:
                with conn.cursor(DictCursor) as cursor:
                    executed = cursor.execute(query, params=data).fetchall()
            conn.close()
            return executed

Has anyone else dealt with this before? I’d like to simply inject the DictCursor class if possible, but from the way this block of code is written it looks like the only way to do it is to subclass

SnowflakeQuery

and overwrite the entire

run

method.