Is there a channel for feature requests? Now that I have everything set up in ECS I have some suggestions, particularly around the build control plane 🙂

When I set up Prefect in ECS, I went ahead and created the prefect worker task definition, but I noticed that there is also a prefect task definition for the Prefect flow run execution. Was that created by Prefect?

I am seeing a weird issue with tasks and pydantic models. This is my example. The exclude_unset=True is not working once when the task is called. If I use user_json.fn() it works as excepted. Does anyone know why the @task decorator would be messing with the pydantic model and if a known workaround is available?

class User(BaseModel):
    """Pydantic model for the User."""

    email: Optional[str]
    firstName: Optional[str]
    lastName: Optional[str]


@task()
def user_json(user: User):
    """Should return a JSON representation of only the set fields in the User model."""
    print(user)
    # Outputs: email=None firstName='John' lastName=None
    print(user.json(exclude_unset=True))
    # Outputs: {"email": null, "firstName": "John", "lastName": null}
    return user.json(exclude_unset=True)


@flow()
def fl_create_user():
    """Creates a new user with only the first_name set and calls the user_json task."""
    user = User(firstName="John")
    print(user)
    # Outputs: email=None firstName='John' lastName=None
    print(user.json(exclude_unset=True))
    # Outputs: {"firstName": "John"}
    json_user = user_json(user)
    print(json_user)
    # Outputs: {"email": null, "firstName": "John", "lastName": null}

I'm running dependent subflows with

run_deployment

(run from a task). I notice that if one fails, then the downstream ones don't fail. How do I make subflow failures propagate to the parent flow? I'm using the

ConcurrentTaskRunner

. Some need to run concurrently with each other.

for an example: I want the ones without the

wait_for

argument to all run at the same time, while any Failure kills the whole thing.

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


def fail_hook(flow, flow_run, state):
    print("Failed")
    raise Exception(state.message)


@task(log_prints=True, on_failure=[fail_hook])
def my_task():
    time.sleep(15)
    # some code here
    raise Exception("This task has failed.")  # this will set the task to Failed state


@task(log_prints=True, on_failure=[fail_hook])
def my_task2():
    time.sleep(25)
    return "sdfsdf"


@flow(
    task_runner=ConcurrentTaskRunner(),
)
def my_flow():
    result1 = my_task.submit(return_state=True)
    result2 = my_task2.submit(return_state=True)
    result3 = my_task.submit(return_state=True)
    result4 = my_task2.submit(return_state=True, wait_for=[result1])

hi all, i have a question about whether prefect is the tool I'm looking for. I'll post the description inside the thread - it's on the longer side, looking forward to hearing your thoughts marvin

We are running prefect for 90 days in production. Today we noticed that the Prefect Server is querying Postgresql at a rate of 14MB/s. Grew from 3MB/s linearly over a course of a month. Is this something you are aware of? We nuked the db now and checking if this is related to the number of flows we ran so far.

Hi All, i am a newbie i have setup the dev environment when i am trying to run the command "prefect dev start" i am able to go to the local host link.But i am not table to see any UI it is showing blank page for me. Please help me on this. Thanks,

I am using this in debug mode

Are there examples of using 1 flow to run 2 subflows in Prefect 2.0? The documentation is not very clear on the matter. How do I test it locally? On my firm's local server?

Hi all, has anyone experienced the following issue with the prefect worker utilizing Kubernetes failing to submit flows to the infrastructure after running fine for several hours? A restart is required in order for the worker to function again. Here's the error trace

Failed to submit flow run '8093ef5c-64f8-42eb-9f4b-045ed7e7d6ad' to infrastructure.
Traceback (most recent call last):
  File "/usr/local/lib/python3.11/site-packages/urllib3/connection.py", line 174, in _new_conn
    conn = connection.create_connection(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/urllib3/util/connection.py", line 95, in create_connection
    raise err
  File "/usr/local/lib/python3.11/site-packages/urllib3/util/connection.py", line 85, in create_connection
    sock.connect(sa)
OSError: [Errno 99] Cannot assign requested address

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/local/lib/python3.11/site-packages/urllib3/connectionpool.py", line 714, in urlopen
    httplib_response = self._make_request(
                       ^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/urllib3/connectionpool.py", line 403, in _make_request
    self._validate_conn(conn)
  File "/usr/local/lib/python3.11/site-packages/urllib3/connectionpool.py", line 1053, in _validate_conn
    conn.connect()
  File "/usr/local/lib/python3.11/site-packages/urllib3/connection.py", line 363, in connect
    self.sock = conn = self._new_conn()
                       ^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/urllib3/connection.py", line 186, in _new_conn
    raise NewConnectionError(
urllib3.exceptions.NewConnectionError: <urllib3.connection.HTTPSConnection object at 0x7f60db485950>: Failed to establish a new connection: [Errno 99] Cannot assign requested address

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/local/lib/python3.11/site-packages/prefect/workers/base.py", line 834, in _submit_run_and_capture_errors
    result = await self.run(
             ^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect_kubernetes/worker.py", line 506, in run
    job = await run_sync_in_worker_thread(
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/utilities/asyncutils.py", line 91, in run_sync_in_worker_thread
    return await anyio.to_thread.run_sync(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/anyio/to_thread.py", line 33, in run_sync
    return await get_asynclib().run_sync_in_worker_thread(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/anyio/_backends/_asyncio.py", line 877, in run_sync_in_worker_thread
    return await future
           ^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/anyio/_backends/_asyncio.py", line 807, in run
    result = context.run(func, *args)
             ^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect_kubernetes/worker.py", line 628, in _create_job
    job = batch_client.create_namespaced_job(
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/kubernetes/client/api/batch_v1_api.py", line 210, in create_namespaced_job
    return self.create_namespaced_job_with_http_info(namespace, body, **kwargs)  # noqa: E501
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/kubernetes/client/api/batch_v1_api.py", line 309, in create_namespaced_job_with_http_info
    return self.api_client.call_api(
           ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/kubernetes/client/api_client.py", line 348, in call_api
    return self.__call_api(resource_path, method,
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/kubernetes/client/api_client.py", line 180, in __call_api
    response_data = self.request(
                    ^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/kubernetes/client/api_client.py", line 391, in request
    return <http://self.rest_client.POST|self.rest_client.POST>(url,
           ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/kubernetes/client/rest.py", line 276, in POST
    return self.request("POST", url,
           ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/kubernetes/client/rest.py", line 169, in request
    r = self.pool_manager.request(
        ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/urllib3/request.py", line 78, in request
    return self.request_encode_body(
           ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/urllib3/request.py", line 170, in request_encode_body
    return self.urlopen(method, url, **extra_kw)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/urllib3/poolmanager.py", line 376, in urlopen
    response = conn.urlopen(method, u.request_uri, **kw)
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/urllib3/connectionpool.py", line 826, in urlopen
    return self.urlopen(
           ^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/urllib3/connectionpool.py", line 826, in urlopen
    return self.urlopen(
           ^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/urllib3/connectionpool.py", line 826, in urlopen
    return self.urlopen(
           ^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/urllib3/connectionpool.py", line 798, in urlopen
    retries = retries.increment(
              ^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/urllib3/util/retry.py", line 592, in increment
    raise MaxRetryError(_pool, url, error or ResponseError(cause))
urllib3.exceptions.MaxRetryError: HTTPSConnectionPool(host='172.30.0.1', port=443): Max retries exceeded with url: /apis/batch/v1/namespaces/prefect-2x/jobs (Caused by NewConnectionError('<urllib3.connection.HTTPSConnection object at 0x7f60db485950>: Failed to establish a new connection: [Errno 99] Cannot assign requested address'))

Question regarding a deployment's default parameters. If I run a deployment with

run_deployment

, are only the parameters I submit overridden, or are all parameters overridden and any not explicitly passed fall back on the flow function's defaults?

Hi all. I have been dealing with an issue related to the prefect-ray integration that i am not able to solve. Hope someone here can help me 🙂 I have a prefect agent and Ray cluster running inside my kubernetes cluster. Both seems to work as intended as a standalone. I.e i can submit jobs to the Ray cluster and i can also run simple prefect flows. However, when i try to use the prefect-ray integration and specify resources for a task i.e.

from prefect import flow, task
from prefect_ray.task_runners import RayTaskRunner
from prefect_ray.context import remote_options

@task
def process(x):
    return x + 1


@flow(task_runner=RayTaskRunner("ray://<my-ray-service>:10001"))
def my_flow():
    # equivalent to setting @ray.remote(num_cpus=4)
    with remote_options(num_cpus=4):
        process.submit(42)

It fails with a runtime error:

RuntimeError: There is no current event loop in thread 'ray_client_server_1'.

If i do not specify resources for my task, everything works as intended. I have submitted an issue, but it has not been picked up yet, so i thought maybe i could get some help here 🙂 https://github.com/PrefectHQ/prefect/issues/10542

Has anyone run into this error? Only happens when we use the Dask executor, always at the same spot (right before the first mapped task). Also only happens in kubernetes, not when running locally.

Crash detected! Execution was interrupted by an unexpected exception: PrefectHTTPStatusError: Client error '408 Request Timeout' for url '<https://api.prefect.cloud/api/accounts/><xxx>/workspaces/<xxx>/task_runs/'

Hi all! I noticed that theres a preference for workers over agents now, but workers dont support storage blocks. Does this mean storage blocks are no longer recommended? If I've stored my flows in an s3 bucket, does this mean there's no way to use them with a worker? Should I be doing something completely different? We've got two use cases for flows: one set that runs tasks in a dask task runner that simple data ingestion and transformation, and another use case where we use

prefect-aws

to launch ECS tasks for longer running compute. Im unsure how a "worker" would change or improve these two use cases

Hrm, logging from within a Flow Hook isn't working? (Running on ECS)

def make_fail(flow, flow_run, state):
    print("Failed")
    flow_run_ctx = FlowRunContext.get()
    print(flow_run_ctx)
    print("Flow")
    print(flow)
    print("Flow run")

@flow(
    log_prints=True,
    on_completion=[make_fail],
    on_failure=[make_fail],
    task_runner=ConcurrentTaskRunner(),
    result_serializer=JSONSerializer(),
    result_storage=S3(bucket_path=bucket_path),
    persist_result=True,
)
def my_flow():
    result1 = my_task.submit()
    result2 = my_task2.submit()
    result3 = my_task.submit()
    result4 = my_task2.submit(wait_for=[result1])

None of the

print

statements show up

are there any means to query cached task runs the same way you might in a database?

Hi everyone ! I have recently started using prefect. Currently, I want to do some scraping with prefect as an experiment. My prefect is set up as prefect-server in kubernetes. How can I do the following?　 - Install the package with the configuration described in POETRY. (Or I would like to know the best practice for installing packages.) - Upload chromedriver. (Is it common to upload local files directly or mount persistent volume?)

are there known issues with the

/task-runs/task-run/:uuid

route? • the task inputs tab is showing

{"id": []}

when it should be (for example)

{"id": "123"}

• the cache key in the sidebar is not populated, but I provided a

cache_key_fn

and the caching feature itself seems to be working (re-running task with same ID uses cached result)

hi everyone! i was trying to create deployment which is supposed to run on workpools using worker i want the code to be pulled from aws s3 .. created a deployment yaml for the same but getting Attribute error (nonetype has no attribute name) this is what the yaml looks like

# configuration for deploying your flows. We recommend committing this file to source
# control along with your flow code.

# Generic metadata about this project
name: activities-split-sync
prefect-version: 2.13.3

# build section allows you to manage and build docker images
build: null

# push section allows you to manage if and how this project is uploaded to remote locations
push:
- prefect_aws.deployments.steps.push_to_s3:
    id: push_code
    requires: prefect-aws>=0.3.4
    bucket: prod-prefect-s3-block
    folder: prefect
    credentials: "{{ prefect.blocks.aws-credentials.s3cred }}"

# pull section allows you to provide instructions for cloning this project in remote locations
pull:
- prefect_aws.deployments.steps.pull_from_s3:
    id: pull_code
    requires: prefect-aws>=0.3.4
    bucket: '{{ push_code.bucket }}'
    folder: '{{ push_code.folder }}'
    credentials: "{{ prefect.blocks.aws-credentials.s3cred }}"

# the deployments section allows you to provide configuration for deploying flows
deployments:
- name: activities-mode-split-sync-flow
  version: '1.0'
  tags: [
    activities_mode_split_sync_flow
  ]
  description: null
  schedule: {}
  flow_name: activities-mode-split-sync-flow
  entrypoint: flows/process_sso_integrations_flows/activities_mode_split_sync_flow.py:activities_mode_split_sync_flow
  parameters: {}
  work_pool:
    name: sso
    work_queue_name: activities-sso-queue
    job_variables: {}

I have a bunch of scheduled flows at regular intervals. Is there anyway to control that the subsequent run is only triggered post the successful/failure (not running state) of the previous run. I don't want 2 jobs of the same task running in parallel.

Hi folks - I’m reading the Storage page under Workers and agents. Does Prefect support git repos in Azure DevOps?

hi all - getting a strange behaviour: when running a flow on my (type Process) agent I have 2 stages under

pull

stage 1. cloning from git 2. installing requirements one of my flow requirements is boto3 (and obviously it is in the requirements.txt file) every first flow run after agent's startup fails with this error: "Flow could not be retrieved from deployment."

"exc_info":{"type":"ScriptError","message":"Script at 'flows/full_scan.py' encountered an exception: ModuleNotFoundError(\"No module named 'boto3'\")"

and the second flow onwards succeeds if I install boto3 manually on the agent's terminal it works fine

but if I add this step to pull:
- prefect.deployments.steps.run_shell_script:
    id: force-install-boto3
    script: python3 -m pip install boto3==1.28.0
    stream_output: true

the aforementioned behaviour continues help?

hello Community, is it possible to use just a flow name (without flow id) in trigger configuration in deployments ?

triggers:
      - enabled: true
        match:
          prefect.resource.id: prefect.flow-run.*
        expect:
          - prefect.flow-run.Completed
        match_related:
# This doesn't work prefect.resource.name: prefect.flow.hello_world_flow1
          prefect.resource.id: prefect.flow.24b5f5df-145c-4ce7-bca2-b0159159d4f1
          prefect.resource.role: flow

Are there any means to automatically cancel upcoming runs of a deployment scheduled automatically for the next 2 hours using the prefect API from within a flow that originates from the same deployment?

Hey, I am getting this error when running a deployment using an AWS ECS worker.

RuntimeError: Failed to run ECS task, cluster 'default' not found. Confirm that the cluster is configured in your region.

How can I tell the deployment/prefect the name of the cluster? I really don't know where else to look. Thx

Hi all, having some issues with scheduling. We're using the python SDK to define our schedules and deployments like this

from prefect.client.schemas import schedules
from prefect.client.schemas.schedules import DateTimeTZ

pst = "America/Los_Angeles"

daily_midnight_offset_30 = schedules.IntervalSchedule(
    anchor_date=DateTimeTZ(year=2020, month=1, day=1, hour=0, minute=30), timezone=pst, interval=timedelta(hours=24)
)

But the schedule runs at 17:30 Pacific (00:30 UTC). Do python schedules not respect the time zone?

oh nvm, found what was wrong, the answer is in the docs

i’m not able to find any documentation about the prefect.yaml file in the docs, am i somehow missing it? all of the documents that i see mention it link

prefect.yaml

to places that aren’t it

are the attributes defined in this Deployment python class that same key/value pairs i need to use in prefect.yaml?