Does anybody know if

task concurrency limit tags

are enforced even when the task has already been cached before [i.e. when fetching data off the cache] ?

Hey guys, I have following dummy code, which saves the values to the database and returns a value to the client. I use prefect flow and task. Change the code, so the flow_url_upload() and url_upload() will not wait for the long task save_to_db(). The task save_to_db() should be just pushed to some queue, and the function flow_url_upload() should not wait for its execution. Is there some way?

import fastapi
from prefect import flow, task
import time

@task()
def save_to_db(a,b):
    time.sleep(100)
    print("Saving to db...")
    print("Saved to DB")

@flow()
def flow_my_endpoint(a,b):
    save_to_db(a,b)
    suma = a + b
    return {'suma': suma}

@router.post("/csv/url")
async def my_endpoint(a, b):
    return flow_my_endpoint(a,b)

Hi all, has anyone seen flows crash with messages like this?

13:29:44.526 | INFO    | prefect.engine - Engine execution of flow run '0f87d3ab-658a-4fb8-bf9d-9daf903bcaf1' aborted by orchestrator: This run cannot transition to the RUNNING state from the RUNNING state.

13:29:43.035 | INFO    | Flow run 'unnatural-leech' - Downloading flow code from storage at None
  warn(RuntimeWarning(msg))
/usr/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

prefect version

:

Version:             2.8.7
API version:         0.8.4
Python version:      3.10.10
Git commit:          a6d6c6fc
Built:               Thu, Mar 23, 2023 3:27 PM
OS/Arch:             linux/x86_64
Profile:             default
Server type:         ephemeral
Server:
  Database:          sqlite
  SQLite version:    3.31.1

Hi peoples. Following scenario. I have 3 hosts A,B and C. I have an deployment which runs an flow. I will load balancing this flow between this 3 hosts. At the moment I use the api and python code to update de deployment to start the flow on different worker-queues. My problem lies on starting the agents. The agent must known the worker-queue on start time or using rhe match argument. But is there a way to start an agent which listen on a specific Pool and fetch all queses in this pool?

Hello! Any idea why I get this exception retrieving flow runs. I just upgraded to 2.9 bc of the gzip functionality and it times out sometimes with the traceback in the thread

Hey everyone, I'm getting some weird behavior between running a prefect flow locally and running on

DockerContainer

infra on a prefect agent. So I'm trying to pass an instance of an object as an argument to my first task, this object happens to be a dataclass but more complex. Locally it works fine, but on the prefect agent, the engine tries to resolve my object and create a new instance of it before passing it to the task !? I'm on prefect 2.9.0 and python3.11 (details about the error in the thread ⬇️)

Hello, I have a deployment on an hourly schedule in GKE. Occasionally I have a jobs that appear as Running in the UI but when I look at the pod it has a state like:

chocolate-cheetah-bwnpz-zlnxj        0/1     Completed                0          3h42m

The meta data in the Prefect Cloud UI in the side bar matches that 3h42m

Flow Run
Start Time
2023/04/05 06:00:05 AM
Duration
3h 44m
Run Count
1
Created
2023/04/05 03:00:09 AM

What would you recommend I do? Will this result in bloat over time where Prefect shows lots of running processes even though they are not? At the deployment infra level I have in the KubernetesJob

job_watch_timeout_seconds = 3600,

Hello, I am struggling with identifying a flow_run state, when it is a 'TriggerFail' object. I need to be able to detect this state as there are missing attributes like 'name' and such on a flow_run object when the state is 'TriggerFail'. I have searched all over, and I have tried multiple approaches, including flow_run.state.type which does not exist, even though the documentation suggests it should. There is no flow_run.state.is_triggerfail() method. The state output looks like a dictionary object if you output it to a logger. How can I simply detect if the state object is 'TriggerFail'?

This message was deleted.

Hello y’all! is it possible to pass and/or execute a

*.sql

file in any

prefect_sqlalchemy.SqlAlchemyConnector

methods like

fetch_all

?

I want to store some state in a JSON block, but there needs to be an individual state per incoming flow parameter. Is it possible to create JSON blocks in the job? do I just wrap a

JSON.load(state/{mykey}

in a try, and then in the except block do

JSON(value={initvalue})

+

json_block.save

?

roast my flow: https://gist.github.com/mattfysh/916827271255fea30c5b2407189d76a1

Hey, has anyone tried to use new @Marvin bot library to migrate from Prefect1 to Prefect2? It seems to me an interesting challenge, but maybe someone has tried it already.

is it possible to run 2 tasks from 1 task? For example:

--> task_b --
        |            |
task_a -             ----> task_d
        |            |
         --> task_c --

If so, will

task_d

wait for both

task_b

and

task_c

to finish before executing? Any guidance on this would be greatly appreciated- Thanks!

Hey. Does Prefect 2 support on cancel flow handler?

Hello, I'm trying to use a custom kubernetes-job block, every run gives this error :

data = data[field]
KeyError: 'env'

I reduced the job template to practically nothing custom here in case it was from my additionnal configuration, but still have this error. What am I doing wrong ?

Hello all, I'm seeing the strangest error when pulling down flows from a samba block, it's working in two other environments, and permissions to the shared folder look ok. I've trying getting debug logs and putting the api in debug mode, but I'm not seeing any error messages:

Downloading flow code from storage at 'EzeCastle'

08:11:02 AM

DEBUG

Importing flow code from 'ezecastle_daily_p_and_l_loader.py:hourly_loader'

08:11:02 AM

ERROR

Flow could not be retrieved from deployment.

Traceback (most recent call last):

File "<frozen importlib._bootstrap_external>", line 846, in exec_module

File "<frozen importlib._bootstrap_external>", line 982, in get_code

File "<frozen importlib._bootstrap_external>", line 1039, in get_data

FileNotFoundError: [Errno 2] No such file or directory: 'ezecastle_daily_p_and_l_loader.py'

The above exception was the direct cause of the following exception:

Traceback (most recent call last):

File "C:\PrefectAgent\.venv\lib\site-packages\prefect\engine.py", line 268, in retrieve_flow_then_begin_flow_run

flow = await load_flow_from_flow_run(flow_run, client=client)

File "C:\PrefectAgent\.venv\lib\site-packages\prefect\client\utilities.py", line 47, in with_injected_client

return await fn(*args, **kwargs)

File "C:\PrefectAgent\.venv\lib\site-packages\prefect\deployments.py", line 187, in load_flow_from_flow_run

flow = await run_sync_in_worker_thread(load_flow_from_entrypoint, str(import_path))

File "C:\PrefectAgent\.venv\lib\site-packages\prefect\utilities\asyncutils.py", line 91, in run_sync_in_worker_thread

return await <http://anyio.to|anyio.to>_thread.run_sync(

File "C:\PrefectAgent\.venv\lib\site-packages\anyio\to_thread.py", line 31, in run_sync

return await get_asynclib().run_sync_in_worker_thread(

File "C:\PrefectAgent\.venv\lib\site-packages\anyio\_backends\_asyncio.py", line 937, in run_sync_in_worker_thread

return await future

File "C:\PrefectAgent\.venv\lib\site-packages\anyio\_backends\_asyncio.py", line 867, in run

result = context.run(func, *args)

File "C:\PrefectAgent\.venv\lib\site-packages\prefect\flows.py", line 786, in load_flow_from_entrypoint

flow = import_object(entrypoint)

File "C:\PrefectAgent\.venv\lib\site-packages\prefect\utilities\importtools.py", line 195, in import_object

module = load_script_as_module(script_path)

File "C:\PrefectAgent\.venv\lib\site-packages\prefect\utilities\importtools.py", line 158, in load_script_as_module

raise ScriptError(user_exc=exc, path=path) from exc

prefect.exceptions.ScriptError: Script at 'ezecastle_daily_p_and_l_loader.py' encountered an exception: FileNotFoundError(2, 'No such file or directory')

Hey all, When using

task.map

on a large sequence of inputs (while defining a tag concurrency limit of 20), prefect seems to crash a lot while trying

set_state

of tasks with a status code 500 (traceback in the thread ⬇️ ) Do you have any recommandations on how to work around this ? can define retries for prefect internal API ? is it something I'm doing wrong ?

Within a flow run is there a way to access the deployment_name of the flow? I am currently able to access the deployment ID by using

from prefect.context import get_run_context

and then accessing the MetaData from there. Just no deployment name within there sadly.

Hello everyone!, I would like to ask for your assistance regarding an issue I've encountered with the

prefect-gcp

library when configuring the

vpc-connector

. It appears that the library is not properly setting up the call to GCP's REST API. Here's the error message I received:

HttpError 400 when requesting <https://us-west2-run.googleapis.com/apis/run.googleapis.com/v1/namespaces/ba-basic/jobs?alt=json> returned "metadata.annotations[<http://run.googleapis.com/vpc-access-connector|run.googleapis.com/vpc-access-connector>]: Annotation '<http://run.googleapis.com/vpc-access-connector|run.googleapis.com/vpc-access-connector>' is not supported on resources of kind 'Job'. Supported kinds are: Revision, Execution"

Could anyone please provide some guidance on how to resolve this issue, or point me in the right direction? Any help would be greatly appreciated!

Hi! I’m very excited to see 2.10.1 release. Just curious, what’s the difference between using variables and using string blocks? It seems you can tag a variable?

Having trouble solving this bug. When I test the flow locally, it runs fine. But when I test it with an ECS Task, it throws this error.

Crash detected! Execution was interrupted by an unexpected exception: Traceback (most recent call last):
  File "/usr/local/lib/python3.11/site-packages/prefect/client/orchestration.py", line 1031, in create_block_schema
    response = await <http://self._client.post|self._client.post>(
               ^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/httpx/_client.py", line 1848, in post
    return await self.request(
           ^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/httpx/_client.py", line 1533, in request
    return await self.send(request, auth=auth, follow_redirects=follow_redirects)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/client/base.py", line 269, in send
    response.raise_for_status()
  File "/usr/local/lib/python3.11/site-packages/prefect/client/base.py", line 132, in raise_for_status
    raise PrefectHTTPStatusError.from_httpx_error(exc) from exc.__cause__
prefect.exceptions.PrefectHTTPStatusError: Client error '409 Conflict' for url '<https://api.prefect.cloud/api/accounts/e5e970ea-d0a6-4ac5-992d-3796d5308920/workspaces/dbe8a5b3-2746-4ebd-8735-afdcc6550e47/block_schemas/>'
Response: {'detail': 'Data integrity conflict. This usually means a unique or foreign key constraint was violated. See server logs for details.'}
For more information check: <https://httpstatuses.com/409>

The above exception was the direct cause of the following exception:

ObjectAlreadyExists

Hi everyone! I am running prefect local and work fine, but when I try to run a agent in ec2 i got this error:

prefect.exceptions.ScriptError: Script at 'test.py' encountered an exception: FileNotFoundError(2, 'No such file or directory')

What am i doing wrong? My storage is on github!

Recently Upgraded prefect from 2.9.0 to 2.10.1. Everything was working fine before with 2.9.0 but now getting error. <FileNotFoundError: [Errno 2] No such file or directory: '/opt/prefect/src/downloader_0.8.py'>. The code is stored in the root directory . Using docker container. Code stored on minio bucket.

Hii Community , if my prefect workflow consists of Pypark code i want to run on data proc . how do i do that ?

Hi! I have a high-level question- I am working on a Prefect 1 -> Prefect 2 migration, and am trying to understand the new version of Agents. We currently deploy our ECS infrastructure using the AWS Python CDK, and deploy an agent on there via the "prefect agent start" command in docker. That agent listens for work and executes it. Simple. 1. Is Prefect 2, does THE AGENT lay down infrastructure instead of the CDK? 2. If so, does THE AGENT need to control the infrastructure? or can it point to existing infrastructure? Feeling very confused on where to go from here - seems like blocks would replace our current AWS Python CDK script? Yet I see Terraform configs in some Prefect recipes.. TLDR: Is it possible to use our current design pattern using the Python CDK with Prefect 2? Again, sorry for all the questions- I'm new to a lot of these technologies. Any insights would be so very appreciated!

QQ Strategy question. I have written my first prefect deployment to get a days worth of event logs from one of our production servers (Prefect-SQLAlchermy + pandas). This works well but due to capacity limitations and table design which I have no control over, At most I can pull about 2 days worth of data before the query becomes unsustainable. My current workflow works fine for single day pulls and supports data ingestion moving forward, but I am considering strategies to use prefect effectively to backfill data more efficiently. This is a new initiative and I have room to play, but I was hoping to understand if there was a prefect-minded way to execute multiple smaller queries within a flow and return the combined dataset.

Hi! In prefect 1.0 I used

with prefect.context({"date": datetime.datetime(2011, 1, 1)}):

to set context variables and test my flows. Is there something similar in prefect 2.0 to set the

start_date

variable from the flow run context (

prefect.context.get_run_context()

)? Thank you!

Hi, Is there any way to deploy Prefect agent on AWS EKS with Terraform but not using Helm ?

Hi, I'm running into an issue that's been posted a few times before... first there's a warning...

[WARNING] Task run 'bcc56e8e-a1bd-4145-abdb-aad866d63c6f' received abort during orchestration: The enclosing flow must be running to begin task execution. Task run is in PENDING state.

then an error:

[ERROR] prefect.exceptions.MissingResult: State data is missing. Typically, this occurs when result persistence is disabled and the state has been retrieved from the API.

Previous advice has been that this is caused by an upstream problem, but I can't see anything in the logs (journalctl on ec2 instance) that would cause the enclosing flow to not be in a running state. Our infra is an prefect agent running on ec2, spawning docker containers on the same box. I checked the monitoring on the box - no cpu/mem spikes. I've added task retries, but I guess that's not going to fix it either, if the flow is already dead? What else should I look for?