.deploy(...)

docker login

.deploy()

Flow.deploy

DockerImage

from prefect import flow
from prefect.docker import DockerImage

@flow
def my_flow():
    ...

# Source could be Git, local path, etc.
flow.from_source(
    source="<https://github.com/acme/prefect-repo>",
    entrypoint="flows/my_flow.py:my_flow",
).deploy(
    name="prod",
    work_pool_name="docker-pool",
    image=DockerImage(
        name="<http://artifactory.myco.com/prefect/my-flow|artifactory.myco.com/prefect/my-flow>",  # <- your Artifactory host/repo
        tag="2025-09-30",
        dockerfile="Dockerfile",
        # Optional: pass Docker build args (e.g., for Artifactory PyPI mirrors, proxies, etc.)
        buildargs={
            "PIP_INDEX_URL": "<https://artifactory.myco.com/api/pypi/pypi/simple>"
        },
    ),
    build=True,
    push=True,
)

docker login <http://artifactory.myco.com|artifactory.myco.com>

FROM python:3.11-slim

FROM <http://artifactory.myco.com/docker-hub-proxy/library/python:3.11-slim|artifactory.myco.com/docker-hub-proxy/library/python:3.11-slim>

docker login <http://artifactory.myco.com|artifactory.myco.com>

.deploy(...)

DOCKER_AUTH_CONFIG

docker login

steps:
    - prefect_docker.deployments.steps.build_docker_image:
        image_name: <http://artifactory.myco.com/prefect/my-flow|artifactory.myco.com/prefect/my-flow>
        tag: "{{ version }}"
    - prefect_docker.deployments.steps.push_docker_image:
        image_name: <http://artifactory.myco.com/prefect/my-flow|artifactory.myco.com/prefect/my-flow>
        tag: "{{ version }}"
        credentials:
          username: "{{ prefect.variables.ARTIFACTORY_USER }}"
          password: "{{ prefect.variables.ARTIFACTORY_TOKEN }}"
          registry_url: "<http://artifactory.myco.com|artifactory.myco.com>"

from prefect_docker.credentials import DockerRegistryCredentials

    creds = DockerRegistryCredentials(
        username="svc-prefect",
        password="***",  # load from env/secret manager
        registry_url="<http://artifactory.myco.com|artifactory.myco.com>",
    )
    creds.save("my-artifactory-creds", overwrite=True)

{
  "image": "{{ image }}",
  "env": "{{ env }}",
  "command": "{{ command }}",
  "registry_credentials": "{{ block('docker-registry-credentials', 'my-artifactory-creds') }}"
}

.deploy(...)

DockerImage.build()

curl -v <https://registry-1.docker.io/v2/>

docker pull alpine:3.20
docker pull hello-world

docker login

"registry-mirrors"

docker pull prefecthq/prefect:3-python3.12

docker build

FROM <http://registry.internal.example.com/prefect/prefect:3-python3.12|registry.internal.example.com/prefect/prefect:3-python3.12>
    # your additional steps...

from prefect import flow
    from prefect.docker import DockerImage

    image = DockerImage(
        name="<http://registry.internal.example.com/your-team/my-flow|registry.internal.example.com/your-team/my-flow>",
        tag="2025-10-07",
        dockerfile="Dockerfile",
        # Optionally help Docker during build if needed:
        # build_kwargs={"network_mode": "host"}
    )

    my_flow.deploy(
        name="my-deployment",
        image=image,
        push=False,  # still builds locally, doesn’t push
    )

image = DockerImage(
        name="<http://registry.internal.example.com/your-team/my-flow|registry.internal.example.com/your-team/my-flow>",
        dockerfile="Dockerfile",
        build_kwargs={
            "buildargs": {
                "HTTP_PROXY": "<http://proxy:3128>",
                "HTTPS_PROXY": "<http://proxy:3128>",
                "NO_PROXY": "localhost,127.0.0.1,.<http://internal.example.com|internal.example.com>"
            }
        },
    )

network_mode=host

build_kwargs={"network_mode": "host"}

my_flow.deploy(...)

prefect.docker.DockerImage

image=

docker pull alpine:3.20

local-docker_client_kas/prefect/numpy-mul

Flow.deploy()

build=True

image

build=False

push=False

from prefect import flow

@flow
def my_flow():
    ...

my_flow.deploy(
    name="my-deployment",
    work_pool_name="your-docker-work-pool",
    image="local-docker_client_kas/prefect/numpy-mul:latest",  # or your specific tag
    build=False,  # <- prevents Prefect from building (no Docker Hub pull)
    push=False,   # <- prevents push
)

local-docker_client_kas

docker login <registry>

local-docker_client_kas

local-docker_client_kas

prefect

local-docker_client_kas/prefect/numpy-mul

prefect

docker run --rm local-docker_client_kas/prefect/numpy-mul:latest prefect version
docker run --rm local-docker_client_kas/prefect/numpy-mul:latest which prefect

# Option A: start from Prefect base
FROM prefecthq/prefect:3-python3.12
# add your code/deps here...

# Option B: extend your image and add Prefect
FROM local-docker_client_kas/prefect/numpy-mul:latest
RUN python -m pip install --no-cache-dir "prefect>=3,<4"

my_flow.deploy(
  name="my-deployment",
  work_pool_name="your-docker-work-pool",
  image="local-docker_client_kas/prefect/numpy-mul:with-prefect",
  build=False,
  push=False,
)

prefect ...

my_flow.deploy(
  name="my-deployment",
  work_pool_name="your-docker-work-pool",
  image="local-docker_client_kas/prefect/numpy-mul:latest",
  build=False,
  push=False,
  job_variables={
    "container_create_kwargs": {"entrypoint": []}
  },
)

prefect

job_variables={"image_pull_policy": "Never"}

local-docker_client_kas

docker login local-docker_client_kas

prefect

prefect

FROM prefecthq/prefect:3-python3.12
# add your code/deps...

FROM local-docker_client_kas/prefect/numpy-mul:latest
RUN python -m pip install --no-cache-dir "prefect>=3,<4"

docker run --rm your-image prefect version

job_variables={"container_create_kwargs": {"entrypoint": []}}

prefect ...

prefect ...

# Dockerfile
    COPY entrypoint.sh /entrypoint.sh
    ENTRYPOINT ["/entrypoint.sh"]

# entrypoint.sh
    #!/bin/sh
    set -e
    # your init steps here
    exec "$@"  # <- runs the CMD passed by the worker, e.g. the Prefect command

job_variables={"container_create_kwargs": {"entrypoint": []}}

from prefect import flow, task
import docker

@task
def run_child_container():
    client = docker.from_env()
    # Runs your image with its own entrypoint/CMD
    logs = client.containers.run(
        "local-docker_client_kas/prefect/numpy-mul:latest",
        detach=False,
        remove=True,
        environment={"FOO": "bar"},
    )
    print(logs.decode() if isinstance(logs, bytes) else logs)

@flow
def my_flow():
    run_child_container()

# Deploy using a Prefect-enabled image for the ORCHESTRATION container
my_flow.deploy(
    name="my-deployment",
    work_pool_name="your-docker-work-pool",
    image="prefecthq/prefect:3-python3.12",
    build=False,
    push=False,
    job_variables={
        # allow the flow to talk to the host Docker daemon
        "volumes": ["/var/run/docker.sock:/var/run/docker.sock"],
        # optionally set DOCKER_HOST if needed
        "env": {"DOCKER_HOST": "unix:///var/run/docker.sock"},
    },
)

container_create_kwargs.entrypoint=[]

docker ps -a --no-trunc --filter "status=exited" --format "{{.ID}}\t{{.Names}}\t{{.Status}}"

docker inspect <container_id> --format '{{.Config.Entrypoint}} {{.Config.Cmd}}'
docker logs <container_id>

job_variables={"stream_output": True}

docker logs

docker run --rm YOUR_IMAGE prefect version
docker run --rm --entrypoint '' YOUR_IMAGE prefect version   # bypass ENTRYPOINT if needed

exec "$@"

PREFECT_LOGGING_LEVEL=DEBUG prefect worker start -p your-docker-work-pool

docker inspect <container_id> --format '{{.Config.Entrypoint}} {{.Config.Cmd}}'

docker logs <container_id>

prefect ...

# No ENTRYPOINT; Prefect supplies the command at runtime

# Dockerfile
  COPY entrypoint.sh /entrypoint.sh
  ENTRYPOINT ["/entrypoint.sh"]

# entrypoint.sh
  #!/bin/sh
  set -e
  # optional init work...
  exec "$@"   # critical: runs the Prefect command that the worker passes

my_flow.deploy(
    name="my-deployment",
    work_pool_name="your-docker-work-pool",
    image="your/image:tag",
    build=False,
    push=False,
    job_variables={
      "container_create_kwargs": {
        "entrypoint": []   # clears ENTRYPOINT for this run
      }
    },
  )

ENTRYPOINT [""]

entrypoint: []

container_create_kwargs

exec "$@"

prefect

docker run --rm your/image:tag prefect version

docker run --rm your-image prefect version

exec "$@"

job_variables={"container_create_kwargs": {"entrypoint": []}}

PREFECT_LOGGING_LEVEL=DEBUG prefect worker start -p your-docker-work-pool

docker run --rm your-image prefect version

# Start your worker with a resolvable URL
PREFECT_API_URL=http://<server-host-or-ip>:4200/api prefect worker start -p your-docker-work-pool

my_flow.deploy(
  ...,
  job_variables={
    "env": {"PREFECT_API_URL": "http://<server-host-or-ip>:4200/api"}
  },
)

docker inspect <prefect-server-container-name> --format '{{json .NetworkSettings.Networks}}' | jq .

my_flow.deploy(
  ...,
  job_variables={
    "networks": ["<the-server-network-name>"]
  },
)

curl -sS http://<server-host-or-ip>:4200/api/health

docker run --rm --network <the-server-network-name> curlimages/curl -sS <http://prefect-server:4200/api/health>

FROM prefecthq/prefect:3-python3.12

WORKDIR /app
COPY . /app

# install your deps if needed
# RUN pip install -r requirements.txt

my_flow.deploy(
  name="image-average",
  work_pool_name="your-docker-work-pool",
  image="your-registry/your-image:with-code",
  build=False,
  push=False,
)

deployments/local_deployment_image_average.py:cluster_image_and_process

my_flow.deploy(
  name="image-average",
  work_pool_name="your-docker-work-pool",
  image="your-registry/your-image:tag",
  build=False,
  push=False,
  job_variables={
    "volumes": [
      "/abs/path/on/worker/host/project:/app"  # must exist on the worker host
    ],
    # optional: set working dir if needed
    "container_create_kwargs": {"working_dir": "/app"}
  },
)

from prefect import flow

flow = flow.from_source(
    source="<https://github.com/your-org/your-repo>",
    entrypoint="deployments/local_deployment_image_average.py:cluster_image_and_process",
)

flow.deploy(
  name="image-average",
  work_pool_name="your-docker-work-pool",
  image="prefecthq/prefect:3-python3.12",
  build=False,
  push=False,
)

source

docker run --rm -it your-image:with-code ls -l /app/deployments

log_prints=True

from prefect import flow, task

  @task(log_prints=True)
  def my_task():
      print("hello from task")

  @flow(log_prints=True)
  def my_flow():
      print("hello from flow")
      my_task()

# in code
  from prefect.settings import PREFECT_LOGGING_LOG_PRINTS, temporary_settings
  # or set env var before starting worker/run: PREFECT_LOGGING_LOG_PRINTS=true

my_flow.deploy(
    ...,
    job_variables={"stream_output": True}
  )

print(..., flush=True)

PYTHONUNBUFFERED=1

my_flow.deploy(
      ...,
      job_variables={"env": {"PYTHONUNBUFFERED": "1"}}
    )

log_prints=True

stream_output=True

log_prints

docker logs <container_id>

job_variables

<http://infra-repos.ohb-system.de:443/local-docker_client_kas/prefect/image_average:0.0.3|infra-repos.ohb-system.de:443/local-docker_client_kas/prefect/image_average:0.0.3>

localhost

<http://prefect-server:4200/api>

http://<host-ip>:4200/api

<http://host.docker.internal:4200/api>

"true"

stream_output=True

log_prints=True

cluster_image_and_process.deploy(
    name="my-seventh-docker-deployment",
    work_pool_name="docker-worker-pool-in-compose-on-brekasdev06",
    image="<http://infra-repos.ohb-system.de:443/local-docker_client_kas/prefect/image_average:0.0.3|infra-repos.ohb-system.de:443/local-docker_client_kas/prefect/image_average:0.0.3>",
    build=False,
    push=False,
    job_variables={
        "env": {
            "PREFECT_API_URL": "<http://prefect-server:4200/api>",   # or http://<host-ip>:4200/api
            "PYTHONUNBUFFERED": "1",
            "PREFECT_LOGGING_LOG_PRINTS": "true"
        },
        "stream_output": True,
        # If your Prefect Server runs on a compose network, add it here:
        # "networks": ["<server_network_name>"]
    },
)

from prefect import flow, task

@task(log_prints=True)
def process_kernel():
    print("hello from task")

@flow(log_prints=True)
def cluster_image_and_process():
    print("hello from flow")
    process_kernel()

curl -sS $PREFECT_API_URL/health

docker logs <container_id>

PREFECT_LOGGING_LEVEL=DEBUG prefect worker start -p docker-worker-pool-in-compose-on-brekasdev06

PREFECT_API_URL

networks