Hi there! I’m looking at transitioning to Prefect 2.0 and trying to wrap my head around some of the new stuff. In my existing workflow, I construct a flow programmatically by walking a set of yaml files. This has the effect of needing the

add_task

method from the

flow

class. I don’t think the new API has this. Is there a way to iteratively add tasks to a flow?

hi everyone! Probably this is straightforward, I'm trying to define a waiter using prefect_aws.client_waiter but I'm getting: Missing required parameter in input: "jobs" Any help would be great!! Traceback in the thread.

Hello, How can I delete a single run of a flow from the history in prefect 1.4 ? Thanks in advance

Hi, Do all tasks within an asynchronous flow need to be asynchronous, or can synchronous tasks be run within an asynchronous flow? I've attached a minimal example and the output from the Prefect UI. The asynchronous task runs successfully, but the synchronous task seems to hang indefinitely. Any help is greatly appreciated!

from prefect import flow, get_run_logger, task


@task
async def async_task():
    return "async_task"


@task
def synchronous_task():
    return "synchronous_task"


@flow
async def my_flow():
    logger = get_run_logger()
    <http://logger.info|logger.info>("Executing async task:")
    async_result = await async_task()
    <http://logger.info|logger.info>(f"Result of async task was {async_result}")
    <http://logger.info|logger.info>("Executing synchronous task")
    synchronous_result = synchronous_task()
    <http://logger.info|logger.info>(f"Result of synchronous task was {synchronous_result}")

Hello , I am trying to set up a prefect Orion server. I am having trouble creating a work queue. I had the error :

Error occured when trying to create new work queue

in a 405 response for a request to

<http://xx.xx.xx.xx:4200/xx.xx.xx.xx:4200/api/work_queues/>

We use this Dockerfile to run the server:

FROM python:3.9.16

RUN pip install prefect prefect[aws]
RUN prefect config set PREFECT_API_URL=http:/xx.xx.xx.xx:4200/api
CMD ["prefect", "orion", "start", "--host", "0.0.0.0"]

Thanks a lot for your help !

Hello, Would it be possible to read all files in a s3 sub-folder ?

Hi folks - I've just written a fairly long blog post introducing how to use Prefect with a real-world example (part of a series I'm creating). Would anybody that's familiar with Prefect do me a solid and proof-read it for me so I can make sure everything I've said is factual? I would be forever in your debt...

https://source.coveo.com/2023/01/04/creating-custom-sentry-block-prefect-2/ Here's my latest blog post on how to integrate Sentry with Prefect 2 using a custom block. Enjoy

Hi, With result persistence on, is it possible to trigger a task-rerun if the persisted result got deleted or corrupted? I am thinking about a case where the cache-key for the orion-DB is available, but the result can't be retrieved from local storage. In such a case I would like to invalidate the cache-key and trigger a new task-run to obtain the missing/corrupted result.

Hi all!

Hi all! Does anyone know how to get the Prefect CLI integrated with shell to allow for autocomplete?

Prefect 1.0 If have a "flow of flows", which calls

wait_for_flow_run

several times. I want this parent flow's success to depend on the success of all child flows. I've tried setting

upstream_tasks

to the result of

wait_for_flow_run

, but I realize that that's a

FlowRunView

object, not a

Task

. From the child flow, I can

get_tasks

, and set

upstream_tasks

to that. But I have some tasks I allow to fail in the child flows (they are not reference tasks of the child flow.) What I'd really like to do is to be able to reference

<child_flow>.state

, and set the parent flow's reference tasks to be expecting a

<Success: "All reference tasks succeeded.">

Is there a built-in way of going about this, instead of just manually inspecting the child flow results, and raising the appropriate SIGNAL ?

I opened a thread here: https://discourse.prefect.io/t/prefect-2-7-5-2-7-6-flow-stuck-at-pending-forever/2149

Prefect 2.0. Is there a way to get the email address of whomever started the run? I tried using FlowRunContext.get().flow_run.created_by.display_value which looks like an email address but is missing the @ and periods.

Hey team! I am new to prefect and I using: • EC2 (ubuntu) • S3 storage block • I am build a flow that takes data from mixpanel and dumps into an s3 bucket. I get the following error (trace in thread)

Flow could not be retrieved from deployment.

from boto3 when I try and run the deployment into the ec2 instance. I can deploy to s3 locally and connect to bucket from the ec2 using the AWS cli. Please can anyone help me debug this?

Prefect 2.0 Does Prefect 2.0 support artifacts like 1.0 does? I want to attach some text or a link the a flow as a result. Is there a way to do this?

Anyone able to assist me one that:

Anyone able to assist me on this: after I updated to 2.7.6, all my scheduled flow is not able to run by the local agent.....they will show as Pending forever. I've posted my issue here as well: https://discourse.prefect.io/t/prefect-2-7-5-2-7-6-flow-stuck-at-pending-forever/2149

Hi, anyone has a link to documentation on best-practices regarding passing secrets as Task Input in Prefect 2 ? I would ideally like that my secrets never get sent in clear text to Prefect Cloud also.

Is it possible to access the flow name from a task run context? I can use

get_run_context().flow.name

from the flow context but am unable to find an equivalent on the task level

One more annoying question about Prefect Helm chart. this https://github.com/PrefectHQ/prefect-helm/blob/main/charts/prefect-orion/values.yaml#L233 is for an admin user capable of creating a new database within the server. Is it also possible to specify username and password for the “prefect” user?

the notification setting on 2.7.6 seems off....should there be 'what to sent the notification' there?

Hi, when I used aws credential block to access S3 via boto3, I have this error

boto3.exceptions.S3UploadFailedError: Failed to upload. An error occurred (SignatureDoesNotMatch) when calling the PutObject operation: The request signature we calculated does not match the signature you provided. Check your key and signing method.

Kubernetes Autopilot GCP is killing my pods for scale down nodes 😞 I have the

safe-to-evict: false

on my template file, but it is not working

<http://cluster-autoscaler.kubernetes.io/safe-to-evict|cluster-autoscaler.kubernetes.io/safe-to-evict>: "false"

Does anyone knows how to prevent k8s scale down while a job is still running ?

Can anyone help me with understanding the concurrency concept in prefect? I have created a sample flow to understand the concurrency at both queue level and task(tag) level. I used the below script

from prefect import task, flow
from time import sleep
from prefect.task_runners import ConcurrentTaskRunner

@task(tags=['concurrent-task'])
def perf_task(input):
    sleep(25)
    print(input)

@flow(task_runner=ConcurrentTaskRunner)
def test_flow():
    for i in range(100):
        perf_task(i)


if __name__ == '__main__':
    test_flow()

I have created a deployment with queue

concurrent_test

and set the concurrency limit to 3. In my understanding this is responsible for the number of flow runs which can be executed concurrently. after that I have created a concurrency limit for the tag i have included in the task.

prefect concurrency-limit create concurrent-task 5

My understanding of this setting is for number of tasks in which could be executed concurrently. Once both the settings are done, I have started an agent with

prefect agent start -q concurrent_test

When I initiate the flow run, below is the execution log, in which it is very clear there is no concurrency (check the time stamp for every task execution)

Starting v2.7.6 agent with ephemeral API...

  ___ ___ ___ ___ ___ ___ _____     _   ___ ___ _  _ _____
 | _ \ _ \ __| __| __/ __|_   _|   /_\ / __| __| \| |_   _|
 |  _/   / _|| _|| _| (__  | |    / _ \ (_ | _|| .` | | |
 |_| |_|_\___|_| |___\___| |_|   /_/ \_\___|___|_|\_| |_|


Agent started! Looking for work from queue(s): concurrent_test...
07:54:51.280 | INFO    | prefect.agent - Submitting flow run '9034dd75-b32e-4b39-8f48-eb03c55e3fe5'
07:54:51.336 | INFO    | prefect.infrastructure.process - Opening process 'strange-starling'...
07:54:51.348 | INFO    | prefect.agent - Completed submission of flow run '9034dd75-b32e-4b39-8f48-eb03c55e3fe5'
/opt/homebrew/Cellar/python@3.10/3.10.9/Frameworks/Python.framework/Versions/3.10/lib/python3.10/runpy.py:126: RuntimeWarning: 'prefect.engine' found in sys.modules after import of package 'prefect', but prior to execution of 'prefect.engine'; this may result in unpredictable behaviour
  warn(RuntimeWarning(msg))
07:54:55.882 | INFO    | Flow run 'strange-starling' - Created task run 'perf_task-c7bf4036-0' for task 'perf_task'
07:54:55.883 | INFO    | Flow run 'strange-starling' - Executing 'perf_task-c7bf4036-0' immediately...
07:54:55.909 | INFO    | Task run 'perf_task-c7bf4036-0' - Finished in state Completed()
07:54:55.921 | INFO    | Flow run 'strange-starling' - Created task run 'perf_task-c7bf4036-1' for task 'perf_task'
07:54:55.921 | INFO    | Flow run 'strange-starling' - Executing 'perf_task-c7bf4036-1' immediately...
07:54:55.945 | INFO    | Task run 'perf_task-c7bf4036-1' - Finished in state Completed()
07:54:55.955 | INFO    | Flow run 'strange-starling' - Created task run 'perf_task-c7bf4036-2' for task 'perf_task'
07:54:55.955 | INFO    | Flow run 'strange-starling' - Executing 'perf_task-c7bf4036-2' immediately...
07:54:55.979 | INFO    | Task run 'perf_task-c7bf4036-2' - Finished in state Completed()
07:54:55.989 | INFO    | Flow run 'strange-starling' - Created task run 'perf_task-c7bf4036-3' for task 'perf_task'
07:54:55.990 | INFO    | Flow run 'strange-starling' - Executing 'perf_task-c7bf4036-3' immediately...
07:54:56.012 | INFO    | Task run 'perf_task-c7bf4036-3' - Finished in state Completed()
07:54:56.022 | INFO    | Flow run 'strange-starling' - Created task run 'perf_task-c7bf4036-4' for task 'perf_task'
07:54:56.022 | INFO    | Flow run 'strange-starling' - Executing 'perf_task-c7bf4036-4' immediately...
07:54:56.045 | INFO    | Task run 'perf_task-c7bf4036-4' - Finished in state Completed()
07:54:56.054 | INFO    | Flow run 'strange-starling' - Created task run 'perf_task-c7bf4036-5' for task 'perf_task'
07:54:56.054 | INFO    | Flow run 'strange-starling' - Executing 'perf_task-c7bf4036-5' immediately...
07:54:56.077 | INFO    | Task run 'perf_task-c7bf4036-5' - Finished in state Completed()
07:54:56.086 | INFO    | Flow run 'strange-starling' - Created task run 'perf_task-c7bf4036-6' for task 'perf_task'
07:54:56.087 | INFO    | Flow run 'strange-starling' - Executing 'perf_task-c7bf4036-6' immediately...
07:54:56.109 | INFO    | Task run 'perf_task-c7bf4036-6' - Finished in state Completed()
07:54:56.120 | INFO    | Flow run 'strange-starling' - Created task run 'perf_task-c7bf4036-7' for task 'perf_task'
07:54:56.121 | INFO    | Flow run 'strange-starling' - Executing 'perf_task-c7bf4036-7' immediately...
07:54:56.144 | INFO    | Task run 'perf_task-c7bf4036-7' - Finished in state Completed()
07:54:56.153 | INFO    | Flow run 'strange-starling' - Created task run 'perf_task-c7bf4036-8' for task 'perf_task'
07:54:56.153 | INFO    | Flow run 'strange-starling' - Executing 'perf_task-c7bf4036-8' immediately...
07:54:56.177 | INFO    | Task run 'perf_task-c7bf4036-8' - Finished in state Completed()
07:54:56.187 | INFO    | Flow run 'strange-starling' - Created task run 'perf_task-c7bf4036-9' for task 'perf_task'
07:54:56.188 | INFO    | Flow run 'strange-starling' - Executing 'perf_task-c7bf4036-9' immediately...
07:54:56.210 | INFO    | Task run 'perf_task-c7bf4036-9' - Finished in state Completed()
07:54:56.221 | INFO    | Flow run 'strange-starling' - Created task run 'perf_task-c7bf4036-10' for task 'perf_task'
07:54:56.222 | INFO    | Flow run 'strange-starling' - Executing 'perf_task-c7bf4036-10' immediately...
07:54:56.244 | INFO    | Task run 'perf_task-c7bf4036-10' - Finished in state Completed()
07:54:56.253 | INFO    | Flow run 'strange-starling' - Created task run 'perf_task-c7bf4036-11' for task 'perf_task'
07:54:56.253 | INFO    | Flow run 'strange-starling' - Executing 'perf_task-c7bf4036-11' immediately...
07:54:56.278 | INFO    | Task run 'perf_task-c7bf4036-11' - Finished in state Completed()
07:54:56.289 | INFO    | Flow run 'strange-starling' - Created task run 'perf_task-c7bf4036-12' for task 'perf_task'
07:54:56.289 | INFO    | Flow run 'strange-starling' - Executing 'perf_task-c7bf4036-12' immediately...
07:54:56.311 | INFO    | Task run 'perf_task-c7bf4036-12' - Finished in state Completed()
07:54:56.323 | INFO    | Flow run 'strange-starling' - Created task run 'perf_task-c7bf4036-13' for task 'perf_task'
07:54:56.323 | INFO    | Flow run 'strange-starling' - Executing 'perf_task-c7bf4036-13' immediately...
07:54:56.345 | INFO    | Task run 'perf_task-c7bf4036-13' - Finished in state Completed()
07:54:56.355 | INFO    | Flow run 'strange-starling' - Created task run 'perf_task-c7bf4036-14' for task 'perf_task'
07:54:56.355 | INFO    | Flow run 'strange-starling' - Executing 'perf_task-c7bf4036-14' immediately...
07:54:56.380 | INFO    | Task run 'perf_task-c7bf4036-14' - Finished in state Completed()
07:54:56.389 | INFO    | Flow run 'strange-starling' - Created task run 'perf_task-c7bf4036-15' for task 'perf_task'
07:54:56.389 | INFO    | Flow run 'strange-starling' - Executing 'perf_task-c7bf4036-15' immediately...
07:54:56.411 | INFO    | Task run 'perf_task-c7bf4036-15' - Finished in state Completed()
07:54:56.424 | INFO    | Flow run 'strange-starling' - Created task run 'perf_task-c7bf4036-16' for task 'perf_task'
07:54:56.424 | INFO    | Flow run 'strange-starling' - Executing 'perf_task-c7bf4036-16' immediately...
07:54:56.447 | INFO    | Task run 'perf_task-c7bf4036-16' - Finished in state Completed()

I am using prefect 2.7.6. Thanks in advance

@here Is there a way to delete a JSON block via python?

I have tried to use .delete but it throws the following error JSON.delete("prefect-kv-test") AttributeError: type object 'JSON' has no attribute 'delete'

I am using Prefect 2.0

Hi guys, I am running a huge queue ( 15 tasks) , it runs into this SSLWantReadError Traceback (most recent call last): File "/home/ubuntu/function-triggers/env/lib/python3.10/site-packages/anyio/streams/tls.py", line 130, in _call_sslobject_method result = func(*args) File "/usr/lib/python3.10/ssl.py", line 917, in read v = self._sslobj.read(len) ssl.SSLWantReadError: The operation did not complete (read) (_ssl.c:2548) During handling of the above exception, another exception occurred: Traceback (most recent call last): File "/home/ubuntu/function-triggers/env/lib/python3.10/site-packages/httpcore/backends/asyncio.py", line 33, in read return await self._stream.receive(max_bytes=max_bytes) Traceback (most recent call last): File "/home/ubuntu/function-triggers/env/lib/python3.10/site-packages/httpcore/backends/asyncio.py", line 33, in read return await self._stream.receive(max_bytes=max_bytes) File "/home/ubuntu/function-triggers/env/lib/python3.10/site-packages/anyio/streams/tls.py", line 195, in receive data = await self._call_sslobject_method(self._ssl_object.read, max_bytes) File "/home/ubuntu/function-triggers/env/lib/python3.10/site-packages/anyio/streams/tls.py", line 137, in _call_sslobject_method data = await self.transport_stream.receive() File "/home/ubuntu/function-triggers/env/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 1265, in receive await self._protocol.read_event.wait() File "/usr/lib/python3.10/asyncio/locks.py", line 214, in wait await fut asyncio.exceptions.CancelledError During handling of the above exception, another exception occurred: Traceback (most recent call last): File "/home/ubuntu/function-triggers/env/lib/python3.10/site-packages/httpcore/_exceptions.py", line 8, in map_exceptions yield File "/home/ubuntu/function-triggers/env/lib/python3.10/site-packages/httpcore/backends/asyncio.py", line 31, in read with anyio.fail_after(timeout): File "/home/ubuntu/function-triggers/env/lib/python3.10/site-packages/anyio/_core/_tasks.py", line 118, in exit raise TimeoutError TimeoutError During handling of the above exception, another exception occurred:

Traceback (most recent call last):

Hi guys, I am running a huge queue ( 15 tasks) , it runs into this SSLWantReadError

Traceback (most recent call last):
  File "/home/ubuntu/function-triggers/env/lib/python3.10/site-packages/anyio/streams/tls.py", line 130, in _call_sslobject_method
    result = func(*args)
  File "/usr/lib/python3.10/ssl.py", line 917, in read
    v = self._sslobj.read(len)
ssl.SSLWantReadError: The operation did not complete (read) (_ssl.c:2548)

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/home/ubuntu/function-triggers/env/lib/python3.10/site-packages/httpcore/backends/asyncio.py", line 33, in read
    return await self._stream.receive(max_bytes=max_bytes)

Traceback (most recent call last):
  File "/home/ubuntu/function-triggers/env/lib/python3.10/site-packages/httpcore/backends/asyncio.py", line 33, in read
    return await self._stream.receive(max_bytes=max_bytes)
  File "/home/ubuntu/function-triggers/env/lib/python3.10/site-packages/anyio/streams/tls.py", line 195, in receive
    data = await self._call_sslobject_method(self._ssl_object.read, max_bytes)
  File "/home/ubuntu/function-triggers/env/lib/python3.10/site-packages/anyio/streams/tls.py", line 137, in _call_sslobject_method
    data = await self.transport_stream.receive()
  File "/home/ubuntu/function-triggers/env/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 1265, in receive
    await self._protocol.read_event.wait()
  File "/usr/lib/python3.10/asyncio/locks.py", line 214, in wait
    await fut
asyncio.exceptions.CancelledError

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/home/ubuntu/function-triggers/env/lib/python3.10/site-packages/httpcore/_exceptions.py", line 8, in map_exceptions
    yield
  File "/home/ubuntu/function-triggers/env/lib/python3.10/site-packages/httpcore/backends/asyncio.py", line 31, in read
    with anyio.fail_after(timeout):
  File "/home/ubuntu/function-triggers/env/lib/python3.10/site-packages/anyio/_core/_tasks.py", line 118, in __exit__
    raise TimeoutError
TimeoutError
During handling of the above exception, another exception occurred:

Traceback (most recent call last):
File "/home/ubuntu/functions-triggers/env/lib/python3.10/site-packages/httpcore/_exceptions.py", line 12, in map_exceptions
    raise to_exc(exc)  
httpcore.ReadTimeout

Any idea what does this error mean or why is it crashing ?