Deployment.build_from_flow()

prefect deployment build

Deployment.build_from_flow()

prefect deployment build

prefect deploy

flow.from_source(...).deploy(...)

build/apply

python
my_flow.from_source(
    source="<https://github.com/org/repo>",
    entrypoint="flows.py:my_flow",
).deploy(
    name="prod",
    work_pool_name="my-pool",
)

prefect.toml

pyproject.toml

.env

prefect.yaml

bash
   pip install -U prefect

bash
   pip install -U prefect-aws prefect-gcp prefect-kubernetes prefect-docker

bash
   prefect server database upgrade

prefect deploy

flow.from_source(...).deploy(...)

Deployment.build_from_flow()

Deployment(...)

KubernetesJob.load(...)

job_variables

python
from prefect.deployments import Deployment
from prefect.infrastructure.kubernetes import KubernetesJob

bash
pip install -U prefect prefect-kubernetes

bash
pip install -U prefect prefect-kubernetes prefect-aws

Deployment.build_from_flow()

flow.from_source(...).deploy(...)

flow.deploy(...)

*If code is baked into the image*
If your flow code is already in the execution image, you may use:
```python
from prefect import flow

@flow
def my_flow():
    print("hello")

if __name__ == "__main__":
    my_flow.deploy(
        name="my-flow",
        work_pool_name="eks-pool",
        image="<http://123456789012.dkr.ecr.us-east-1.amazonaws.com/my-image:latest|123456789012.dkr.ecr.us-east-1.amazonaws.com/my-image:latest>",
    )

namespace

image

image_pull_policy

service_account_name

env

labels

annotations

finished_job_ttl

cpu

memory

infra_overrides

Deployment.build_from_flow()

prefect-kubernetes

prefect-aws

flow.from_source(...).deploy(...)

flow.deploy(...)

job_variables

job_variables

python
my_flow_instance = my_flow.from_source(...)
deployment_id = my_flow_instance.deploy(...)

.from_source(...)

.deploy(...)

job_variables

from_source(...)

deploy(...)

work_pool_name

image

job_variables

python
my_flow.from_source(...).deploy(...)

Deployment.build_from_flow()

python
my_flow_instance = my_flow.from_source(...)
deployment_id = my_flow_instance.deploy(...)

bash
pip install -U prefect prefect-kubernetes prefect-aws

python
from_source(...).deploy(...)

.from_source(...)

.deploy(...)

from_source(...).deploy(...)

prefect-kubernetes

prefect-aws

.from_source(...)

prefect-kubernetes

prefect-aws

from_source(...)

my_flow.from_source(...)

source=

entrypoint=

.from_source(...)

deploy(...)

.deploy(...)

name=

work_pool_name=

work_queue_name=

job_variables=

image=

job_variables

text
prefecthq/prefect:3-latest-python3.12-kubernetes

3-latest

text
prefecthq/prefect:3.7.5-python3.12-kubernetes

-kubernetes

prefecthq/prefect:3-latest-python3.12-kubernetes

prefecthq/prefect:<exact-version>-python3.12-kubernetes

text
prefecthq/prefect:3.7.5-python3.12-kubernetes

prefect-aws

boto3

dockerfile
FROM prefecthq/prefect:3.7.5-python3.12-kubernetes

RUN pip install --no-cache-dir prefect-aws boto3

COPY . /opt/prefect/my-project
WORKDIR /opt/prefect/my-project

text
prefecthq/prefect:<pinned-3.x-version>-python3.12-kubernetes

pydantic

pydantic-core

pydantic

pydantic-core

pydantic

text
pydantic>=2

text
pydantic-core

python
from pydantic import BaseModel, ConfigDict

class MyModel(BaseModel):
    model_config = ConfigDict(arbitrary_types_allowed=True)

python
from pydantic import BaseModel

class MyModel(BaseModel):
    class Config:
        arbitrary_types_allowed = True

pydantic

pydantic-core

pydantic>=2

pydantic-core

text
pydantic>=2,<3

.deploy(parameters=...)

BaseModel

class Config

@validator

@root_validator

.dict(

.json(

parse_obj

from_orm

python
from pydantic import BaseModel

class Params(BaseModel):
    x: int

    class Config:
        arbitrary_types_allowed = True
        orm_mode = True

python
from pydantic import BaseModel, ConfigDict

class Params(BaseModel):
    x: int

    model_config = ConfigDict(
        arbitrary_types_allowed=True,
        from_attributes=True,
    )

class Config

model_config = ConfigDict(...)

orm_mode = True

from_attributes=True

python
from pydantic import BaseModel, validator

class Params(BaseModel):
    name: str

    @validator("name")
    def validate_name(cls, v):
        return v.strip()

python
from pydantic import BaseModel, field_validator

class Params(BaseModel):
    name: str

    @field_validator("name")
    @classmethod
    def validate_name(cls, v):
        return v.strip()

@validator

@field_validator

@root_validator

@model_validator

values

@classmethod

python
from pydantic import BaseModel, root_validator

class Params(BaseModel):
    start: int
    end: int

    @root_validator
    def check_range(cls, values):
        if values["end"] < values["start"]:
            raise ValueError("end must be >= start")
        return values

python
from pydantic import BaseModel, model_validator

class Params(BaseModel):
    start: int
    end: int

    @model_validator(mode="after")
    def check_range(self):
        if self.end < self.start:
            raise ValueError("end must be >= start")
        return self

@root_validator

@model_validator

mode="before"

mode="after"

.dict()

.json()

.model_dump()

.model_dump_json()

python
params = Params(name="abc")

payload = params.model_dump()
payload_json = params.model_dump_json()

.dict()

.model_dump()

.json()

.model_dump_json()

python
Params.parse_obj(data)

python
Params.model_validate(data)

python
Params.from_orm(obj)

python
Params.model_validate(obj)

python
model_config = ConfigDict(from_attributes=True)

parse_obj

model_validate

from_orm

model_validate(... )

from_attributes=True

python
from prefect import flow
from pydantic import BaseModel

class Params(BaseModel):
    customer_id: int
    dry_run: bool = False

@flow
def my_flow(params: Params):
    ...

python
my_flow.deploy(
    name="prod",
    work_pool_name="eks-pool",
    parameters={
        "params": Params(customer_id=123).dict()
    },
)

python
my_flow.deploy(
    name="prod",
    work_pool_name="eks-pool",
    parameters={
        "params": Params(customer_id=123).model_dump()
    },
)

Literal

json_encoders

Model(...)

Model.model_validate({...})

Model.model_dump()

python
from prefect import flow
from pydantic import BaseModel, ConfigDict

class Params(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    customer_id: int
    dry_run: bool = False

@flow
def test_flow(params: Params):
    print(params)

if __name__ == "__main__":
    p = Params.model_validate({"customer_id": 123})
    print(p.model_dump())
    test_flow(p)

class Config

@validator

@root_validator

.dict()

.json()

parse_obj

from_orm

python
from pydantic import BaseModel, validator

class RunConfig(BaseModel):
    customer_id: int
    env: str

    class Config:
        orm_mode = True

    @validator("env")
    def validate_env(cls, v):
        return v.lower()

python
from pydantic import BaseModel, ConfigDict, field_validator

class RunConfig(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    customer_id: int
    env: str

    @field_validator("env")
    @classmethod
    def validate_env(cls, v):
        return v.lower()

python
parameters = {
    "config": RunConfig(customer_id=1, env="PROD").model_dump()
}

PREFECT_KUBERNETES_WORKER_STORE_PREFECT_API_IN_SECRET

PREFECT_INTEGRATIONS_KUBERNETES_WORKER_CREATE_SECRET_FOR_API_KEY

Secret

bash
PREFECT_KUBERNETES_WORKER_STORE_PREFECT_API_IN_SECRET=true

bash
PREFECT_INTEGRATIONS_KUBERNETES_WORKER_CREATE_SECRET_FOR_API_KEY=true

bash
# old
PREFECT_KUBERNETES_WORKER_STORE_PREFECT_API_IN_SECRET=true

# new
PREFECT_INTEGRATIONS_KUBERNETES_WORKER_CREATE_SECRET_FOR_API_KEY=true

get

create

update

delete

secrets

prefect-kubernetes

PREFECT_API_URL

PREFECT_API_URL

PREFECT_API_KEY

PREFECT_API_KEY

PREFECT_LOGGING_LEVEL

PREFECT_LOGGING_LEVEL

PREFECT_KUBERNETES_CLUSTER_UID

PREFECT_KUBERNETES_CLUSTER_UID

PREFECT_KUBERNETES_CLUSTER_UID

kube-system

PREFECT_KUBERNETES_WORKER_STORE_PREFECT_API_IN_SECRET

PREFECT_INTEGRATIONS_KUBERNETES_WORKER_CREATE_SECRET_FOR_API_KEY

bash
# old
PREFECT_KUBERNETES_WORKER_STORE_PREFECT_API_IN_SECRET=true

# preferred
PREFECT_INTEGRATIONS_KUBERNETES_WORKER_CREATE_SECRET_FOR_API_KEY=true

PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_ENABLED

PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_NAMESPACES

bash
PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_ENABLED=true
PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_NAMESPACES=prefect,default

bash
PREFECT_INTEGRATIONS_KUBERNETES_WORKER_API_KEY_SECRET_NAME=my-prefect-api-key
PREFECT_INTEGRATIONS_KUBERNETES_WORKER_API_KEY_SECRET_KEY=api-key

AWS_ACCESS_KEY_ID

AWS_SECRET_ACCESS_KEY

AWS_SESSION_TOKEN

job_variables

bash
PREFECT_API_URL=...
PREFECT_API_KEY=...
PREFECT_LOGGING_LEVEL=INFO

# optional, only if needed
PREFECT_KUBERNETES_CLUSTER_UID=...

# preferred secret behavior
PREFECT_INTEGRATIONS_KUBERNETES_WORKER_CREATE_SECRET_FOR_API_KEY=true

# recommended if you want observer scoping
PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_ENABLED=true
PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_NAMESPACES=prefect

PREFECT_API_URL

PREFECT_API_URL

PREFECT_API_KEY

PREFECT_API_KEY

PREFECT_LOGGING_LEVEL

PREFECT_LOGGING_LEVEL

PREFECT_KUBERNETES_CLUSTER_UID

PREFECT_KUBERNETES_CLUSTER_UID

PREFECT_KUBERNETES_WORKER_STORE_PREFECT_API_IN_SECRET

PREFECT_INTEGRATIONS_KUBERNETES_WORKER_CREATE_SECRET_FOR_API_KEY

PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_ENABLED

PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_NAMESPACES

PREFECT_API_URL

PREFECT_API_KEY

PREFECT_KUBERNETES_WORKER_STORE_PREFECT_API_IN_SECRET

task.submit()

PrefectFuture.result()

await future.result()

python
future = await my_task.submit()
result = await future.result()

python
future = my_task.submit()
result = future.result()

await

python
from prefect import flow, task

@task
async def my_task(x: int) -> int:
    return x * 2

@flow
async def my_flow():
    future = my_task.submit(5)
    result = future.result()
    return result

python
future = await my_task.submit()
result = await future.result()

TypeError: object PrefectConcurrentFuture can't be used in 'await' expression

future = task.submit(...)

future = await task.submit(...)

result = future.result()

result = await future.result()

submit()

.result()

result_async()

aresult()

await .*\.submit\(

await .*\.result\(

PrefectFuture

await

.submit()

await

.result()

python
@flow
async def my_flow():
    future = await some_task.submit(1)
    value = await future.result()
    return value

python
@flow
async def my_flow():
    future = some_task.submit(1)
    value = future.result()
    return value

SequentialTaskRunner()

SequentialTaskRunner

.submit()

python
@flow(name="my-flow")
def my_flow():
    a = task_a()
    b = task_b(a)
    c = task_c(b)
    return c

SequentialTaskRunner()

python
from prefect import flow
from prefect.task_runners import SequentialTaskRunner

@flow(
    name="my-flow",
    task_runner=SequentialTaskRunner(),
)
def my_flow():
    ...

python
from prefect import flow

@flow(name="my-flow")
def my_flow():
    ...

python
x = my_task()
y = other_task(x)

python
future = my_task.submit()

ThreadPoolTaskRunner

.submit()

SequentialTaskRunner()

task_runner=SequentialTaskRunner()

.submit()

python
@flow(name="my-flow", task_runner=SequentialTaskRunner())
def my_flow():
    a = task_a.submit()
    b = task_b.submit(wait_for=[a])
    return b.result()

python
@flow(name="my-flow")
def my_flow():
    a = task_a()
    b = task_b(a)
    return b

wait_for=

SequentialTaskRunner()

.submit()

wait_for=

Failed()

Completed()

wait_for=

Completed()

Failed()

.submit()

.submit()

Failed()

wait_for=

.submit()

wait_for=

python
from prefect import flow, task

@task
def step_a():
    print("A")
    return "a"

@task
def step_b():
    print("B")
    return "b"

@task
def step_c():
    print("C")
    return "c"

@flow
def my_flow():
    a = step_a()
    b = step_b()
    c = step_c(wait_for=[a, b])
    return c

python
a = step_a()
b = step_b()
c = step_c()

wait_for=

wait_for=

Failed()

Completed()

.submit()

State

Completed()

Failed()

python
from prefect import flow, task
from prefect.states import Failed, Completed

@task
def validate(x):
    if x < 0:
        return Failed(message="x must be non-negative")
    return Completed(data=x)

@flow
def my_flow():
    state = validate(-1)
    return state

python
return Failed(message="bad input")

python
raise ValueError("bad input")

Failed()

python
from prefect import flow, task
from prefect.states import Failed, Completed

@task
def validate(x):
    if x < 0:
        return Failed(message="x must be non-negative")
    return Completed(data=x)

@task
def process(x):
    return x * 2

@flow
def my_flow(x):
    validation_state = validate(x)

    if validation_state.is_failed():
        return validation_state

    result = process(validation_state.result())
    return result

SequentialTaskRunner

wait_for=

SequentialTaskRunner()

wait_for=

Failed()

Completed()

python
@flow(
    name="my-flow",
    task_runner=SequentialTaskRunner(),
)
def my_flow():
    a = task_a()
    b = task_b(wait_for=[a])
    return b

python
@flow(name="my-flow")
def my_flow():
    a = task_a()
    b = task_b(wait_for=[a])
    return b

b

python
@flow(name="my-flow")
def my_flow():
    a = task_a()
    b = task_b(a)
    return b

Failed()

This is usually simpler if failure should actually fail the flow.

*Best for explicit orchestration / branching on task outcome*
Return `Completed()` / `Failed()` if you intentionally want the flow to inspect task states and decide what to do next.

Example:
```python
@task
def validate(x):
    if x < 0:
        return Failed(message="invalid input")
    return Completed(data=x)

@flow
def my_flow(x):
    state = validate(x)
    if state.is_failed():
        return state
    return next_step(state.result())

task_runner=SequentialTaskRunner()

.submit(..., wait_for=...)

wait_for=...

return Failed(...)

return Completed(...)

wait_for=

Failed()

raise

SequentialTaskRunner()

wait_for=

return Failed(...)

wait_for=

.submit()

Failed()

Completed()

.submit()

.submit()

Failed()

Completed()

SequentialTaskRunner + wait_for + state-returning task

.submit()

.submit()

python
future = get_credentials.submit(...)

PrefectFuture[...]

python
result = get_credentials(...)

python
result_future: PrefectFuture[bool, Literal[False]] = get_credentials.submit(...)

.submit()

python
result: bool = get_credentials(...)

PrefectFuture

python
PrefectFuture[bool, Literal[False]]

python
PrefectFuture[bool]

.submit()

.submit()

PrefectFuture

State

.submit()

python
result_future = get_credentials.submit(...)
result = result_future.result()

python
result = get_credentials(...)

wait_for=[future]

.result()

.wait()

.submit()

.submit()

python
credentials_future = get_credentials.submit(...)
ok = credentials_future.result()

.submit()

python
ok = get_credentials(...)

.submit()

python
credentials_state = get_credentials(...)
if credentials_state.is_failed():
    return credentials_state
ok = credentials_state.result()

.submit()

PrefectFuture[bool, Literal[False]]

async

python
from prefect import flow, task

@task
def step_1():
    return 1

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

@flow
def my_flow():
    a = step_1()
    b = step_2(a)
    return b

async

async def

await ...

httpx.AsyncClient

asyncpg

python
from prefect import flow, task

@task
async def fetch_data():
    ...

@flow
async def my_flow():
    data = await fetch_data()
    return data

SequentialTaskRunner()

.submit()

python
from prefect import flow, task

@task
def get_credentials() -> bool:
    return True

@task
def do_work(credentials_ok: bool):
    if not credentials_ok:
        raise ValueError("Missing credentials")
    return "done"

@flow(name="my-flow")
def my_flow():
    ok = get_credentials()
    result = do_work(ok)
    return result

await

def

async

def

async

def

async def

task_runner=SequentialTaskRunner()

.submit()

python
@flow(
    name="generic-dbt-flow",
    on_cancellation=[handle_cancelled],
    timeout_seconds=24 * 60 * 60,
)
def generic_dbt_flow():
    ...

async def

def task_c

python
@task
def task_c(...):
    ...

python
result = task_c(...)

async def task_a

async def

await

def

python
@task
async def task_a(...):
    ...

async def

python
@flow(
    name="generic-dbt-flow",
    on_cancellation=[handle_cancelled],
    timeout_seconds=24 * 60 * 60,
)
async def generic_dbt_flow():
    a = await task_a(...)
    c = task_c(a)
    return c

python
@task
def task_a(...):
    ...

async def task_b

@sync_compatible

@sync_compatible

@sync_compatible

python
@task
def task_b(...):
    ...

python
@task
async def task_b(...):
    ...

@flow(
    name="generic-dbt-flow",
    on_cancellation=[handle_cancelled],
    timeout_seconds=24 * 60 * 60,
)
async def generic_dbt_flow():
    b = await task_b(...)
    return b

def

async def

async def

def

@sync_compatible

async def task_a

task_b

If you convert the async tasks to sync, then:

```python
@flow(...)
def generic_dbt_flow():
    a = task_a(...)
    b = task_b(...)
    c = task_c(...)
    return c

SequentialTaskRunner()

async def

def

@sync_compatible

def

@task def task_c(...)

def

@task async def task_a(...)

@task async def task_a(...)

def

@task @sync_compatible async def task_b(...)

def

@task @sync_compatible async def task_b(...)

python
from prefect import flow, task

@task
def task_a(...):
    ...

@task
def task_b(...):
    ...

@task
def task_c(...):
    ...

@flow(
    name="generic-dbt-flow",
    on_cancellation=[handle_cancelled],
    timeout_seconds=24 * 60 * 60,
)
def generic_dbt_flow():
    a = task_a(...)
    b = task_b(...)
    c = task_c(...)
    return c

python
from prefect import flow, task

@task
async def task_a(...):
    ...

@task
async def task_b(...):
    ...

@task
def task_c(...):
    ...

@flow(
    name="generic-dbt-flow",
    on_cancellation=[handle_cancelled],
    timeout_seconds=24 * 60 * 60,
)
async def generic_dbt_flow():
    a = await task_a(...)
    b = await task_b(...)
    c = task_c(...)
    return c

SequentialTaskRunner()

def

async def task_a

@sync_compatible async def task_b

def task_c

async def

async def

python
from prefect import flow, task

@task
async def task_a():
    print("a")
    return 1

@task
async def task_b(x):
    print("b")
    return x + 1

@flow
async def my_flow():
    a = await task_a()
    b = await task_b(a)
    return b

task_a

task_b

python
from prefect import flow, task

@task
def task_a():
    return 1

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

@flow
async def my_flow():
    a = task_a()
    b = task_b(a)
    return b

.submit()

asyncio.gather(...)

python
f1 = task_a.submit()
f2 = task_b.submit()

python
a, b = await asyncio.gather(task_a(), task_b())

async def flow

async def

.submit()

python
from prefect import flow, task

@task
async def task_a():
    return "a"

@task
async def task_b(a):
    return f"{a}-b"

@task
def task_c(b):
    return f"{b}-c"

@flow
async def my_flow():
    a = await task_a()
    b = await task_b(a)
    c = task_c(b)
    return c

async def

.submit()

is_schedule_active

flow.deploy(...)

is_schedule_active

python
await my_flow.deploy(
    name="my-deployment",
    work_pool_name="my-pool",
    cron="0 0 * * *",
    is_schedule_active=False,
)

python
from prefect.schedules import Cron

my_flow.deploy(
    name="my-deployment",
    work_pool_name="my-pool",
    schedule=Cron("0 0 * * *", active=False),
)

is_schedule_active

Schedule(..., active=...)

Cron(..., active=...)

flow.deploy(...)

is_schedule_active

python
is_schedule_active=False

python
schedule=Cron("0 0 * * *", active=False)

python
schedules=[
    Cron("0 0 * * *", active=True),
    Interval(3600, active=False),
]

deploy()

python
deployment_id = await my_flow.deploy(...)

python
deployment_id = my_flow.deploy(...)

await flow.deploy(...)

await

schedule

schedules

interval=

cron=

rrule=

schedule=

schedules=

paused=

schedule

schedules

prefect.schedules

python
from prefect.schedules import Cron, Interval

my_flow.deploy(
    name="daily",
    work_pool_name="my-pool",
    schedule=Cron("0 0 * * *", active=True),
)

python
my_flow.deploy(
    name="multi",
    work_pool_name="my-pool",
    schedules=[
        Cron("0 0 * * *", active=True),
        Interval(3600, active=False),
    ],
)

paused

is_schedule_active

paused

flow.deploy(...)

is_schedule_active

paused=True

active=False

python
is_schedule_active=False

paused=True

python
schedule=Cron(..., active=False)

job_variables

flow.deploy(...)

python
job_variables={...}

python
my_flow.deploy(
    name="generic-dbt-flow",
    work_pool_name="eks-pool",
    job_variables={
        "namespace": "prefect",
        "service_account_name": "prefect-worker",
        "image": "<http://123456789012.dkr.ecr.us-east-1.amazonaws.com/dbt:prod|123456789012.dkr.ecr.us-east-1.amazonaws.com/dbt:prod>",
    },
)

version_type

python
version_type=...

enforce_parameter_schema

from_source(...).deploy(...)

python
my_flow_instance = my_flow.from_source(...)
deployment_id = my_flow_instance.deploy(...)

await

deploy(...)

is_schedule_active

python
deployment_id = await my_flow.from_source(
    source=...,
    entrypoint="flows.py:my_flow",
).deploy(
    name="generic-dbt-flow",
    work_pool_name="eks-pool",
    cron="0 0 * * *",
    is_schedule_active=False,
    parameters={"env": "prod"},
)

python
from prefect.schedules import Cron

deployment_id = my_flow.from_source(
    source=...,
    entrypoint="flows.py:my_flow",
).deploy(
    name="generic-dbt-flow",
    work_pool_name="eks-pool",
    schedule=Cron("0 0 * * *", active=False),
    parameters={"env": "prod"},
)

cron=

interval=

rrule=

python
my_flow.deploy(
    name="x",
    work_pool_name="pool",
    cron="0 0 * * *",
)

python
from prefect.schedules import Cron

my_flow.deploy(
    name="x",
    work_pool_name="pool",
    schedule=Cron("0 0 * * *", active=False),
)

flow.deploy(...)

await ...deploy(

is_schedule_active=

paused

cron=

interval=

rrule=

active

await

.deploy(...)

is_schedule_active=

schedule=

schedules=

active=...

paused=

flow.deploy(...)

is_schedule_active

Cron(...)

Interval(...)

RRule(...)

Schedule(...)

active=...

deploy()

await

paused

is_schedule_active

deploy(...)

infra_overrides

job_variables

job_manifest

job_manifest

base_job_template

job_variables

KubernetesJob.job_manifest

infra_overrides

base_job_template.job_configuration.job_manifest

job_variables

KubernetesJob

job_manifest

infra_overrides

base_job_template

job_variables

job_manifest

job_manifest

json
{
  "variables": { ... },
  "job_configuration": {
    "job_manifest": { ... }
  }
}

job_manifest

infra_overrides

infra_overrides

job_variables

python
Deployment.build_from_flow(
    my_flow,
    name="my-deployment",
    infrastructure=KubernetesJob.load("my-k8s-job"),
    infra_overrides={
        "namespace": "production",
        "image": "my-image:latest",
        "image_pull_policy": "Always",
    },
)

python
my_flow.from_source(
    source=...,
    entrypoint="flows.py:my_flow",
).deploy(
    name="my-deployment",
    work_pool_name="my-k8s-pool",
    job_variables={
        "namespace": "production",
        "image": "my-image:latest",
        "image_pull_policy": "Always",
    },
)

infra_overrides

variables

job_configuration.job_manifest

{{ variable_name }}

job_variables

python
job_manifest = {
    "apiVersion": "batch/v1",
    "kind": "Job",
    "metadata": {
        "namespace": "{{ namespace }}"
    },
    "spec": {
        "template": {
            "spec": {
                "serviceAccountName": "{{ service_account_name }}",
                "containers": [
                    {
                        "name": "prefect-job",
                        "image": "{{ image }}",
                        "imagePullPolicy": "{{ image_pull_policy }}"
                    }
                ]
            }
        }
    }
}

python
infra_overrides={
    "namespace": "prod",
    "service_account_name": "prefect-runner",
    "image": "myrepo/dbt:1.2.3",
    "image_pull_policy": "Always",
}

Then your deployment code becomes:

```python
my_flow.from_source(
    source=...,
    entrypoint="flows.py:my_flow",
).deploy(
    name="my-deployment",
    work_pool_name="my-k8s-pool",
    job_variables={
        "namespace": "prod",
        "service_account_name": "prefect-runner",
        "image": "myrepo/dbt:1.2.3",
        "image_pull_policy": "Always",
    },
)

infra_overrides

namespace

image

service_account_name

env

labels

annotations

finished_job_ttl

job_variables

variables

job_configuration.job_manifest

infra_overrides

job_variables

infra_overrides

infra_overrides={"namespace": "prod"}

job_variables={"namespace": "prod"}

infra_overrides={"image": "repo/image:tag"}

job_variables={"image": "repo/image:tag"}

infra_overrides={"image_pull_policy": "Always"}

job_variables={"image_pull_policy": "Always"}

infra_overrides={"env": {...}}

job_variables={"env": {...}}

infra_overrides

job_manifest

job_variables

json
"namespace": "{{ namespace }}"

job_manifest

job_variables

job_variables

base_job_template

job_variables

infra_overrides

job_variables

job_manifest

base_job_template

variables

job_configuration.job_manifest

job_manifest

infra_overrides

job_variables

prefect-worker

prefect-worker

bash
helm get values <release-name> -n <namespace> > current-values.yaml
helm get values -a <release-name> -n <namespace> > current-values-all.yaml

current-values.yaml

current-values-all.yaml

bash
helm inspect values prefect/prefect-worker --version <target-chart-version> > new-defaults.yaml

bash
helm inspect values prefect/prefect-worker --version <old-chart-version> > old-defaults.yaml
helm inspect values prefect/prefect-worker --version <target-chart-version> > new-defaults.yaml
diff -u old-defaults.yaml new-defaults.yaml

bash
helm template prefect-worker prefect/prefect-worker \
  --version <old-chart-version> \
  -f current-values.yaml > old-rendered.yaml

helm template prefect-worker prefect/prefect-worker \
  --version <target-chart-version> \
  -f current-values.yaml > new-rendered.yaml

diff -u old-rendered.yaml new-rendered.yaml

helm diff

bash
helm plugin install <https://github.com/databus23/helm-diff>

bash
helm diff upgrade <release-name> prefect/prefect-worker \
  --namespace <namespace> \
  --version <target-chart-version> \
  -f current-values.yaml \
  --suppress-secrets

values.yaml

<http://eks.amazonaws.com/role-arn|eks.amazonaws.com/role-arn>

PREFECT_API_URL

PREFECT_API_KEY

PREFECT_INTEGRATIONS_KUBERNETES_WORKER_CREATE_SECRET_FOR_API_KEY

helm template

helm diff upgrade

bash
# current release values
helm get values <release> -n <ns> > current-values.yaml
helm get values -a <release> -n <ns> > current-values-all.yaml

# inspect chart defaults
helm inspect values prefect/prefect-worker --version <old-version> > old-defaults.yaml
helm inspect values prefect/prefect-worker --version <new-version> > new-defaults.yaml

# compare defaults
diff -u old-defaults.yaml new-defaults.yaml

# compare rendered output
helm template <release> prefect/prefect-worker \
  --version <old-version> \
  -f current-values.yaml > old-rendered.yaml

helm template <release> prefect/prefect-worker \
  --version <new-version> \
  -f current-values.yaml > new-rendered.yaml

diff -u old-rendered.yaml new-rendered.yaml

# preview upgrade
helm diff upgrade <release> prefect/prefect-worker \
  --namespace <ns> \
  --version <new-version> \
  -f current-values.yaml \
  --suppress-secrets

helm diff upgrade

values.yaml

bash
helm list -n <namespace>

bash
helm status <release-name> -n <namespace>

bash
helm get metadata <release-name> -n <namespace>

bash
helm repo add prefect <https://prefecthq.github.io/prefect-helm>
helm repo update

bash
helm search repo prefect/prefect-worker --versions

bash
helm search repo prefect/prefect-server --versions

bash
helm show chart prefect/prefect-worker --version <chart-version>

bash
helm status <release-name> -n <namespace>

CHART

APP VERSION

async

Sync task called directly from an async flow: this mixed sync/async pattern should be tested carefully during the Prefect 3 migration. An

async

flow can still execute sequentially, and plain

def

tasks may still be used, but directly calling synchronous tasks inside an async flow can block progress in ways that are less predictable than a fully sync flow or a fully async flow. If the flow remains async, validate these sync tasks under the real Prefect 3 runtime image in staging, especially around

wait_for=

dependencies and tasks that return

Failed()

/

Completed()

states.

async

async def

def

wait_for=

Failed()

python
@flow
def my_flow():
    a = task_a()
    b = task_b(a)
    c = task_c(b)
    return c

python
@flow
async def my_flow():
    a = await task_a()
    b = await task_b(a)
    c = task_c(b)
    return c

async def

def

wait_for=

Failed()

wait_for=

Failed()

Completed()

Mixed sync/async execution in Prefect 3: The flow may remain

async

, while some tasks remain plain synchronous

def

tasks. This mixed pattern should be validated carefully in the staging app and staging environment under the real Prefect 3 image. In particular, directly calling sync tasks from an async flow, using

wait_for=

dependencies, and branching on returned

Failed()

/

Completed()

states should all be tested end-to-end, since these patterns are more migration-sensitive than a fully synchronous flow or a fully async flow.

2024.3.23210927

2026.6.18212926

values.yaml

helm diff

worker.config

Role

ClusterRole

RoleBinding

values.yaml

bash
helm get values <release-name> -n <namespace> > current-values.yaml
helm get values -a <release-name> -n <namespace> > current-values-all.yaml

bash
helm inspect values prefect/prefect-worker --version 2024.3.23210927 > old-defaults.yaml
helm inspect values prefect/prefect-worker --version 2026.6.18212926 > new-defaults.yaml
diff -u old-defaults.yaml new-defaults.yaml

bash
helm template prefect-worker prefect/prefect-worker \
  --version 2024.3.23210927 \
  -f current-values.yaml > old-rendered.yaml

helm template prefect-worker prefect/prefect-worker \
  --version 2026.6.18212926 \
  -f current-values.yaml > new-rendered.yaml

diff -u old-rendered.yaml new-rendered.yaml

helm diff

bash
helm diff upgrade <release-name> prefect/prefect-worker \
  --namespace <namespace> \
  --version 2026.6.18212926 \
  -f current-values.yaml \
  --suppress-secrets

helm diff

2024.3.23210927

2026.6.18212926

prefect-worker

eks-v2

eks-v3

text
eks-v2
eks-v3

text
eks-dev-v2
eks-dev-v3
eks-stage-v2
eks-stage-v3
eks-prod-v2
eks-prod-v3

text
eks-dbt-v2
eks-dbt-v3
eks-etl-v2
eks-etl-v3

text
prefect-worker-v2
prefect-worker-v3

text
prefect-worker-stage-v2
prefect-worker-stage-v3
prefect-worker-prod-v2
prefect-worker-prod-v3

text
namespace: prefect

text
prefect-v2
prefect-v3

text
prefect-stage-v2
prefect-stage-v3

text
prefect-worker-v2
prefect-worker-v3

text
prefect-run-v2
prefect-run-v3

text
arn:aws:iam::<acct>:role/prefect-worker-v2
arn:aws:iam::<acct>:role/prefect-worker-v3

text
arn:aws:iam::<acct>:role/prefect-run-v2
arn:aws:iam::<acct>:role/prefect-run-v3

text
prefect-api-key

text
prefect-api-key-v2
prefect-api-key-v3

text
prefect-worker-config-v2
prefect-worker-config-v3

python
my_flow.from_source(
    source=...,
    entrypoint="flows/my_flow.py:my_flow",
).deploy(
    name="generic-dbt-flow",
    work_pool_name="eks-v3",
    job_variables={
        "namespace": "prefect-v3",
        "service_account_name": "prefect-run-v3",
    },
)

text
Workspace: my-team-prod

text
eks-prod-v2
eks-prod-v3

text
prefect-worker-prod-v2
prefect-worker-prod-v3

text
prefect

text
prefect-v2
prefect-v3

*Secrets*
```text
prefect-api-key-v2
prefect-api-key-v3

eks-v3

eks-v2

text
Workspace: acme-prod

Work pools:
- eks-prod-v2
- eks-prod-v3

Worker Helm releases:
- prefect-worker-prod-v2
- prefect-worker-prod-v3

Namespaces:
- prefect-prod-v2
- prefect-prod-v3

ServiceAccounts:
- prefect-worker-v2
- prefect-worker-v3
- prefect-run-v2
- prefect-run-v3

Secrets:
- prefect-cloud-api-key-v2
- prefect-cloud-api-key-v3