Hello all! I would like to set up a flow that pulls a GH repo of many other flows to execute them, similar to the dbt prefect flow of flows (minus the dbt) that @Anna Geller authored so well. In the repo, some flows should be executed daily, while others hourly. Does anyone have best practices on achieving this such that the parent flow runs on the hourly cadence and only picks up the hourly flows, while the daily ones get executed the one time? Further, for all child flows that are successful, how do they act as a potential dependency to trigger another flow, say a dbt task, without needing the whole flow to be successful?

Am seeing this exception while trying to run a simple flow:

File "/opt/conda/envs/coiled/lib/python3.9/site-packages/prefect/client/base.py", line 125, in raise_for_status
Exception: PrefectHTTPStatusError("Client error '404 Not Found' for url '<http://ephemeral-orion/api/task_runs/add94276-cb3e-417d-a3f9-033f935ae5ec/set_state>'\nResponse: {'exception_message': 'Task run with id add94276-cb3e-417d-a3f9-033f935ae5ec not found'}\nFor more information check: <https://httpstatuses.com/404>")

Any ideas as to what might cause this?

question: For using the default infastructure 'Process' how do i add environment variables to that in deployment?

Is there an issue with service accounts? The button on the ui is grayed out and inaccessible to create new SAs

I'm submitting a flow to Prefect Server for execution using

LocalRun

. I have a single Prefect agent running in a container. I use

IntervalSchedule

and set

interval=timedelta(seconds=60)

. I expected to see my flow run once every minute indefinitely, but instead, I see 10 flows scheduled on the Dashboard. I tried other schedulers -- Cron, etc -- but it always schedules 10 and only 10 runs. Am I missing something or misunderstanding how schedulers work? Thanks

hey all. Question about the

INTERNAL_SERVER_ERROR

and prefect's secret client throwing the

KeyError: 'SLACK_WEBHOOK_URL'

. I know there is a current intermittent issue in regards to retrieving secrets. We pass our secrets to the dask worker via the docker container so we aren't using the prefect secrets stored in the UI. However we are still getting this error and it has brought down multiple flows this morning. The error, other information and questions will be in the threads. Thank you in advance!

Hi folks - wondering if anyone can explain what exactly goes into storage? I'm running something relatively small-scale locally and my /.prefect/storage folder is starting to get quite bloated. I'm not running this through a deployment, so my understanding is that this isn't my whole codebase being stored as per the docs on storage. It seems implied (and would align with the files that I'm seeing) that individual task/flow result objects are getting saved here, but nothing in the prefect 2.0 docs actually say that explicitly - is that what's going on, and is there a way to manage them other than deleting/moving them manually?

Hi there! I was struggling to deploy a simple flow from code using

KubernetesJob

. However, just before posting a question in this channel, I figured I should "give it one last try" and to my surprise it actually worked 🤡 As the post is already written, I figured I might as well share was I experienced and the solutions so that anyone else having simiular issues may have somewhere it start. I was trying to create a deployment using

KubernetesJob

as infrastructure. The deployment looked like this (see code further down in the thread) The container image is built from a dockerfile based on

prefecthq/prefect:2-latest

and copies the script containing the flow to

/opt/prefect/flows/flow.py

. This deploys just fine, and I can see that the kubernetes job is created and the Prefect engine starts, however it does not seem like the flow runst, and I instead get the following error (see traceback further down in the thread) It turns out that the base docker-image I had chosen was the issue, as the prefect version did not match the version the server was running (

2.4.5

vs

2.4.2

). From the release-notes for

2.4.3

-

2.4.5

i see that block protection is an issue, and I speculate that this might be the root cause. Hope it helps 🙂 Edit: Moved code and traceback to the thread

Is there an expectation or best practice for where flows and supporting files are supposed to be to enable deploying?

Hi, I have been struggling to extract the logs for flow runs from prefect cloud using the api. In prefect 1 I was able to do this with the graph ql query below. How can I do this in prefect 2?

Hello, I am experiencing some strange behavior with Exceptions and I can't find anything in discourse about it. Exceptions being thrown from flow code are caught, but exceptions being thrown from libraries imported in the virtual environment are not being caught. Details in thread ...

Hello. Is there any best practice for handling large task results? I’ve got a task that can return >5GB of data as a result, and I assume that’s what’s causing the flow to slowdown creating and running subsequent tasks.

When will paused states be supported in Prefect 2.0? Currently this first class functionality is missing, which impacts our ability to migrate from 1.0

ents/Developer/Library/Frameworks/Python3.framework/Versions/3.9/Headers -arch arm64 -arch x86_64 -Werror=implicit-function-declaration -I/Users/navinchaganti/codebase/prefect-dev/prefect_env/include -I/Applications/Xcode.app/Contents/Developer/Library/Frameworks/Python3.framework/Versions/3.9/Headers -c asyncpg/pgproto/pgproto.c -o build/temp.macosx-10.9-universal2-cpython-39/asyncpg/pgproto/pgproto.o -O2 -fsigned-char -Wall -Wsign-compare -Wconversion
  asyncpg/pgproto/pgproto.c:31:10: fatal error: 'Python.h' file not found
  #include "Python.h"
           ^~~~~~~~~~
  1 error generated.
  error: command '/usr/bin/clang' failed with exit code 1
  ----------------------------------------
  ERROR: Failed building wheel for asyncpg
Failed to build asyncpg
ERROR: Could not build wheels for asyncpg which use PEP 517 and cannot be installed directly

Anyone faced the asyncpg error while installation ?

Hello, I have some absolutely baffling behavior trying to implement state handlers using a

fail_signal = signals.FAIL(message_here); raise fail_signal

-> the state handler isn’t triggering, although it does trigger when a task otherwise changes status AND it triggers if a task fails for a reason other than a

raise fail_signal

. What even more baffling is that I can run the sample code with

mission_critical_task

from here and I get the expected behavior and the expected triggers. It’s a complicated flow, but I’ve ripped out as many pieces and simplified as much as possible and now I’m out of ideas on why I’m getting this lack of behavior. I’m tied to prefect 0.15.13

Hi everybody. When attempting to orchestrate dbt transformations via trigger_dbt_cli_command using an existing profile

~\.dbt\profiles.yml

, I receive the following runtime error and I'm having difficulty interpreting the issue,

exit code 4294967295

. Has anyone else countered this? Error:

"..~\venv\lib\site-packages\prefect_shell\commands.py", line 90, in shell_run_command raise RuntimeError(msg)

RuntimeError: Command failed with exit code 4294967295:

Script:

from prefect import flow
from prefect_dbt.cli.commands import trigger_dbt_cli_command

@flow
def trigger_dbt_cli_command_flow():
    result = trigger_dbt_cli_command("dbt debug")
    return result

trigger_dbt_cli_command_flow()

Versions: Windows 10 Prefect: 2.4.5 Prefect-dbt: 0.2.2 Thanks very much!

@Anna Geller I think I’m somewhat confuse. I’m using

awswrangler

package to wrangle data and move same to an S3 bucket. Before signing on prefect cloud I could use

wr

by simply indicating an s3 bucket path and my data would go to this path. Recently I created an S3 block via prefect cli. Now in an attempt to build deployment with the

-sb

flag I got an error that I indicated wrong s3 path. Clarifying questions: 1. Must I use the prefect cloud s3 storage block as the

path

under

wr

package? 2. Can I have a separate s3 bucket for storing output of my data wrangling different from the s3 bucket where my flow code should reside? I would appreciate if your answer include some basic scenarios. I have gone through all your repo and none appear to address this issue probably there is, maybe implicitly .

Failed to load and execute flow run: ClientError([{'path': ['secret_value'], 'message': 'Unable to complete operation. An internal API error occurred.', 'extensions': {'code': 'API_ERROR'}}])

facing this issue multiple times? Any solutions?

Hi. I'm curious about testing

pypy 3.9

since it has JIT. I'm trying to install prefect as the minimal dependency but there are some problems that will probably be outside of the 'supported things' -- prefect has

orjson

as a dependency which is not pypy-supported. I wonder how big of a role that dependency plays in prefect and whether there's some way to hack a bit to make it use some other lib?

Hey, community! How do I set up task caching for k8s jobs? not really clear what setting

PREFECT_LOCAL_STORAGE_PATH

does in this context, I guess points to pod’s filesystem and persistent storage/bucket has to be mounted to a path

in prefect 0.15.13 is there a good way to check whether a task and its output is serializable? I know you can use

is_serializable

to check if entire flows can be serialized but that doesnt allow you to narrow down which tasks are responsible for your serialization issues. what is the best way to debug in this situation?

A question on how to use async tasks: Context

class Result:
  pass

@task
async def my_task_1(arg: str) -> Result:
  ...

@task
async def my_task_2(arg: Result) -> None:
  ...

Question Doc says I can do

await my_task_1('arg')

in a flow, although mypy says

my_task_1('arg')

is not awaitable Is following a best practice?

task_result = await my_task.submit('arg')
result = await task_result.result()

await my_task_2.submit(result)

I am suddenly finding that mapped tasks are not reflecting that the child tasks have completed. I'm not positive that it's not because I introduced some kind of bug while developing, but I've done my best to validate that it's not something I did. Has anyone else seen this? Is something just going wonky with the cloud service at the moment? For example, the flow in the image has finished successfully, with all child tasks having completed. But it still shows 0/13.

I seem to be running into an issue where sub 5 minute flows aren't being scheduled. If I re-register the flow it'll schedule the next 10 runs but once they have been completed now more are scheduled. I've tried using cron clocks and interval clocks and manually creating a schedule in the UI but keep seeing the same result. I'm using Prefect Cloud 1.0 I have two agents both deployed on AWS ECS.

Anyone notices issues with Prefect Cloud today? Our flows show all tasks as "pending", but we can see it logs that they get executed. This happens for all the flows, there wasn't any updates or deployments at our organization since quite a few days. Yesterday all was working fine, quite a few flows completed successfully. The issue seems to be happening since a few hours 🤔

Hello Prefect team, Since yesterday we're experiencing many issues: • restarting flow/task doesn't work • tasks with manual approval don't start after being approved • manually changing task state doesn't work

Hi all, I am trying to figure out whether there is an option for running orion server in url subpath like my-domain.com/prefect. Is there any documentation for this case?

Hi - I am seeing lots of these errors over last few days;

Exception raised while calling state handlers: ClientError([{'path': ['secret_value'], 'message': 'Unable to complete operation. An internal API error occurred.', 'extensions': {'code': 'API_ERROR'}}])
Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/prefect/client/secrets.py", line 140, in get
    value = secrets[self.name]
KeyError: 'SLACK_WEBHOOK_URL'

Hi there, I have a question to the Prefect team on the hybrid model but also want to raise awareness on a potential security/data protection issue for other users. According to the Prefect website, code and data should never leave a customers infrastructure. The "exception" here are of course the log entries which can contain any kind of data. In Prefect 1, this was ok for us as we sanitized our own log entries and the logs generated by Prefect itself did not seem to contain any sensitive data. However, in Prefect 2, the log entries generated by Prefect may contain all arguments passed to a function! So far I only found the example below where it happens when the passed argument names do not match what the task expects. However, are there more of these cases were data can leak into the logs? Is this really the intended behavior?

from prefect import task, flow

@task
def task1(arg1):
    pass

@flow
def main_flow():
    task1(arg1_wrong=1)

main_flow()

Logs on Prefect Cloud: