Hello We've been using prefect internally for about two years now. Congrats on the cadence of innovation. I'ts been a pleasure using your tools. At this stage we have two environments setup, one for ETL runs and another for internal reporting mechanisms. We are on the standard plan and would like to expand the number of users. I've tried twice now to get access to the sales team via the main prefect portal. Haven't heard anything from you guys in over two weeks, Can someone please assist me in having the relevant conversation ?

I need a script that restart a lot of flow run. I'm able to get the flow runs, however I'm not sure if the mutation to restart them. We had about 600 failed run because of infra that I need to restart

Hi! I am getting below error sometimes when i try to update a Prefect secret, i suspect multiple flows updating same secret at same time causing this issue, still i am not sure. Can someone help.

prefect.exceptions.ClientError: [{'path': ['secret_value'], 'message': 'An unknown error occurred.', 'extensions': {'code': 'INTERNAL_SERVER_ERROR'}}]

Hi all, I am using prefect 2.0 more and more these days and loving it! Is there a way to quickly see execution's time of a flow run from the UIs listings? I guess I could filter runs, but it would be nice to have a faster way to select a run based on its execution time. It seems that the visualization has enough space available in the runs' rows and maybe timestamp could easily fit on it:

Hi guys, I'm trying to pass from a flow of 6 task to 6 flow of 1 task with daskExecutor

Hi I'm currently trying to do some deployment testing, and I'm getting the following error: Specification in 'newflow.py', line 13 failed validation! You have not configured default storage on the server or set a storage to use for this deployment but this deployment is using a Universal flow runner which requires remote storage. I've currently got a file called: from prefect import flow @flow def hello_world(name="world"): print(f"Hello {name}!") # Note: a deployed flow does not need a command to # explicitly run the flow. The API handles this for you. # hello_world() from prefect.deployments import DeploymentSpec DeploymentSpec( flow=hello_world, name="hello-world-daily", ) and the error comes out when I run the following: prefect deployment create newflow.py

After upgrading Prefect I get this warning on tasks with retry settings:

UserWarning: Task <Task: task_name> has retry settings but some upstream dependencies do not have result types. Cloud features such as input caching and resuming task runs from failure may not work properly.

The (automatic) retrying seems to be working fine, not sure what this warning refers to. I am aware that I cannot resume task run from failure, but not sure why that is related to retry settings

I was wondering how can one access/modify Prefect Context variables at Task/Flow run on Prefect 2.0 (similar to how it was done in Porter 1 -> https://docs.prefect.io/core/concepts/execution.html#prefect-supplied-context)

I am getting a very wired error on Prefect Orion 2.0b3 when I am attempting to run a Prefect Shell task Here's how my flow is structured - Scenario 1 - Primary Flow: Flow is not a async flow. Uses sequential task runner, a bunch of processing tasks. The final task is to build a CLI and trigger a shell task that runs a particular test, makes few API calls, etc.

@flow(name='primary_flow', version='1', task_runner=SequentialTaskRunner())
def primary_flow(**kwargs):
    ... # perform data processing tasks, all tasks are non-async
    for cli_command in parsed.result():
        shell_run_command(cli_command) # I've also tried using asyncio.run for running the shell task

Scenario 2 - Primary Flow: Uses sequential task runner, a bunch of processing tasks. The final task to build and call a CLI is done through another async subflow

@flow(name='primary_flow', version='1', task_runner=SequentialTaskRunner())
def primary_flow(**kwargs):
    ... # perform data processing tasks, all tasks are non-async
    for _ in parsed.result():
        cli_flow(_) # I've also tried using asyncio.run for running the shell task

@flow(name='cli_flow', version='1', task_runner=SequentialTaskRunner())
async def cli_flow(parsed):
    for _ in parsed:
        ... # Build CLI
        await shell_run_command(cli_command)

I am getting this exception thrown (attached screenshot) and the CLI task is not getting started/executed. Can someone help me out on this one please (I am terrible with asnyc processes)?

Hi all, I’m having trouble diagnosing a GitHub storage problem. I’ve created a trivial testing flow similar to the example script-based workflow for GitHub. I’ve deployed a k8s agent using

prefect k8s agent install

. I’ve uploaded my flow to a private GitHub repo and registered it with Prefect. And I’ve added a Cloud Secret called

GITHUB_ACCESS_TOKEN

that holds a valid GitHub personal access token. When I run my flow the agent’s GitHub storage gives me an

UnknownObjectException(404, 'Not Found')

error. If I change the flow to use a different Cloud Secret key for the PAT, such as

access_token_secret='MYKEY'

, then the agent tells me

ValueError('Local Secret "MYKEY" was not found.')

. How can I introspect the kubernetes agent to verify that the GitHub PAT secret is being loaded from Prefect Cloud correctly?

👋 hullo folks! have a general question about flows vs. tasks vs. just regular functions in prefect 2.0. are there any high level guidelines on where we may want to use each? i'll try to describe a scenario in the thread

Hi, i have a quick question about referencing a past flow run in the current-flow. Lets say i'm processing files in one-run of the flow, the next time i'm executing (the flow) it i want to only process the new filesnot yet processed, so i could process only those that were created after the last successful run (maybe add some overlap just in case) Pretty sure doing it times based is not the best solution but it would work for my case. For simplicity i don't want to keep a state of already processed files. What would be a common solution to this problem?

Hi All, General Question - Can we deploy prefect on open shift ? Is there a doc for this ? Appreciate your help!

Hi all! I'm looking for some guidance on handling logging of context attributes. To best organize our events/metrics within out internal systems, I'm looking to prefix all Prefect context attributes with

prefect.

, e.g.

prefect.flow_name

. Given the documentations cautioning of renaming context attributes, I'm trying to brainstorm the simplest and most effective way of doing this across all our flows. Do you have suggestions?

i ran a flow of flows process that was supposed to kick off 12,000 flows on GKE. it started 10,000 flows, and then failed with the message

Pod prefect-job-5e3af599-tl2xs failed. No container statuses found for pod

where can i look for a more detailed message? any idea what actually caused it to fail? also, of those jobs, all but one passed. the one that failed had the same message (no container statuses found).

Okay, prefect is officially awesome. I just launched the server locally and realized i didn’t see the whole picture from the docs. Thanks for building such an amazing product.

Every few days, some of our DockerRun scripts fail with a weird 500 Internal Server Error: ("error while mounting volume '/var/lib/docker/volumes/sraprod67-output/_data': failed to mount local volume: mount but most flow runs are successful, with no mounting issues. Has anyone else experienced this?

Hi Prefect Experts - Im looking for a little guidance on using the

ExecuteNotebook

Task.

im looking to run a flow on a schedule of every day except for every other monday (on those days i want to run a different flow). what is the best way to do that via a schedule?

I’m looking to set up an automated way to prune old/test flows in a project. I found the GraphQL endpoint to remove flows, but to determine old flows vs current flows I was thinking to: 1. Pipe the output of

prefect register -p project

to get the names of the current flows that would have been registered 2. Get the IDs of the flows through the GraphQL API 3. Delete flow IDs that that do not have names from step 1 Is there a more sensible way to get the “current flows” in a project beside the

prefect register

trick?

Hrm, looks like the Client is dying shortly after launching the Cluster? It runs a few tasks, then right before mapping a big-ish one, the Client dies (cloudwatch logs just say

Killed

) and the mapping stays in a

mapped

state

Hey Prefect team, Is there appetite for adding a delete to the CLI deployments command for orion? Recently ran into a case where I need to delete some deployments, but didn't find it available via CLI It looks like the change would require two changes: Add Api method for delete_deployemnt https://github.com/PrefectHQ/prefect/blob/d591e730ee160878eb21755464a234cef1f16b47/src/prefect/client.py#L168 Add Delete command to cli calling above delete_deployment: https://github.com/PrefectHQ/prefect/blob/d591e730ee160878eb21755464a234cef1f16b47/src/prefect/cli/deployment.py

Debugging Prefect Flows with breakpoints. Hello friends. I love coding Python in VSCode, the debugging with breakpoints make me so efficient. But the debugging fun stops with Prefect Flows, when we initialize them from CLI:

$ prefect run --path myflow.py --log-level INFO --param myparam=val

Here the development flow becomes rather blackboxed/trial-and-error. Is it possible to somehow set breakpoints in VSCode, start a prefect flow from CLI, and attach the debugger to the python process? (Running with the LocalExecutor)

Hi all, I’m trying to set a scheduled dependency between flows so that flow B starts when flow A finishes (after being scheduled by time). I found this: https://docs.prefect.io/core/idioms/flow-to-flow.html but it’s not very clear to me how I can implement it.

Hey Guys I want to setup a flow dependency for my task, basically I am running a flow, inside which I have a flow which downloads some data and then after this data has been downloaded I want to run some tasks, but when I try to do so the download flow runs successfully but the tasks after that end up failing, any idea why could that be happening and how can I fix it?

Hi, I'm looking through https://github.com/PrefectHQ/prefect/discussions/4042 to define the most optimal deployment approach. We have a single Agent acting more as an orchestrator and 1000 Dataflows that are using DockerRun() and S3 Storage. In order to set up CI\CD we have to : 1. Copy the whole project to the Agent host and register all flows by means of some custom Python scripts (which we have to write). Dataflows will be saved on S3 at this moment. 2. Ensure that the image used in DockerRun() has all requirements/env variables installed and set. 3. Execute DataFlow from UI, which will go to S3 to pick flow, run new Docker container from the predefined image, and execute flow on that image. I'm bit confused here that copy of all DataFlows will sit on Agent and in fact Agent is not a worker. Can this approach be simplified?

I’ve have a scheduled flow that is failing due to a credential issue:

Error downloading Flow from S3: An error occurred (ExpiredToken) when calling the GetObject operation: The provided token has expired.

I’m using AWS SSO credentials to upload the flow to S3:

flow.storage = S3(
    bucket=Config.PREFECT_S3_BUCKET,
    client_options={
        "aws_access_key_id": Config.AWS_ACCESS_KEY_ID,
        "aws_secret_access_key": Config.AWS_SECRET_ACCESS_KEY,
        "aws_session_token": Config.AWS_SESSION_TOKEN,
    },
)

Initially, the flow runs. However, flows scheduled after the token expiry fail with the above error. How do I both use SSO to upload flows to S3 and have them scheduled on Prefect Cloud?

Hi, Do you have example of building flow to AWS S3 with credentials like aws token? The example in doc does not have credentials

from prefect import Flow
from prefect.storage import S3

flow = Flow("s3-flow", storage=S3(bucket="<my-bucket>"))

Hi I am new to Prefect but over last few days I have managed to create a few flows and executed them from Prefect Cloud and Local Agents. Now I have been playing around with using docker containers for the agents and scripts are on the containers. That is working when the container is running locally. So next I decided to try to build a docker and run it with GCP Cloud Run and CI/CD pipeline with Github Actions. That builds but failes in deployment and I get this error: Cloud Run error: The user-provided container failed to start and listen on the port defined provided by the PORT=8080 environment variable. Logs for this revision might contain more information. Logs URL: https://console.cloud.google.com/logs/viewer?project=dkpos-dev&resource=cloud_run_revision/service_name/dkpos-dev/revision_name/dkpos-dev-00002-dam&advancedFilter=resource.type%3D%22cloud_run_revision%22%0Aresource.labels.service_name%3D%22dkpos-dev%22%0Aresource.labels.revision_name%3D%22dkpos-dev-00002-dam%22 For more troubleshooting guidance, see https://cloud.google.com/run/docs/troubleshooting#container-failed-to-start Has anyone managed to get this running in GCP Cloud run? Any hints are most welcomed and I have no idea how to solve this port issue. I am of course open for other ways to deploy and run my projects in containers. Basically looking for the simplest solution if possible 😄 Thanks in advance, Birkir

When using classes with @tasks, I'm assuming that the class needs to be instantiated each time for each task? For example:

@task
def task_a():
    common_class = Common(Secret())
    output = common_class.task_a()
    return output

@task
def task_b():
    common_class = Common(Secret())
    output = common_class.task_b()
    return output