This message was deleted.

Hello everyone! I am transitioning my prefect 1 workflows to prefect 2 this month. My prefect 1 code uses the generic git storage to pull code from an Azure Devops repo. I don't see a generic git block in the documentation. Is Azure DevOps not supported as storage in Prefect 2?

Been receiving these error messages. Weird! Can anyone explain?

Submission failed. kubernetes.client.exceptions.ApiException: (500) Reason: Internal Server Error HTTP response headers: HTTPHeaderDict({'Audit-Id': '6d00e224-73e9-4d44-bfce-d0589101f116', 'Cache-Control': 'no-cache, private', 'Content-Type': 'application/json', 'Warning': '299 - "unknown field \\"spec.template.spec.parallelism\\"", 299 - "unknown field \\"spec.template.spec.completions\\""', 'X-Kubernetes-Pf-Flowschema-Uid': 'e8e112c2-396a-440d-a000-51271ccde7dd', 'X-Kubernetes-Pf-Prioritylevel-Uid': 'a49f8ad5-8431-4216-90ae-cd35c64b0736', 'Date': 'Wed, 10 May 2023 18:30:59 GMT', 'Content-Length': '161'}) HTTP response body: {"kind":"Status","apiVersion":"v1","metadata":{},"status":"Failure","message":"rpc error: code = Unknown desc = exec (try: 500): database is locked","code":500}

hi ya! does anyone use Prefect for a background worker type of pattern? As like a Celery or queue/serverless function alternative?

HI guys any luck on debugging this error ?

/usr/bin/python: No module named prefect.engine.__main__; 'prefect.engine' is a package and cannot be directly executed

Hello Prefect Community! I saw a thread from 1 year ago about running multiple Orion servers behind a load balancer. That thread mentioned it was necessary to run only a single scheduler service, but the rest of the services on the server could be run on each server service. Is that still the case?

Hi Everyone, is there a way to have a deployment retry not from point of failure, even though I have persist_result=True? For example, I have a deployment with 10 tasks, and it fails on task 5. I would like it to restart from task 2 with different params, even though there may be other times where I want it to restart from task 5. Is there a way to achieve this without needing to alter the code?

Hello Prefect team, Are subflows supposed to be canceled when parent flows are cancelled? If that is not a default behavior, what is the best way for me to get the ids of subflows (assuming I have the parent flow id) so I can manually cancel them?

Hi all! Is there a way to avoid hardcoding the db credentials for

SQLAlchemyConnector

block and fetch it directly from AWS Secrets?

Hey folks. Two questions on running flows on Docker - 1. The deployment guide mentions

You'll need Docker Engine installed and running on the same machine as your agent

My agent runs on EKS using the official helm chart, but fails with

RuntimeError: Could not connect to Docker

Do I need to install DockerEngine separately on the agent? 2. Can someone point me towards resources on running flows embedded in ECR images through ECS? ( My agent would continue to run on EKS though ) Any help would be largely appreciated 🙏

can we use dataproc for storing the code while we use the prefect api to trigger the flow

Hello everyone. I'm having a really weird issue with one of my subflows, where it simply crashes with the State message

Flow run infrastructure exited with non-zero status code 1.

The only log message I can see in the console is

Downloading flow code from storage at None

. The parent flow works correctly. Are there any server-side logs I can potentially take a look at here? I'm using Prefect

2.8.5

, but I'll try upgrading to

>=2.10.7

(couldn't until now because of the pinned

fsspec

version up

2.10.7

) to see if that might help. Weirdly, it worked the day before and there have been no changes to the storage blocks...

Hello everyone! Does sb know if there is a on_cancelled hook that we can use for flows to execute code on cancellation (e.g like cleanup of things)?

Just got a flow called crystal-moth. Nice

Hi, I want to create a DAG using prefect and run the tasks such that all dependencies are satisfied, essentially in a topologically sorted manner. I referrred to https://discourse.prefect.io/t/how-to-define-dag-like-flow-in-prefect-2-0/2069/7 but here are some constraints I am facing.

my_network_graph = {
    "task_1": [],
    "task_2": [],
    "task_3": ["task_1", "task_2"],
    "task_4": ["task_3"],
    "task_5": ["task_4"],
}

This is the graph mentioned in the example, if I change this to include

my_network_graph = {
    "task_5": ["task_4"],
    "task_1": [],
    "task_2": [],
    "task_3": ["task_1", "task_2"],
    "task_4": ["task_3"],
}

then task 5 would be executed first regardless of its dependency on task 4. I want to give a dictionary with deps, but the order of keys in the dictionary shouldnt matter. This becomes especially important in my use case where I have 100+ nodes in the graph, and its complicated to chart out the deps such that all upstream deps are mentioned before in the dictionary. How can I solve this?

Essentially the way things are defined in this example, you first get the PrefectFutures of tasks without any dependencies by executing them, then use these futures for downstream dependencies. I want to define a dictionary with a node and its dependent nodes, and for prefect to determine and execute which should be run first then use that future for the downstream dependencies

Hey folks -- question about Pydantic & JSON. I'm adding Prefect into an application that makes pretty heavy use of Pydantic right now. We have some subtypes of builtins that serialize fine using our Pydantic setup but don't seem to work when passing these as parameters to a Prefect deployment -- or specifically the orjson serializer. For our models, we are overriding some of the serializers, though not for this specific type (more below). Is there a way to configure Prefect not use orjson or to respect our serializers ... or something else? Here's the prefect side of the stack trace I'm getting:

File "/opt/executor/.venv/lib/python3.10/site-packages/prefect/client/orchestration.py", line 460, in create_flow_run_from_deployment
    json=flow_run_create.dict(json_compatible=True),
         │               └ <function PrefectBaseModel.dict at 0xffffa59ff2e0>
         └ DeploymentFlowRunCreate(state=StateCreate(type=StateType.SCHEDULED, name='Scheduled', message=None, data=None, state_details=...
  File "/opt/executor/.venv/lib/python3.10/site-packages/prefect/server/utilities/schemas.py", line 293, in dict
    return json.loads(self.json(*args, **kwargs))
           │    │     │    │     │       └ {}
           │    │     │    │     └ ()
           │    │     │    └ <function PrefectBaseModel.json at 0xffffa59ff250>
           │    │     └ DeploymentFlowRunCreate(state=StateCreate(type=StateType.SCHEDULED, name='Scheduled', message=None, data=None, state_details=...
           │    └ <function loads at 0xffffb5241090>
           └ <module 'json' from '/usr/local/lib/python3.10/json/__init__.py'>
  File "/opt/executor/.venv/lib/python3.10/site-packages/prefect/server/utilities/schemas.py", line 247, in json
    return super().json(*args, **kwargs)
                         │       └ {}
                         └ ()
  File "/opt/executor/.venv/lib/python3.10/site-packages/pydantic/main.py", line 504, in json
    return self.__config__.json_dumps(data, default=encoder, **dumps_kwargs)
           │    │          │          │             │          └ {}
           │    │          │          │             └ functools.partial(<function custom_pydantic_encoder at 0xffffb440e4d0>, {<class 'pydantic.types.SecretField'>: <function Pref...
           │    │          │          └ {'state': {'type': StateType.SCHEDULED, 'name': 'Scheduled', 'message': None, 'data': None, 'state_details': {'flow_run_id': ...
           │    │          └ <function orjson_dumps at 0xffffa59fee60>
           │    └ <class 'pydantic.config.Config'>
           └ DeploymentFlowRunCreate(state=StateCreate(type=StateType.SCHEDULED, name='Scheduled', message=None, data=None, state_details=...
  File "/opt/executor/.venv/lib/python3.10/site-packages/prefect/server/utilities/schemas.py", line 126, in orjson_dumps
    return orjson.dumps(v, default=default).decode()
           │      │     │          └ functools.partial(<function custom_pydantic_encoder at 0xffffb440e4d0>, {<class 'pydantic.types.SecretField'>: <function Pref...
           │      │     └ {'state': {'type': StateType.SCHEDULED, 'name': 'Scheduled', 'message': None, 'data': None, 'state_details': {'flow_run_id': ...
           │      └ <built-in function dumps>
           └ <module 'orjson' from '/opt/executor/.venv/lib/python3.10/site-packages/orjson/__init__.py'>

TypeError: Type is not JSON serializable: WrappedFloat

In this case, I have a fairly complex graph of Pydantic objects, but specifically the one causing issue is defined something like this:

class WrappedFloat(float):
  ...

class MyObject(MyBaseModel):
  probability: WrappedFloat = WrappedFloat(1.0)

Obviously that is simplified, but effectively removing the WrappedFloat value fixes the issue ... but this is definitely a code quality compromise to change this just so that Prefect can serialize it. Pydantic itself does not complain about serializing this; base classes of known types are supported (thanks to, I believe, https://github.com/pydantic/pydantic/pull/1291) Are there other workarounds we should pursue? Thanks in advance!

Hey everyone, CONTEXT So I have a for loop that creates a bunch of response futures using

run_deployment()

. I will

await

for all of these flow runs to complete, at which point I check if all of the futures have a state of "Completed". However, this deployment I'm running also creates a sublow just before the parent flow completes, and I would like to check if the state of the subflows are complete as well. The future ( parent flow ) returns a state when the parent flow completes, but I have no way to check if the subflow completes QUESTION Is there a way to use the response of the

run_deployment()

method to check if a subflow is Complete? How can I programmatically get the state of a sublow if a subflow is created with

run_deployment()

?

HI. I have a question regarding the deployment of the prefect scheduler. If I want to deploy a new scheduler program, will the existing programs deployed on the prefect server be unaffected, allowing me to deploy the new programs? For example, let's say A, B, and C schedulers are already deployed, and I want to deploy D. It seems inefficient to use the prefect build command. I handle the version management of prefect-related scheduler programs in Jenkins.

The file structure is as follows: Under the "service/scheduler" directory, developers can write code to develop and deploy the desired scheduler program. Afterwards, the code can be containerized using Docker for server operation. The "run.sh" file contains the execution command for deployment. Here is an example of the code:

#!/bin/bash
directories=$(find "$SCHEDULER_DEPLOYMENT_PATH" -type d -not -name "__pycache__" -not -path "$SCHEDULER_DEPLOYMENT_PATH")
for directory in ${directories}; do
  files=$(find "$directory/deployment.py");
  python "$files";
done;

Hi all! I have been trying to send slack notification for flow run states. I have tried to use one automation for all the four states pending, running completed and failed. But the notification contents gets overlapped and I get same notification thrice, and sometimes it sends the notification late. I also created 4 separate automation, one for each state. By this way I get correct notification message and also in time. But the thing is one workspace is limited to 10 automation only. So is it possible to create one automation for each queue and get the notifications for each of the 4 flow_run states in time?

Hi everyone! I have a self hosted Prefect ec2 instance. I’ve noticed a strange behaviour in the web ui. When I filter by date (no other filters applied), from whatever is the starting date to today I can see runs only from 03-05-2023. I cannot see runs older than 03-05-2023. Why is that happening?

Hi, I am planning to create a PR with commit. Before I post the PR I am getting an email that tells me that it is not compatible:

Check compatibility with Prefect 2.9
  Process completed with exit code 1.
Check compatibility with Prefect 2.10
  Process completed with exit code 1.
Check compatibility with Prefect 2.6
  Process completed with exit code 1.
Check compatibility with Prefect 2.8
  Process completed with exit code 1.
Check compatibility with Prefect 2.7
  Process completed with exit code 1.

I am not able find out what is not compatible. Could you please give me an hint?

HI, I am some queries about

testing

.

@flow
def flow():
    #SOME STUFF HERE
    normal_function()
    #SOME OTHER STUFF HERE

def normal_function():
    return some_task()

@task
def some_task():
    return True

Hello everyone, I have a strange case where my Flow gets executed 2 times in a row while the retries are set to 0 and execution is successful. I have persistence disabled and second run is breaking due to missing result storage. Here is a screen, i have 2 runs & retry set to 0. Any ideas what is wrong here? What is more on the first run is completes successfully and immediately runs again.

Hi everyone, we are running Prefect (2.10.7) in AKS using the orchestrator pattern. A subflow failed, but it does not stop recreating itself over and over again. Even canceling the current subflow instance via CLI does not help. It only stops when I completely delete the corresponding deployment. We haven't configured retries so far (neither within the flow code nor via env vars), so I am wondering what's the reason for this behaviour? Thanks in advance!

@flow
def georgia_table_flow(files):
    truncate = truncate_tables(files)
    import_stg = import_data(files)
    alter_import = switcheroo(files)


@flow
def georgia_etl():
    table_flows = [georgia_table_flow(file) for file in files]

georgia_etl()

These are my flows and my sub flow. I am having two problems 1. I want to name my subflow based upon where my main flow is in the loop this will be helpful to identify business logic tied to the subflow. Secondly when one of my subflows failes it stops the main flow. Is there away to allow the rest of the subflows to continue to run. Thanks!

hey all, on prefect 1.0 my team uses CICD to register flows to different projects on a pr merge. we have

PREFECT__CLOUD__REQUEST_TIMEOUT=60

in our CICD env, but I see timeouts happening in 14s sometimes. anyone have any tips on what could be going on?

Hi all, we are running Prefect (2.8.7) with a github storage block, and I am running into the error

FileNotFoundError: [Errno 2] No such file or directory: 'git'

. It was working fine, until we re-deployed a fresh EC2, and now it's not working. Is there any kind of configuration I'm missing, other than having a git-hub block & having

prefect-github

installed on the EC2? Any help is so appreciated! (Full stack trace in thread)

Hi All, We're using Prefect 2 and our project contains multiple deployments created from a Python objects (with S3 as a storage). Once we apply the deployment it's being uploaded to the storage along with all other deployments from the same directory, which is not desired behaviour as it means that each time we reupload the same files all over again. Is there any way to restrict the upload only to the deployment that we build? I don't think that .prefectignore file will be very helpful here as at some point we will need to upload other deployments as well