Great question. Assuming you’re on Prefect 3.x, here’s a concise pattern that works well in practice: Preferred way to use Pydantic Settings with Prefect - Use Pydantic Settings (v2) for your application config and validation. - Feed it with environment variables provided by your deployment/worker (via job_variables.env). - Layer in Prefect data: - Secrets: Prefect Secret blocks for encrypted credentials. - Variables: Prefect Variables for values that change frequently between runs. Example

from pydantic_settings import BaseSettings, SettingsConfigDict
from prefect import flow, task
from prefect.variables import Variable
from prefect.blocks.system import Secret

class AppConfig(BaseSettings):
    # Will read from ENV by default (API_KEY, ENVIRONMENT, TIMEOUT)
    api_key: str | None = None
    environment: str = "development"
    timeout: int = 30

    # Optional .env for local dev; ENV wins in production
    model_config = SettingsConfigDict(env_file=".env", case_sensitive=True)

@task
def use_config(cfg: AppConfig):
    print(f"Env={cfg.environment}, timeout={cfg.timeout}")
    # Use cfg.api_key to call external services

@flow
def my_flow():
    # 1) Base from ENV (set by job_variables.env or your worker)
    cfg = AppConfig()

    # 2) Dynamic values that may rotate between runs
    #    (e.g., endpoint, small non-sensitive toggles)
    endpoint = Variable.get("api-endpoint", default=None)
    if endpoint:
        print(f"Endpoint from Variable: {endpoint}")

    # 3) Sensitive credentials (encrypted at rest)
    #    If you keep `api_key` in a Secret block, load it at run time
    try:
        cfg.api_key = Secret.load("my-api-key").get()
    except Exception:
        # Fall back to env if not using a Secret block
        pass

    use_config(cfg)

Handling rotating “environment variables” between runs You have a few good options depending on sensitivity and control: 1) Runtime overrides via job variables (per-run, easy to audit) - When you trigger a run, pass a different value each time. This is ideal for per-run overrides like tokens, feature flags, and ad-hoc config. CLI

prefect deployment run <FLOW_NAME>/<DEPLOYMENT_NAME> \
  --job-variable env='{"EXECUTION_ENV":"staging","TIMEOUT":"45"}'

Python

from prefect.deployments import run_deployment

run_deployment(
    "my-flow/my-deployment",
    job_variables={"env": {"EXECUTION_ENV": "staging", "TIMEOUT": "45"}}
)

Docs: - Trigger ad-hoc deployment runs - Customize job variables 2) Prefect Variables for frequently changing values - Store the latest value and read it at run time. Good for values that rotate often and don’t need to be encrypted by Prefect. - Note: Variables are not encrypted. If the value is sensitive, prefer a Secret block or fetch from your secret manager at runtime.

from prefect.variables import Variable

# Update this in CI/automation whenever it rotates
Variable.set("rotating-api-token", "new-token-123", overwrite=True)

# Flow read
token = Variable.get("rotating-api-token")

Docs: - Variables concept 3) Secret blocks for encrypted credentials - Store sensitive data (API keys, passwords) in a Prefect Secret block. To “rotate,” overwrite the block value before runs; subsequent runs will read the latest value. - For very short-lived tokens, consider a flow or external job that refreshes the block/variable before main runs, or fetch directly from your external secret manager at runtime.

from prefect.blocks.system import Secret

# Set/rotate (in setup/automation)
Secret(value="sk-live-456").save("my-api-key", overwrite=True)

# Flow read
api_key = Secret.load("my-api-key").get()

Docs: - Store secrets

- Blocks 4) Define environment for deployments via YAML or .deploy() - Use job_variables.env in your deployment so Pydantic Settings will pick them up. prefect.yaml

deployments:
  - name: my-deployment
    entrypoint: flows.py:my_flow
    work_pool: { name: local }
    job_variables:
      env:
        EXECUTION_ENV: "production"
        TIMEOUT: "60"

Docs: - Define deployments with YAML (templating with $ENV and prefect.variables is covered here) Best practices - Secrets/credentials: Secret blocks (or fetch from your external secret store at runtime). - Values that change frequently: Prefect Variables or per-run job variables. - App config: Pydantic Settings to validate and provide sane defaults, reading from ENV set by deployment/worker. - Rotate per run: Prefer

--job-variable env='{"KEY":"value"}'

or

run_deployment(..., job_variables={"env": {...}})

. - Avoid committing secrets to .env or code; prefer Secret blocks or your external secret manager. Helpful docs - Variables: Concepts and Share configuration - Secret block: Store secrets - Job variables and YAML: Customize job variables and Define deployments with YAML - Triggering runs with overrides: Trigger ad-hoc runs - Prefect settings/profiles (if helpful): Settings and profiles Follow-up - Which worker/infrastructure are you using (Local, Docker, Kubernetes)? - Do you want rotation driven externally (CI/secret manager) or at trigger time per run? With that, I can suggest a concrete wiring pattern for your setup.

Got it — you’ve got lots of per-job configuration that you don’t want as flow parameters, but you want to inject at run time. In Prefect 3.x, the cleanest pattern is: - Define a Pydantic Settings model (with nested support) for validation. - Inject values per deployment/run via job variables (env). - For very large or frequently-changing configs, reference a single “pointer” and fetch the config at runtime (Variables/Blocks/remote file), instead of passing dozens of env vars. Here are three practical patterns — pick the one that fits your scale and security needs. 1) Many small knobs: nested env vars + job variables - Use Pydantic Settings with

env_prefix

and

env_nested_delimiter

so you can inject deeply nested values using environment variables. - Pass them per deployment or per run via

job_variables.env

. Code

from pydantic import BaseModel
from pydantic_settings import BaseSettings, SettingsConfigDict

class DB(BaseModel):
    host: str
    port: int = 5432

class Service(BaseModel):
    base_url: str
    timeout: int = 30

class AppConfig(BaseSettings):
    db: DB = DB(host="localhost")
    service: Service = Service(base_url="<http://localhost>")
    feature_flags: dict[str, bool] = {}

    model_config = SettingsConfigDict(
        env_prefix="APP_",
        env_nested_delimiter="__",
        case_sensitive=True,
    )

# Example: cfg reads APP_DB__HOST, APP_SERVICE__BASE_URL, etc.
# APP_DB__HOST="db.prod" APP_SERVICE__TIMEOUT="60" -> cfg = AppConfig()

Deployment YAML

deployments:
  - name: heavy-config
    entrypoint: flows.py:my_flow
    work_pool: { name: local }
    job_variables:
      env:
        APP_DB__HOST: "<http://db.prod.example.com|db.prod.example.com>"
        APP_DB__PORT: "6432"
        APP_SERVICE__BASE_URL: "<https://api.example.com>"
        APP_SERVICE__TIMEOUT: "60"
        APP_FEATURE_FLAGS__EXPERIMENTAL: "true"

Per-run override

prefect deployment run my-flow/heavy-config \
  --job-variable env='{"APP_SERVICE__TIMEOUT": "45", "APP_FEATURE_FLAGS__EXPERIMENTAL": "false"}'

Docs: - Customize job variables - Trigger ad-hoc runs 2) One big payload: single JSON config env var - If you don’t want to inject dozens of env vars, inject a single JSON blob and validate it with Pydantic. - Pass

APP_CONFIG

per deployment or per run; parse it at runtime. Code

import os, json
from pydantic import BaseModel

class DB(BaseModel):
    host: str
    port: int = 5432

class Service(BaseModel):
    base_url: str
    timeout: int = 30

class AppConfig(BaseModel):
    db: DB
    service: Service
    feature_flags: dict[str, bool] = {}

def load_config() -> AppConfig:
    raw = os.environ.get("APP_CONFIG")
    if raw:
        return AppConfig.model_validate_json(raw)
    # Fallback to env-driven defaults or static defaults as needed
    return AppConfig(
        db=DB(host=os.environ.get("APP_DB__HOST", "localhost")),
        service=Service(base_url=os.environ.get("APP_SERVICE__BASE_URL", "<http://localhost>")),
    )

Per-run override

prefect deployment run my-flow/heavy-config \
  --job-variable env='{"APP_CONFIG": "{\"db\":{\"host\":\"db.test\",\"port\":5432},\"service\":{\"base_url\":\"<https://staging.api>\",\"timeout\":45},\"feature_flags\":{\"exp\":true}}"}'

This keeps the deployment clean and centralizes validation in one place. 3) Large or rotating configs: store once, pass a pointer - Store config as JSON in a Prefect Variable (or in a remote file S3/GCS). - Pass a small selector per run (e.g.,

CONFIG_NAME

or

CONFIG_VERSION

) via job variables, then load and validate at runtime. - For secrets: store them in Secret blocks or fetch from your external secret manager; don’t put secrets in Variables or plain env. Code ``` from prefect.variables import Variable from pydantic import BaseModel class AppConfig(BaseModel):

# ... your nested schema ... def get_config_by_name(name: str) -> AppConfig: payload = Variable.get(f"config:{name}") # store JSON under keys like config:prod return AppConfig.model_validate_json(payload) # Per run: pass CONFIG_NAME in job_variables.env # then: import os cfg = get_config_by_name(os.environ.get("CONFIG_NAME", "default"))

Variable setup

from prefect.variables import Variable import json config_prod = { "db": {"host": "db.prod", "port": 5432}, "service": {"base_url": "https://api.prod", "timeout": 60} } Variable.set("config:prod", json.dumps(config_prod), overwrite=True)

Docs:
- <https://docs.prefect.io/v3/concepts/variables|Variables concept>
- <https://docs.prefect.io/v3/how-to-guides/configuration/share-configuration-between-workflows|Share configuration>


Security guidance
- Secrets/credentials: use Secret blocks or your secret manager; pass only IDs/refs per run and resolve at runtime.
- Non-sensitive tuning knobs: env + Variables are fine.
- If configs include some secrets, keep the secrets in Secret blocks and include only non-sensitive fields/connections in the config JSON; look up the secrets by name when building the Pydantic model.

Putting it together in a flow

from prefect import flow, task from prefect.blocks.system import Secret @task def run_job(cfg: AppConfig): # use cfg safely ... @flow def my_flow(): # Choose one of the loaders above cfg = load_config() # Optionally resolve secrets separately try: cfg_api_key = Secret.load("my-api-key").get() # attach to cfg or pass separately to tasks except Exception: pass run_job(cfg) ``` If you share a bit more: - Roughly how large are these config payloads? - Any secrets inside them? - Which infra/worker are you using (Local, Docker, K8s)? I can tailor a concrete wiring (YAML + code + CLI) for your setup. Reference docs - Job variables: Customize job variables - Trigger with overrides: Trigger ad-hoc runs - Define deployments with YAML (templating): Define deployments with YAML - Variables: Variables - Secrets: Store secrets

You’re not missing anything — there isn’t a place to “upload an .env file” for the ephemeral container. In Prefect 3, the runtime environment for a flow run is built by merging: - Work pool defaults (shared across deployments in the pool) - Deployment job_variables.env (the deployment’s base env) - Per-run overrides (job variables passed when you trigger a run) There’s no native env_file support. You inject env via those three layers. What this means for your use case - If you have stable profiles (history channel vs mubu store) you want to run on their own schedules: create one deployment per profile with its own env. This is the simplest, explicit approach. - If you want a single deployment that can run with different configs on demand: use per-run job variables to override env, or pass a small “config pointer” (e.g., CONFIG_NAME) and load the full config at runtime. Practical recipes A) Multiple deployments, one env each (clear, great for schedules)

# prefect.yaml
deployments:
  - name: downloader-history
    entrypoint: flows/downloader.py:download
    work_pool: { name: k8s }
    job_variables:
      env:
        APP_PROFILE: "history"
        SRC_CHANNEL: "history-channel"
        STORE: "history-channel"
        # ... more keys

  - name: downloader-mubu
    entrypoint: flows/downloader.py:download
    work_pool: { name: k8s }
    job_variables:
      env:
        APP_PROFILE: "mubu"
        SRC_CHANNEL: "love"
        STORE: "mubu-store"
        # ... more keys

B) One deployment, override env per run (ad-hoc)

prefect deployment run downloader/generic \
  --job-variable env='{"APP_PROFILE":"history","SRC_CHANNEL":"history-channel","STORE":"history-channel"}'

Docs: Trigger ad-hoc runs, Customize job variables C) One deployment, pass a “config pointer” and load full config at runtime (best when you have lots of settings) - Store large config blobs as Prefect Variables (or in S3/GCS) keyed by name. - Pass only

CONFIG_NAME

as an env var per run or per deployment. - Validate the loaded config with Pydantic.

# during run
prefect deployment run downloader/generic \
  --job-variable env='{"CONFIG_NAME":"history"}'

# flow code
import os
from prefect.variables import Variable
from pydantic import BaseModel

class DownloaderConfig(BaseModel):
  # your nested structure here
  source: str
  store: str
  # ...

def load_config() -> DownloaderConfig:
  name = os.environ.get("CONFIG_NAME", "default")
  payload = Variable.get(f"config:{name}")  # JSON string
  return DownloaderConfig.model_validate_json(payload)

Docs: Variables concept D) Using an .env file you already have - Prefect does not accept an env file directly. Options: 1) At deploy time, template env values from your shell into the deployment: -

export $(cat .env.history | xargs); prefect deploy -n downloader-history

- prefect.yaml:

job_variables:
         env:
           SRC_CHANNEL: "{{ $SRC_CHANNEL }}"
           STORE: "{{ $STORE }}"

Docs: Define deployments with YAML (templating) 2) For Kubernetes: Convert your .env into a Secret/ConfigMap and reference keys via env.valueFrom. Example:

job_variables:
       env:
         - name: API_KEY
           valueFrom:
             secretKeyRef:
               name: downloader-api
               key: api_key

Then keep separate Secrets/ConfigMaps per profile (history, mubu). Note: you still list env entries you need; there’s no env_file shortcut. E) Keep the env surface small - Use Pydantic Settings with nested support and a single JSON env var if you prefer not to enumerate many keys:

- Pass

APP_CONFIG

once per deployment/run; validate with Pydantic.

# run-time override
  prefect deployment run downloader/generic \
    --job-variable env='{"APP_CONFIG":"{\"source\":\"history-channel\",\"store\":\"history\"}"}'

# flow code
  import os
  from pydantic import BaseModel

  class DownloaderConfig(BaseModel):
      source: str
      store: str

  def load_config():
      raw = os.environ.get("APP_CONFIG")
      if raw:
          return DownloaderConfig.model_validate_json(raw)
      # fallback defaults...

How to think about “base envs” - A deployment has one base env (its job_variables.env), but you can: - Create multiple deployments to represent multiple base configs; or - Keep one base config and override per run with job variables; or - Keep one deployment and pass a profile pointer to load the correct config at runtime. - Work pool defaults can hold shared env for many deployments; deployments then add/override specifics; runs can still override again. Security note - Put secrets in Prefect Secret blocks (or your external secret manager) and only pass references/IDs via env. Don’t store secrets in Variables. If you tell me: - Docker or Kubernetes? - Do you want separate scheduled deployments per profile, or a single deployment that switches profiles per run? - Approximate config size and whether secrets are included, I can give you a concrete YAML + code snippet tailored to your setup.

Thanks — since you use both Docker and Kubernetes depending on the case, here’s a clear way to map your “many profiles/configs” setup to Prefect 3 without needing a literal .env file in the ephemeral container. Key idea - Prefect doesn’t mount a .env file into the container. Instead, you set environment variables via: - Work pool defaults (shared) - Deployment job_variables.env (base per deployment) - Per-run overrides (job variables when you trigger a run) - For Kubernetes specifically, you can also reference K8s Secrets/ConfigMaps via env entries or add envFrom in the job manifest. Patterns you can choose from 1) Multiple deployments = multiple base configs (simple and explicit) - Create a deployment per profile (history vs mubu). Each deployment has its own env under job_variables.env. - Good for scheduled runs and when config is relatively stable per profile. Docker example (prefect.yaml)

deployments:
  - name: downloader-history
    entrypoint: flows/downloader.py:download
    work_pool: { name: docker-pool }
    job_variables:
      env:
        APP_PROFILE: "history"
        SRC_CHANNEL: "history-channel"
        STORE: "history"

  - name: downloader-mubu
    entrypoint: flows/downloader.py:download
    work_pool: { name: docker-pool }
    job_variables:
      env:
        APP_PROFILE: "mubu"
        SRC_CHANNEL: "love"
        STORE: "mubu-store"

Kubernetes equivalent (dict env also works; see K8s section below). 2) One deployment, pass profile per run (ad-hoc flexibility) - Keep one “generic” deployment; when you trigger a run, pass the profile’s env as job variables. - Ideal if you often switch configs on demand.

prefect deployment run downloader/generic \
  --job-variable env='{"APP_PROFILE":"history","SRC_CHANNEL":"history-channel","STORE":"history"}'

3) One deployment with a config “pointer” (best for big configs) - Store full config blobs as Prefect Variables or in S3/GCS, keyed by name. - At run time, pass a small pointer like CONFIG_NAME via env and load/validate with Pydantic inside the flow.

prefect deployment run downloader/generic \
  --job-variable env='{"CONFIG_NAME":"history"}'

from prefect.variables import Variable
from pydantic import BaseModel
import os

class DownloaderConfig(BaseModel):
    source: str
    store: str
    # ... lots more fields

def load_config() -> DownloaderConfig:
    name = os.environ.get("CONFIG_NAME", "default")
    payload = Variable.get(f"config:{name}")  # JSON string
    return DownloaderConfig.model_validate_json(payload)

This avoids maintaining many env keys and keeps your YAML clean. 4) Reusing your existing .env files - Prefect won’t mount an .env into the run container. Instead: - At deploy time: export your .env into the shell and template values into the deployment. -

export $(cat .env.history | xargs); prefect deploy -n downloader-history

- prefect.yaml:

job_variables:
        env:
          SRC_CHANNEL: "{{ $SRC_CHANNEL }}"
          STORE: "{{ $STORE }}"

- For Kubernetes: convert your .env to Secret/ConfigMap and reference keys via env/valueFrom or envFrom (see below). Kubernetes-specific options you get - env supports either: - Dict form:

job_variables:
      env:
        LOG_LEVEL: "INFO"
        SRC_CHANNEL: "history-channel"

- List form with valueFrom (K8s-native):

job_variables:
      env:
        - name: DB_PASSWORD
          valueFrom:
            secretKeyRef:
              name: db-secret
              key: password
        - name: API_URL
          valueFrom:
            configMapKeyRef:
              name: app-config
              key: api_url

- envFrom is not a first-class job_variable, but you can add it in the job manifest template and Prefect will pass it through: ``` definitions: work_pool: &k8s_pool name: "kubernetes" job_configuration: job_manifest: spec:

template: spec: containers: - name: prefect-job env: "{{ env }}" # keep Prefect-managed env envFrom: # add your bulk refs here - configMapRef: name: downloader-config - secretRef: name: downloader-secrets ``` - Volumes/volumeMounts are fully supported via job_manifest: add your mounts for configuration files if you prefer file-based config. Docker-specific notes - No native env_file support in Prefect’s Docker worker. Provide env via job_variables.env or bake base config into the image and use job variables for deltas. - If you have to reuse a .env, export it in your CI before calling

prefect deploy

and template values into your deployment (like above). Where to keep secrets - Use Prefect Secret blocks or your cloud secret manager. Pass only references/IDs in env. For K8s, prefer Kubernetes Secrets with valueFrom or envFrom secretRef. Putting Pydantic Settings on top - Let Pydantic Settings assemble the final config from env at runtime and validate it. - For large config, pass a single JSON env var (e.g., APP_CONFIG) and

model_validate_json

, or use the CONFIG_NAME pointer approach. Choosing a path - If the “history” vs “mubu” configs are long-lived with their own schedules: create two deployments with distinct env. - If you need one deployment that can switch profiles on demand: use per-run job variables or the CONFIG_NAME pointer pattern. - Kubernetes: if you already maintain Secrets/ConfigMaps, use valueFrom or envFrom in the manifest for clean separation. If you share: - Whether you maintain configs as files, K8s Secrets/ConfigMaps, or want to centralize them in Prefect Variables, - Whether you want multiple scheduled deployments or one flexible deployment, I can draft a concrete prefect.yaml and flow snippet for both Docker and K8s to match your workflow.

Short answer: - Logger: always create it inside the flow/task with

get_run_logger()

. Never at module/global scope. - Settings: instantiate at runtime (inside the flow or a helper called by the flow), not at import/global scope—so that per-deployment/per-run env and Prefect Variables/Secrets are respected. Why -

get_run_logger()

requires an active run context; at global scope it raises “No active run context.” - If you build

DownloaderSettings()

at import time, it will read whatever env exists then and “freeze” it. You’ll miss deployment/job-variable overrides and any per-run changes. Recommended pattern

from prefect import flow, task, get_run_logger

# Define classes and loader functions at module scope (safe)
class DownloaderSettings(...):
    ...

def load_settings(config_name: str | None) -> DownloaderSettings:
    # read env, Prefect Variables/Secrets, config pointers, etc.
    return DownloaderSettings(...)

@task
def process_item(item, settings: DownloaderSettings):
    logger = get_run_logger()
    <http://logger.info|logger.info>(f"Processing {item} with store={settings.store}")
    ...

@flow
def download_movies(config_name: str | None = None):
    logger = get_run_logger()               # OK here
    settings = load_settings(config_name)   # Build per run
    <http://logger.info|logger.info>(f"Using profile={config_name or 'default'}")

    # Pass settings to tasks so they don’t re-read env repeatedly
    for item in ["a", "b", "c"]:
        process_item.submit(item, settings)

Notes - Keep only definitions (classes, schemas, helper functions) at global scope; do not instantiate config or call

get_run_logger()

there. - If your config is large or expensive to compute, build it once in the flow and pass it to tasks (as shown) rather than recreating it in each task. - If you need an import-time logger for non-flow code, use a standard Python logger as a fallback (e.g.,

logging.getLogger(__name__)

), but prefer

get_run_logger()

inside flow/task for Prefect-integrated logs.