I'm trying to attach job variables to a specific deployment run using the UI. When I use

Variable.get("var")

it's null when the flow runs. My job variable for that deployment is

{ "var":"test" )

. Is there a different approach I should take?

Hi everyone, Can someone help me with adding dynamic tags for the task? Also, how to set different dynamic tags for different flow within a same script, or a subflow within a flow. I want to have different dynamic tags to each subflow. Thank you so much in advance for this help.

@Marvin How do I upgrade from the default SQLite DB to Postgres as seamlessly as possible?

@Marvin I'm trying to set things up locally completely self-hosted. I have • the server running (

prefect server start

) • i have a worker running (

PREFECT_API_URL=<http://127.0.0.1:4200/api> API_prefect worker start -pool=my-pool

) • i am deploying my job using python like so

upload_deploy = <http://upload_flow.to|upload_flow.to>_deployment("upload-file-deploy", work_pool_name="my-pool")

(and i have a flow decorated function called upload_flow) • i am now calling another python script

run_deployment("upload-flow/upload-file-deploy")

the worker receives the job but it crashes like this:

10:13:51.901 | ERROR   | Flow run 'dazzling-ferret' - Unexpected exception encountered when trying to load flow

FileNotFoundError: [Errno 2] No such file or directory: '/private/var/folders/9b/9dclgrxj76172c20y_qrb9tw0000gn/T/tmp74n9b6voprefect/main.py'

what am i doing wrong?

@Marvin can you explain to me how async works within a flow and tasks..specifically regarding example inside an async function why would we not use await for futures. task_future = send_audit_log_query.submit( _report_generator_=report_generator, client=client, _display_name_=display_name, _request_body_=request_body, ) await task_futures.append((idx, date_range, task_future)) # <- Instead the documentation is mentioning to not use await, but I think that's causing a block in code. 212642.672 | ERROR | asyncio - Critical error in *: Event loop is closed 212642.672 | ERROR | Flow run 'busy-mastodon' - Encountered exception during execution: RuntimeError('Event loop is closed') Traceback (most recent call last): File "/usr/local/lib/python3.13/site-packages/prefect/flow_engine.py", line 1299, in run_context yield self File "/usr/local/lib/python3.13/site-packages/prefect/flow_engine.py", line 1357, in run_flow_async await engine.call_flow_fn() File "/usr/local/lib/python3.13/site-packages/prefect/flow_engine.py", line 1316, in call_flow_fn result = await call_with_parameters(self.flow.fn, self.parameters) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/app/***/src/main.py", line 146, in ** async with AsyncClient() as client: ~~~~~~~~~~~^^

What are we supposed to do when flows or tasks are stuck in a non-terminal state? Like this is blows up the amount of data in the database since it keeps writing a states even though the runs are from 2 months ago and there's literally no infra for them anymore. Example: Flow spins up tasks, flow crashes due to memory. Tasks show as still running in dashboard, and database still shows running state continually being written...

Hey all, we ran into a production issue today with Prefect, where the workers were all suddenly "late". The workers seem to be working but they werent polling (seems to be hanging). When I press enter in the cmd window (hosting the worker), it all of a sudden starts working. But without it, it's all frozen. What can be causing this? How can we make it robust? Using Prefect 2.18.3 Server

@Marvin how do I trigger a flow run based on the completion of a previous flow run?

Hi, I'm completely new to prefect, I'm currently testing it as a way to run simple pipelines on azure container instances. I'm currently trying to run a snippet from the azure tutorial

from prefect import flow, get_run_logger, task


@task
def log_hello(name="Marvin"):
    logger = get_run_logger()
    logger.info(f"{name} said hello!")
    return name


@flow
def log_hello_flow():
    for name in ["Marvin", "Alice", "Bob"]:
        log_hello(name)


if __name__ == "__main__":
    # log_hello_flow.deploy(
    #     name="local-deployment",
    #     work_pool_name="local-docker-pool",
    #     image="localhost:5000/prefect-test",
    # )

    log_hello_flow.deploy(
        name="my-first-aci-deployment",
        work_pool_name="my-aci-work-pool",
        image="XXX.azurecr.io/prefect-test",
    )

now when I run it with the local docker pool (commented out section) it seems to be working fine. But when I try to run it on ACI I'm getting

Completed submission of flow run '9a62b3a4-c7ff-4d28-96b5-9769694a03d9'
15:25:21.332 | INFO    | prefect.worker.azure-container-instance.azurecontainerworker 4378b7ad-b02b-4c43-b1b9-0b8609a74da9 - AzureContainerInstanceJob 'AzureContainerWorker 4378b7ad-b02b-4c43-b1b9-0b8609a74da9': Container deleted.
15:25:21.374 | INFO    | prefect.flow_runs.worker - Reported flow run '9a62b3a4-c7ff-4d28-96b5-9769694a03d9' as crashed: Flow run infrastructure exited with non-zero status code 1.

The run ends up in a failed state, but I can't find any more info on the issue, neither in prefect nor in azure. On azure activity log I see no failed tasks, the only seeming relevant information there is the following. So the credentials seem fine, as it responds to flow run by creating some container groups but that's it. The image got pushed to ACR as I see it listed there.

Im doing some async requesting with asyncio and even though im handling retries and not throwing exceptions im getting

Crash detected! Execution was cancelled by the runtime environment.

with no context no logging nothing. Ill reply with the function in question its pretty benign, anyone have any tips on how to diagnose