Hi, Is there any way to check if file exists in blob storage using prefect_azure.blob_storage package? Thanks Nitin

Hi, since upgrading to prefect 2.10.10 yesterday one of our flows has been crashing with the error:

Crash detected! Execution was interrupted by an unexpected exception: Traceback (most recent call last):
  File "/usr/local/lib/python3.10/site-packages/anyio/streams/tls.py", line 130, in _call_sslobject_method
    result = func(*args)
  File "/usr/local/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:

RuntimeError: Event loop is closed

Have there been any other reports of this?

hi all! Trying to test tasks that make use of the run context with

Prefect 1

I could do this:

def test_my_task():
    with prefect.context(dict(flow_run_id="test_id")):
      my_task()
    # assertions

how c an this be done in

Prefect 2

?

Hello - I have a flow containing a task that has a max_retries value of 1. I want to kick off another flow after the task has finished:

flow.set_dependencies(create_my_flow_run(), upstream_tasks=[task_with_1_retry])

But this only runs

create_my_flow_run

after the retry has finished, I want to run the flow after the first failure, so I tried adding:

@task
...
try:
    something_that_might_fail() 
finally:
    StartFlowRun(flow_name=my_flow.name).run()

To the bottom of

task_with_1_retry

- but getting the error:

Unexpected error while running flow: KeyError('Task slug case(1111)-1 is not found in the current Flow. This is usually caused by a mismatch between the flow version stored in the Prefect backend and the flow that was loaded from storage.

Any ideas how to achieve what I described? Thanks!

Is there a way to search in a task/flow log? In the browser ctrl+f does not work, it is a pain scrolling tons of lines of logs

I’m trying to delete a block from cli using

prefect block delete --id <some_id>

but I get

Deployment '<some_id>' not found!

is it a bug?

Hey folks. I am facing issues trying to retrieve the flow code from my remote storage S3. The deployment submission goes through, but then fails with the following simplified stacktrace -

12:30:23.877 | INFO    | Flow run 'refreshing-coyote' - Downloading flow code from storage at ''
12:30:36.187 | ERROR   | Flow run 'refreshing-coyote' - Flow could not be retrieved from deployment.
....
socket.gaierror: [Errno -5] No address associated with hostname
...
aiohttp.client_exceptions.ClientConnectorError: Cannot connect to host <bucket-name>.<http://s3.eu-west-1.amazonaws.com:443|s3.eu-west-1.amazonaws.com:443> ssl:default [No address associated with hostname]
...
botocore.exceptions.EndpointConnectionError: Could not connect to the endpoint URL: "https://<bucket-name>.<http://s3.eu-west-1.amazonaws.com/?list-type=2&prefix=&delimiter=&encoding-type=url|s3.eu-west-1.amazonaws.com/?list-type=2&prefix=&delimiter=&encoding-type=url>"
...

I am using a ECS task infra block ( definition in comments ) and building my deployment like this -

prefect deployment build ecr_flow.py:ecr_flow -n ecr_flow_path -sb s3/prefect-s3-storage --path opt/prefect/flows -a -q rito

Any pointers where I could be going wrong?

For those of us still with prefect cloud 1 accounts working on a transition, anyone know what is the best way to contact support for urgent issues? This nearly all of our schedules got deactivated without any action on our part. Thanks.

Good morning everyone, I got a challenge for the community of prefect. Anyone could suggest a good strategy to create a website that grabs data from a google sheet and presents that data? Right now I have a google form where users would enter basic information. Ideally I would like to set a script that grabs the data from the google sheet every time a new users enters a new row or a change is made on the spreadsheet and displays it on a website ( designed on HTML/CSS). For now I have a basic script and design working but need to run it manually locally because the Google API credentials are pretty vulnerable as of now with something like GitHub. I was reading the block section from prefect and it seems like that functionality could solve some of these problems but i am still doubtful or where I could safely allocate the website files so they run on trigger as I detailed above.

Hi everyone. I have a flow running on ECS. The flow needs to write data to an s3 bucket that is in a different account. I was using AWS keys in a Credential block and the S3 block but they want to deprecate the keys and use roles instead. They AWS admin apparently granted permission to the task execution role to access s3 in the other account. However, I tried using the s3 block without a credential block associated and am getting

Task run encountered an exception: botocore.exceptions.NoCredentialsError: Unable to locate credentials

Is it possible to use the S3 Block/Credential block to assume a specific role?

Prefect 1.0 I have a task which calls a non-task function. In that function I use

print

statements. I am unable to see this output when I run the flow. What am I doing wrong?

I must be missing something simple… I have two production servers … we will call them Primary and Prefect … as you can guess the Prefect server houses my agent and accepts 10% of the web traffic. The Primary server accepts 90% of the web traffic and has at least one

run_deployment

specified in one of it’s routes. When posting to that URL I get an error and it seems like it is looking for the deployment on a local copy of Prefect Server (previously Orion) instead of Prefect Cloud which is what I’m using… Here’s the error without the stack trace…

prefect.exceptions.PrefectHTTPStatusError: Client error '404 Not Found' for url '<http://ephemeral-prefect/api/deployments/name/alert-early-intervention/alert_early_intervention>'

… I have confirmed that the deployment exists in my Prefect Cloud under the name

alert-early-intervention/alert_early_intervention

.

Has anyone used a kafka processing library with Prefect? I'm thinking about creating a prefect flow, using Faust to process kafka topics

@Zanie not sure who else to tag, but I will also create an issue on Prefect’s github. Possible code braking bug found with a complete reproducible code included in this thread. Basically, if a dataclass has a

__post_init__

method and checks and initializes a

self

parameter, the

__post_init__

gets called AGAIN when the a task run gets created.

I'm using prefect 2.10 to manage flows on a Kubernetes cluster, but I'm not connecting to any kind of Prefect cloud server. However, I'm starting to see this error:

Failed("Flow run encountered an exception. PrefectHTTPStatusError: Client error '429 Too Many Requests' for url '<https://api.prefect.cloud/api/accounts/XYZ/workspaces/XYZ/flow_runs/XYZ>'\nResponse: {'detail': 'Orchestration API rate limit reached'}\nFor more information check: <https://httpstatuses.com/429>\n"

I don't know how this is possible - since I'm in no way interacting with prefect server. I've used previous versions of Prefect and never seen this before. I've replaced the specific strings with XYZ

I’m getting an error that’s logged when running async flows locally with prefect that complains:

Exception ignored in: <async_generator object add_event_loop_shutdown_callback.<locals>.on_shutdown at 0x7f3d81b374c0>
Traceback (most recent call last):
  File "/home/jon/miniconda3/envs/abraxas-env/lib/python3.10/site-packages/prefect/utilities/asyncutils.py", line 308, in on_shutdown
AttributeError: 'NoneType' object has no attribute 'pop'

Which seems related to this issue (which I’ve already commented on). I have no idea how to solve this. It seems like some odd threading/multiprocessing bug where EVENT_LOOP_GC_REFS is not being copied over correctly somehow.

Considering code like this.

def child_flow_c():
    d = run_deployment("child-flow-d/dep-child-d")
    # How can I be sure that run_deployment("child-flow-d/dep-child-d") is completed before moving to next line/deployment 
    e = run_deployment("child-flow-d/dep-child-d")

How we check the status of

run_deployment

is completed before executing another

run_deployment

Hi Folks, my team and I are moving over some pipelines that make use of S3, and I cannot figure out how to get this new AWS creds block or the new S3 block working. Has anyone gotten this working? I've tried bringing over all sorts of configuration options, but I keep running into "profile not found for profile _____" and access denied errors. I can connect just fine with boto3 directly, but would like to use these blocks to shorten my code, but at this point, I feel like these blocks are more likely to shorten my lifespan 😆

I'd like to bulk cancel or delete scheduled flow runs but don't know how to in prefect cloud

I upgraded to the latest

prefect==2.10.11

version because of sqlalchemy 2.0 support. Thanks for the fast support there! 🚀 Now I have a problem with my custom block when I try to run my flows:

KeyError: "No class found for dispatch key 'my-custom-block' in registry for type 'Block'."

prefect_aws==0.3.2

is installed and

PREFECT_EXTRA_ENTRYPOINTS

is set. I’m using prefect cloud and the agent is updated to the latest version. I downgraded my prefect docker image to version 2.10.10 and it works fine again. Full log attached in the thread

Hi everyone, I have a question regarding the priority queue in the workpool So let suppose i have 2 queue (q1, q2) in a workpool (w) where q1 in a priority queue with highest priority. Now let say i have to multiple input (input1, input2 and so on) where "input 23" has the highest priority and need to processed as soon as it came to "q1" My question here is: 1. The scenario i mentioned above is okay ? 2. if input 16 come to q1 then all the inputs which are in q2 will be put on to hold till input 16 of q1 is processed ? Can anyone please help me on this.

been getting

Orchestration API rate limit reached

all day, any idea what's up? I have mapped functions that around 3-4k iterations. by the time I am around half of it. I hit this error and it crashes on its own. When I retry, by the time it finds all completed tasks (checkpoints), it will raise the error again and won't move forward. using prefect:2.10.10-python3.10 and running on ecs with s3 result storage

Is there any reason for the release of Prefect 2.10.11 not being announced in #announcements?

is there any prefect-dbt channel?

Hi everyone, Whats the difference between a variable and a StringBlock?

umm if any1 is an idiot like me n hadn't specified sqlalchemy version for their images. Now would be a good time, upgrading to prefect 2.10.11, sqlalchemy v2 was installed instead of v1

When using projects for multi-deployments, it looks like I should be able to have my deployment.yaml file in the parent directory of all the project folders; however, when I try to deploy I get the following error (i’ll put it in the thread)

One of my flow module imports itself imports our SQLAlchemy ORM models. When deployment.build_from_flow is run it throws this error (posted as a reply here). SQLAlchemy version is 1.4.46 Prefect version is 2.10.11 Python version is 3.11.2 I have tried decorating the

class AnnotationAirtable(Base)

declaration with typing.@runtime_checkable, but received the same error. I have tried separating the flow import with another module as a buffer (in desperation, to try to stop the ORM model import from happening!), also to no avail.

Hi everyone, I have a Prefect task that starts a child process that can take a while to run. When I cancel the flow, the child process doesn't get killed with it and this takes resources for nothing. Is there a way to kill it?

Hello. In Prefect2, I have a use-case where I need to share an object between 2 flows: Flow A instantiates a class and if the flow fails, I want to persist that object so that Flow B (started manually by a person) can retrieve that object and perform an operation. I'm thinking of doing this using result persistence, but I couldn't find a way to retrieve a flow run's persisted result from another flow. Any ideas on how to do this? Thank you