Hello! Currently Variables values are strings only, is there an option to set other values from different types? list, dict, etc...

Hello, I am trying to send a MS teams notification along with an artifact table . How to do that in prefect 2 ? I am using prefect.blocks.notifications.MicrosoftTeamsWebhook to send notifcation

Hey team, While executing flows using subprocess on an agent/worker (and ConcurrentTaskRunner), what's the max no of concurrent flows I can run ? From what I understand, each sub-process can only run one flow at a time which means the event-loop (of that process) is shared only by tasks of that flow. So assuming the agent is running on a 8vCPU machine, can I only have about 16 flow runs in parallel ?

Hello, I'm on latest prefect 2.10.6, and I have a lot of tasks that hang indefinitely, I'm running the tasks in parallel inside a flow with a tag concurrency limit and a timeout of 60 seconds set for each task, but it still hangs indefinitely until I kill it, more details in the thread ⬇️

hello, is it possible to send flows to the worker without waiting for their completion? I'm trying to trigger flows/deployments from a web process and I don't want to keep it in the loop until it's finished working. any advice?

Hey all, has anyone encountered this error before while trying to run a flow in a pytest?

AttributeError: 'State' object has no attribute 'services'

Whats the best way to handle graceful shutdown of a flow? I am using the parent-child deployed flow pattern but noticed that some of my flows are still running if the parent is canceled/fails. I want to catch any termination signals and ensure that those child flows also get canceled. Any advice?

Hey, I’m trying to create a flow run using the API, the flow is created but the agents do not pick it - how do I add a queue and pool name to the request?

curl --location --request POST '<https://api.prefect.cloud/api/accounts/0aceff1b-4fdb-4469-adde-197148a98672/workspaces/b7474c31-4033-449b-96b6-f3544e6f9408/flow_runs/>' \
--header 'Content-Type: application/json' \
--header 'Authorization: Bearer api_key' \
--header 'X-PREFECT-API-VERSION: 0.8.4' \
--data-raw '{
    "name": "testing_api_1",
    "flow_id": "46a211a5-a943-4547-8b4d-8e3c09da68d2",
    "deployment_id": "8507e60e-0a93-43c0-81ba-bd050f0ecd53", "infrastructure_document_id": "5ed25162-53fb-4df5-89bc-426afdc83f41", "state": {"type": "SCHEDULED"}
}'

I looked here https://app.prefect.cloud/api/docs#tag/Flow-Runs - create flow run - but I don’t see any queue and pool params. Thanks

Q on Pytest using

prefect_test_harness

fixture: How to unit test with the fixture, when flows depend on blocks? Do I have to register all my blocks for the tests? For secret key blocks this is very inconvenient.

Our work queues are getting backed up because the Prefect agent is failing silently and not restarting in the Kubernetes Helm deployment. This is the error that we're seeing.

Traceback (most recent call last):
  File "/usr/local/lib/python3.10/site-packages/prefect/_internal/concurrency/services.py", line 120, in _run
    async with self._lifespan():
  File "/usr/local/lib/python3.10/contextlib.py", line 199, in __aenter__
    return await anext(self.gen)
  File "/usr/local/lib/python3.10/site-packages/prefect/events/worker.py", line 30, in _lifespan
    async with self._client:
  File "/usr/local/lib/python3.10/site-packages/prefect/events/clients.py", line 118, in __aenter__
    await self._reconnect()
  File "/usr/local/lib/python3.10/site-packages/prefect/events/clients.py", line 136, in _reconnect
    self._websocket = await self._connect.__aenter__()
  File "/usr/local/lib/python3.10/site-packages/websockets/legacy/client.py", line 637, in __aenter__
    return await self
  File "/usr/local/lib/python3.10/site-packages/websockets/legacy/client.py", line 655, in __await_impl_timeout__
    return await self.__await_impl__()
  File "/usr/local/lib/python3.10/site-packages/websockets/legacy/client.py", line 659, in __await_impl__
    _transport, _protocol = await self._create_connection()
  File "/usr/local/lib/python3.10/asyncio/base_events.py", line 1036, in create_connection
    infos = await self._ensure_resolved(
  File "/usr/local/lib/python3.10/asyncio/base_events.py", line 1418, in _ensure_resolved
    return await loop.getaddrinfo(host, port, family=family, type=type,
  File "/usr/local/lib/python3.10/asyncio/base_events.py", line 863, in getaddrinfo
    return await self.run_in_executor(
  File "/usr/local/lib/python3.10/asyncio/base_events.py", line 821, in run_in_executor
    executor.submit(func, *args), loop=self)
  File "/usr/local/lib/python3.10/concurrent/futures/thread.py", line 167, in submit
    raise RuntimeError('cannot schedule new futures after shutdown')
RuntimeError: cannot schedule new futures after shutdown

Not sure where this error is coming from, but I would expect any error that terminates the Prefect agent should trigger a pod restart. I checked the Helm chart and I don't see a liveness probe there. Is this something that the Prefect team can add? I assume the blocker is that the Prefect agent may not have a good way to check that it's still alive.

Hi, I would like to link each running agent/flow run to infrastructure, say I have fixed number of VMs in cloud, and I would like my flow(s) to take one VM (from predefined pool of VMs), start it, send task to be executed and stop it inside flow, is there an ability to get somehow assign unique label/tag when creating agent and get this in flow from runtime/context, to map to VMs instance-name/IP after? I’m not sure that this is the best way of using Prefect, probably there might be another solution with it, that I just don’t see (due to the lack of experience with tool)

Is there anything built in to Prefect flows where a certain threshold of tasks needs to fail before considering the entire flow a failure? I see I have the means to implement that myself but just checking…

Hello, I have the following use case: I have a catalog of Prefect flows and I need to be able to define a DAG with a subset of them (as yaml/json), where each node represents a flow with its required input arguments etc. The DAG file is passed to an orchestrator flow that parses it and runs each flow in Prefect as a subflow. Is this a valid use case for Prefect? Is there any library/specifcation to define such a DAG?

For Prefect 1: Does anyone know if deleting and recreating the ECS service that is our ECS Agent will work or will we need to reconnect it to Prefect Cloud somehow? We want to add labels to the agent but that will require us to delete and recreate the service

@prefect Can someone share me good resources to get started in prefect2.0 statehandler

Hello, still somewhat new to Prefect and still migrating jobs over to 2.0. Why sometimes does the Deployments page not show on the site? Agent is running, flows are executing.

image.png

Hi all, where can I access the prefect 1.0 docs (I realize migration needs to be done by July)

Hi Prefect team and especially @justabill, I have a question about the capabilities of the

on_crashed

parameter just released (!). Is is possible to pass in additional arguments to function so invoked, using objects generated during the course of a flow? Like so:

def cleanup(path_name: str):
    shutil.rmtree(path_name)

def crash_hook(path_name, flow, flow_run, state):
    cleanup(path_name)

@flow(on_crashed=[crash_hook])
def my_flow():
    # stuff
    path_name = other_function()

My use case is wiping very large amounts of data downloaded for temporary processing as a part of my ETLs. Leaving this lying around on temporary storage is expensive and best avoided. The path name for the directory to wipe depends on the ETL so invoked and so a static function won’t cut it; I need to vary the path I apply

shutil.rmtree

to. Thanks for pushing this awesome feature, it directly responds to a problem I was just about to address. Very cool!

Hi, We recently upgraded to enterprise and created a lot more deployments. Suddenly the deployment tab on the left hand side of prefect-cloud is gone. Am I doing this right?

I’m seeing the following error on my workers, they are continuously crashing. This deployment actually doesn’t exist.

raise PrefectHTTPStatusError.from_httpx_error(exc) from exc.__cause__
prefect.exceptions.PrefectHTTPStatusError: Client error '404 Not Found' for url '<http://prefect.mls.gh.st:4200/api/deployments/762eab44-76ca-44ee-8334-f2a5d46d4883>'
Response: {'detail': 'Deployment not found'}
For more information check: <https://httpstatuses.com/404>
An exception occurred.

Hi, I'm getting a

RecursionError: maximum recursion depth exceeded in comparison

very similar to this (unanswered) report from @Austin Weisgrau a few days ago. Full stack trace in thread. Could someone from the Prefect team take a look? (also, we're paying customers, not sure if that entitles us to raise support tickets beyond here and/or github somehow?)

For prefect flows configured in a django application, if i were to add a sigterm handler where should it be added? Tried adding inside the flow definition and got an error saying only valid in the main thread

Hi everyone, Is it possible to remove a result which is cached in memory? I have a task

A

which produces a result which I don't want to persist but just keep in memory. This result is consumed by multiple invocations of task

B

. Once all invocations of task

B

are done I don't need the result of task

A

anymore and would like to remove it from the cache to free up some memory. How could I achieve this? Thanks!

Anyone else having issues with prefect agents telling flows to delete themselves while they're running in 2.10.7? Maybe it's an issue specific to azure container instances.. Never seen it before and it keeps happening today.

Was there a breaking change recently with automations? Since April 17th I haven't received any notifications from crashed flows, even though I see some in the UI.

What’s the easiest way to fetch the logs of a flow run, with the intention of sending them to another application? I’ve searched far and wide, but there doesn’t seem to be any docs or conclusive discussion on this 🤔

Hi Prefect Community, I just upgraded from 2.10.4 to 2.10.7. I used to use

from prefect.packaging.docker import DockerPackager
docker = DockerPackager(
        dockerfile=str(Path("Dockerfile").resolve()),
        image_flow_location=flow_path,
        registry_url=config.get("AZURE_REGISTRY", "DOCKER_REGISTRY_URL"),
    )
docker_manifest = await docker.package(flow=flow)

to build and push images and use the manifest on kubernetes

infra_overrides

. However the

package

method is now raising issue.

File "/home/dcvs/source_code/archimedes_etl_flows/.venv/lib/python3.10/site-packages/docker/api/client.py", line 221, in _retrieve_server_version
    raise DockerException(
docker.errors.DockerException: Error while fetching server API version: HTTPConnection.request() got an unexpected keyword argument 'chunked'

Hello, How do we filter flows by flow tags in the new UI in 2.10.7?

My prefect worker is continuously crashing with the following error:

raise PrefectHTTPStatusError.from_httpx_error(exc) from exc.__cause__
prefect.exceptions.PrefectHTTPStatusError: Client error '404 Not Found' for url '<http://prefect.mls.gh.st:4200/api/deployments/762eab44-76ca-44ee-8334-f2a5d46d4883>'
Response: {'detail': 'Deployment not found'}
For more information check: <https://httpstatuses.com/404>
An exception occurred.

That deployment doesn’t actually exist. Do I need to do some cleaning in the database?