Hello, we sometimes see this error in our Prefect flows:

State data is missing. Typically, this occurs when result persistence is disabled and the state has been retrieved from the API.

The code is something like this:

@flow
async def parallel_flow(list_of_org_ids):
    concurrency_limit = asyncio.Semophore(8)
    async def core_run(org_id: str):
        async with concurrency_limit:
            return await run_deployment(
                name="core_flow/us-east-k8s",
                parameters=dict(org_id=org_id)
            )

    results = await asyncio.gather( *[core_run(org_id) for org_id in list_of_org_ids] )
    return [(result.name, result.state_name) for result in results]

I cannot see the reason for the above error since the results array should be resolved and available at the time of the return statement. Are there other causes for this error to occur?

Hey, I suddenly get this error: ERROR - agent | Failed to query for flow run metadata with the Exception: prefect.exceptions.ClientError: [{'path': ['flow_run', 0, 'id'], 'message': 'Cannot return null for non-nullable field flow_run.id.', 'extensions': {'code': 'INTERNAL_SERVER_ERROR'}}] All flows in the Prefect UI are late, and never get allocated to a computer. What should I do? I created this issue: https://github.com/PrefectHQ/prefect/issues/10072

Hello Prefect Community 🙂 I am relatively new to Prefect and am seeking advice on how best to structure my process. To provide some context, I aim to run Prefect within a Docker container, paired with a suitable database, all set up on my server. I have my crawlers/flows ready, and have created a Dockerfile as well as a Docker Compose. My intention is to manage and schedule these containers/flows through Prefect. More specifically, I want Prefect to initiate the respective Docker container at certain times, execute the flow, and then terminate the container. All these 'Flow' containers will be operating on the same server. My main question is about the deployment process: How should I proceed? I am specifically interested to learn from anyone who has had a similar scenario and has created a setup accordingly. I am open to feedback and would greatly appreciate your support in this learning process. Feel free to share your tips, experiences, or any resources that could be beneficial for my use-case. Thank you in advance for your time and consideration.

Is there a way to manage my workerpools default template as well as create custom templates using Projects/Deployments

Hi Prefect Family! I am running flows using kubernetes in prefect 2.0. However, after running the flow in kubernetes, the pod does not change from “Running” status to “Completed” even though the log is completed as shown below. Actually, the pod is not terminated, and if i run

kubectl logs -f <pod_name>

, i can see that the log is not terminated and is continuously waiting for input. can you help me with this issue?

Hello Prefect Community, I posted my error message (when connecting to Prefect UI from CLI) to Prefect UI. Posting the same here to see if someone can help. I am trying to configure Prefect CLI in my entrypoint.sh file - loading the config with all variables like prefect workspace, api key etc. Then when I try to do the following: # Configure prefect CLI to use server backend echo "Configuring Prefect2 CLI..."

prefect cloud login --key "$PREFECT_TOKEN" --workspace "$PREFECT_WORKSPACE"

echo "Done configuring Prefect CLI." Prefect Cloud ENABLED... Successfully parsed PREFECT_TOKEN (not displayed) Configuring Prefect2 CLI... I get the following httpx connection timeout error: Traceback (most recent call last): File "/usr/local/lib/python3.8/site-packages/httpcore/backends/asyncio.py", line 114, in connect_tcp stream: anyio.abc.ByteStream = await anyio.connect_tcp( File "/usr/local/lib/python3.8/site-packages/anyio/_core/_sockets.py", line 192, in connect_tcp gai_res = await getaddrinfo( File "/usr/local/lib/python3.8/site-packages/anyio/_core/_sockets.py", line 511, in getaddrinfo gai_res = await get_asynclib().getaddrinfo( File "/usr/local/lib/python3.8/site-packages/anyio/_backends/_asyncio.py", line 1690, in getaddrinfo result = await get_running_loop().getaddrinfo( File "/usr/local/lib/python3.8/asyncio/base_events.py", line 825, in getaddrinfo return await self.run_in_executor( asyncio.exceptions.CancelledError During handling of the above exception, another exception occurred: Traceback (most recent call last): File "/usr/local/lib/python3.8/site-packages/httpcore/_exceptions.py", line 10, in map_exceptions yield File "/usr/local/lib/python3.8/site-packages/httpcore/backends/asyncio.py", line 121, in connect_tcp stream._raw_socket.setsockopt(*option) # type: ignore[attr-defined] # pragma: no cover File "/usr/local/lib/python3.8/site-packages/anyio/_core/_tasks.py", line 119, in exit raise TimeoutError TimeoutError The above exception was the direct cause of the following exception: Traceback (most recent call last): File "/usr/local/lib/python3.8/site-packages/httpx/_transports/default.py", line 60, in map_httpcore_exceptions yield File "/usr/local/lib/python3.8/site-packages/httpx/_transports/default.py", line 353, in handle_async_request resp = await self._pool.handle_async_request(req) File "/usr/local/lib/python3.8/site-packages/httpcore/_async/connection_pool.py", line 261, in handle_async_request raise exc File "/usr/local/lib/python3.8/site-packages/httpcore/_async/connection_pool.py", line 245, in handle_async_request response = await connection.handle_async_request(request) File "/usr/local/lib/python3.8/site-packages/httpcore/_async/connection.py", line 92, in handle_async_request raise exc File "/usr/local/lib/python3.8/site-packages/httpcore/_async/connection.py", line 69, in handle_async_request stream = await self._connect(request) File "/usr/local/lib/python3.8/site-packages/httpcore/_async/connection.py", line 117, in _connect stream = await self._network_backend.connect_tcp(**kwargs) File "/usr/local/lib/python3.8/site-packages/httpcore/backends/auto.py", line 31, in connect_tcp return await self._backend.connect_tcp( File "/usr/local/lib/python3.8/site-packages/httpcore/backends/asyncio.py", line 121, in connect_tcp stream._raw_socket.setsockopt(*option) # type: ignore[attr-defined] # pragma: no cover File "/usr/local/lib/python3.8/contextlib.py", line 131, in exit self.gen.throw(type, value, traceback) File "/usr/local/lib/python3.8/site-packages/httpcore/_exceptions.py", line 14, in map_exceptions raise to_exc(exc) from exc httpcore.ConnectTimeout The above exception was the direct cause of the following exception: Traceback (most recent call last): File "/usr/local/lib/python3.8/site-packages/prefect/cli/_utilities.py", line 41, in wrapper return fn(*args, **kwargs) File "/usr/local/lib/python3.8/site-packages/prefect/utilities/asyncutils.py", line 260, in coroutine_wrapper return call() File "/usr/local/lib/python3.8/site-packages/prefect/_internal/concurrency/calls.py", line 245, in call return self.result() File "/usr/local/lib/python3.8/site-packages/prefect/_internal/concurrency/calls.py", line 173, in result return self.future.result(timeout=timeout) File "/usr/local/lib/python3.8/concurrent/futures/_base.py", line 437, in result return self.__get_result() File "/usr/local/lib/python3.8/concurrent/futures/_base.py", line 389, in __get_result raise self._exception File "/usr/local/lib/python3.8/site-packages/prefect/_internal/concurrency/calls.py", line 218, in _run_async result = await coro File "/usr/local/lib/python3.8/site-packages/prefect/cli/cloud.py", line 391, in login workspaces = await client.read_workspaces() File "/usr/local/lib/python3.8/site-packages/prefect/client/cloud.py", line 72, in read_workspaces return pydantic.parse_obj_as(List[Workspace], await self.get("/me/workspaces")) File "/usr/local/lib/python3.8/site-packages/prefect/client/cloud.py", line 92, in get res = await self._client.get(route, **kwargs) File "/usr/local/lib/python3.8/site-packages/httpx/_client.py", line 1757, in get return await self.request( File "/usr/local/lib/python3.8/site-packages/httpx/_client.py", line 1530, in request return await self.send(request, auth=auth, follow_redirects=follow_redirects) File "/usr/local/lib/python3.8/site-packages/httpx/_client.py", line 1617, in send response = await self._send_handling_auth( File "/usr/local/lib/python3.8/site-packages/httpx/_client.py", line 1645, in _send_handling_auth response = await self._send_handling_redirects( File "/usr/local/lib/python3.8/site-packages/httpx/_client.py", line 1682, in _send_handling_redirects response = await self._send_single_request(request) File "/usr/local/lib/python3.8/site-packages/httpx/_client.py", line 1719, in _send_single_request response = await transport.handle_async_request(request) File "/usr/local/lib/python3.8/site-packages/httpx/_transports/default.py", line 353, in handle_async_request resp = await self._pool.handle_async_request(req) File "/usr/local/lib/python3.8/contextlib.py", line 131, in exit self.gen.throw(type, value, traceback) File "/usr/local/lib/python3.8/site-packages/httpx/_transports/default.py", line 77, in map_httpcore_exceptions raise mapped_exc(message) from exc httpx.ConnectTimeout An exception occurred. Failed to register flows At C\WorkProject\dbt CustomerInsights LifeSciences\deploy\deploy dockerfile dev.ps161 char:27 + if (-not $?) {throw "Failed to register flows"} + ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + CategoryInfo : OperationStopped: (Failed to register flows:String) [], Run timeException + FullyQualifiedErrorId : Failed to register flows

[3:51 PM] However, if I run the same command in the same powershell terminal with the actual token and workspace, it is connecting to prefect cloud without any issues.                                                                                                                                                                                                                                                                  prefect cloud login --key "$PREFECT_TOKEN" --workspace "$PREFECT_WORKSPACE"

[3:52 PM] Can you please help me with this connection timeout issue? Using Python 3.8, Prefect 2.0 cloud [3:53 PM] I do run this in my venv. [2:58 PM] Can you please suggest some fixes for this httpx connection timeout error?

Hi everyone, I have an existing streaming lambda function that I deployed using prefect and git actions [ serverless pipeline] my lambda function is triggering prefect flow as it receives event. Now I am adding enhancement to this code where I need to do a data quality check by running queries against data in redshift. So in order for my lambda function to connect to redshift over private VPC I added VPC and subnet to my function , after this I am facing an error where my lambda is no more able to connect to prefect API over internet. Its giving me timeout error. How can I make my lambda function connect to Prefect API and trigger flow and also connect to my redshift cluster over VPC?

Hello, Prefect webhooks and events is only for Prefect Cloud? Are we being left behind if we self-host

Hi, what would be the recommended setup for setting up a data pipeline where a flow run is triggered after another flow run has finished?

Hi I am not receiving activation link to sign up for community version(prefect 2), tried multiple times by providing email, can someone from Prefect help? Thank you

hey - I cant find anything in the docs, do you know if the JSON block has any limit to the size / keys ? if so what is it?

Is there a way to ensure that a new definition of a Task or Flow overwrites an old one? I'm using Jupyter to tinker with a Task, and if I make a tweak and re-run, it gives me this and the new behavior doesn't happen.

UserWarning: A task named 'download_image' and defined at '/tmp/ipykernel_409/4031849273.py:14' conflicts with another task. Consider specifying a unique `name` parameter in the task definition:

I've been using variations on this to get around it:

@task(
    name=f'my_task-{datetime.datetime.now().strftime("%Y%m%d%H%M%S%s%f")}',
)

but it's a pretty annoying bit of boilerplate to have to reach for every time.

hi, I am using ecs task as infra. I want to be able to override the name of the container inside the task. Seems the default is “prefect” which is not very informative. Ideally I would override it when I do

prefect deployment build

by passing

--override

as described here. I tried to pass

--override name="my_task"

and while I see it in the deployment overrides in the prefect UI, it has no impact on the actual ECS task json. Any idea?

Hello community, Does anyone know how to remove an old agent in Prefect UI 1.0? No luck so far in the UI nor graphql. Seems like the old agent has been acting out and interrupting with the active agent as no flows are submitted.

Hello, I recently tried to move my current setup (Server + Worker) from EC2 to a fully self managed ECS cluster elaborating on a Terraform template provided here. The goal here is to hide the server behind a load balancer. However, I have a hard time figuring out how to configure the server so that the UI remain accessible. I pretty much tried all combinations of the following environments variables in my ECS task definition :

command = ["prefect", "server", "start", "--host", "0.0.0.0"]
      environment = [
        {
          name = "PREFECT_UI_API_URL",
          value = "http://${aws_lb.prefect_ecs_alb[0].dns_name}:4200/api"
          #value = "<http://0.0.0.0:4200/api>"
        },
        {
          name = "PREFECT_API_URL",
          value = "http://${aws_lb.prefect_ecs_alb[0].dns_name}:4200/api"
          #value = "<http://0.0.0.0:4200/api>"
        },
        {
          name = "PREFECT_SERVER_API_HOST",
          value = "${aws_lb.prefect_ecs_alb[0].dns_name}",
          #value = "0.0.0.0"
        },
        {
          name = "PREFECT_UI_URL",
          value = "http://${aws_lb.prefect_ecs_alb[0].dns_name}:4200"
          #value = "<http://0.0.0.0:4200>"
        }
      ],

I keep getting

Can't connect to server api

errors or a black screen. It seems I can reach the server from my local development machine though. Is there any ressource available for this kind of issue?

Bump... would truly appreciate any input on this one.

hey, I try to install the prefect_test_harness package, and i get the error that the package is didn't found in pypi repository.

Hi, while using Prefect v1 we needed to identify a user's email in-flight for a flow they triggered, the solution required a GraphQL query. I'm now migrating the same flow to Prefect v2 and wanted to check if the way forward is to use the REST API to fetch the user info, or if there is another way to do this? Thank you

I've been testing Prefect Projects recently and am loving it! It solves a lot of tedious work involved with using the old deployment method. My question is, I know it's in BETA still and does anyone with inside knowledge know if it would be foolish or not to start using it for production work loads? The BETA tag kind of scares me.

Hello community, I was wondering if anyone could shed a light into how can one mock a task run (in prefect 1.x) when testing flow run-end-to-end. I’m aware of this. But I could not decipher how tod o it properly even in a simple case. Given a simple flow definition:

from prefect import Flow, task, Parameter

@task
def add(a, b):
    return a + b

with Flow("test") as flow:
    a = Parameter(name="a")
    b = Parameter(name="b")
    
    add(a,b)

how does write a test case to mock

add()

task? What I’d normally do would be this: but this does not work

@patch.object(flow_module.add, "run")
    def test_foo(self, mock):
        flow_module.flow.run(parameters={"a": 1, "b": 2})

        assert mock.call_count == 1

Advices, suggestions? Please dont say move to prefect 2.0 🙂 Thank you!

hey for some reason when calling the prefect cloud api I get authenticated but not authorized. Just following this: https://docs.prefect.io/2.10.17/api-ref/rest-api/. For example I can call api/me but not flow_runs/filter even though the account id and workspace id are correct. Maye its an issue where the key has to be given permission to the workspace but I didnt see anywhere to do that in the UI. Anyone have this issue?

hello, can someone help me with this prefect.exceptions.ScriptError: Script at 'flows/generate_features_from_response.py' encountered an exception: TypeError("task() got an unexpected keyword argument 'result_storage_key'") but according to the docs we can pass result_storage_key https://docs.prefect.io/2.10.17/concepts/results/#result-storage-key

Hi, I am stuck with an issue when running

dbt test

with

trigger_dbt_cli_command

. Hope someone can share some insights. In prefect 1, I used

DbtShellTask

to run

dbt test

, it always returns a full log even when some tests fail. This log is used to populate a performance tracking table, which is important to us. Now in prefect 2, we have to switch to

trigger_dbt_cli_command

, and this, when any test fails, return RuntimeError (a snippet in comment section). If I put the task in a try/except block, the flow can continue, but still, I am not able to get the log.

Has anyone run into and solved the issue of

prefect deploy path\to\file.py:flow_name

Throwing a

AttributeError: 'NoneType' object has no attribute 'name'

error? I'm using windows 10 powershell, my python version is 3.11.3, and prefect version is 2.10.14. Updating from prefect 2.10.14 to 2.10.17 the error changes to

ModuleNotFoundError: No module named 'resource'

Here's my super basic flow -

import httpx
from prefect import flow, task


@task
def fetch_cat_fact():
    return httpx.get("<https://catfact.ninja/fact?max_length=140>").json()["fact"]


@task
def formatting(fact: str):
    return fact.title()


@task
def write_fact(fact:str):
    with open("fact.txt", "w+") as f:
        f.write(fact)
    return "Success!"


@flow(name='pipe')
def pipe():
    fact = fetch_cat_fact()
    formatted_fact = formatting(fact)
    msg = write_fact(formatted_fact)
    print(msg)

I've seen some others who ran into the issue.

I decided to try other versions of Prefect, and things work as expected in 2.10.16. I think the error from when I ran 2.10.17 is unrelated, but not positive.

Hi All, I need help on prefect-sqlalchemy. I am using its execute() and execute_many() method to perform DML operations on the database. When I run my flow from Python script on local, everything is working fine. But when I ran it inside docker container, DML operations are successful in the logs but changes are not reflecting in the Postgres database. However, when I use fetch_all() to select the records I am successfully able to get the records from the database. Has anybody faced such issues? I am using following versions

psycopg2-binary == 2.9.6
prefect-sqlalchemy == 0.2.4

Driver : POSTGRESQL_PSYCOPG2

Hi 🙂 I am currently trying to get Prefect Server up and running. Would like to run my flows separately in a Docker container. However, when I try to add an infrastructure block I get the following error (see picture). Does anyone have any idea ? I used the following Docker-Compose: https://github.com/rpeden/prefect-docker-compose/blob/main/docker-compose.yml But only with Database, Server and one Agent. Greets

Hey, I'm trying to create a remote deployment using the following

import os
from prefect.deployments import Deployment
from prefect.filesystems import LocalFileSystem

os.environ['PREFECT_API_URL'] = f'http://<remote_host>:4200/api'
local_file_system_block = LocalFileSystem.load('local-system')
deployment = Deployment.build_from_flow(...)
deployment.apply

However find that it doesn't seem to utilise the environ variable and instead hits the ephemeral-prefect host rather than the <remote_host> is this expected? setting this by

prefect config set PREFECT_API_URL=http://<remote_host>:4200/api

works however

Facing below error when setting up prefect server:

Traceback (most recent call last):
  File "/home/infra/prefect_server/bin/docker-compose", line 8, in <module>
    sys.exit(main())
  File "/home/infra/prefect_server/lib/python3.8/site-packages/compose/cli/main.py", line 80, in main
    command_func()
  File "/home/infra/prefect_server/lib/python3.8/site-packages/compose/cli/main.py", line 189, in perform_command
    project = project_from_options('.', options)
  File "/home/infra/prefect_server/lib/python3.8/site-packages/compose/cli/command.py", line 60, in project_from_options
    return get_project(
  File "/home/infra/prefect_server/lib/python3.8/site-packages/compose/cli/command.py", line 152, in get_project
    client = get_client(
  File "/home/infra/prefect_server/lib/python3.8/site-packages/compose/cli/docker_client.py", line 41, in get_client
    client = docker_client(
  File "/home/infra/prefect_server/lib/python3.8/site-packages/compose/cli/docker_client.py", line 170, in docker_client
    client = APIClient(use_ssh_client=not use_paramiko_ssh, **kwargs)
  File "/home/infra/prefect_server/lib/python3.8/site-packages/docker/api/client.py", line 197, in __init__
    self._version = self._retrieve_server_version()
  File "/home/infra/prefect_server/lib/python3.8/site-packages/docker/api/client.py", line 221, in _retrieve_server_version
    raise DockerException(
docker.errors.DockerException: Error while fetching server API version: ('Connection aborted.', FileNotFoundError(2, 'No such file or directory'))
Traceback (most recent call last):
  File "/home/infra/prefect_server/lib/python3.8/site-packages/prefect/cli/server.py", line 623, in start
    subprocess.check_call(
  File "/usr/local/lib/python3.8/subprocess.py", line 364, in check_call
    raise CalledProcessError(retcode, cmd)
subprocess.CalledProcessError: Command '['docker-compose', 'pull']' returned non-zero exit status 1.

Hi Everyone, Can someone correct me if I am understanding it properly? When a flow crashes, it means its a infrastructure error? I have deployed on ECS Fargate with CPU: 512 and Memory: 1024. Do I need to increase both to avoid crashes in future? The only error statement i see:

State message: Flow run infrastructure exited with non-zero status code 137.

Any help would really be appreciated !