oleander-whirlwind
01/02/2026, 7:49 AMMarvin
01/02/2026, 7:49 AMMarvin
01/02/2026, 7:52 AMfrom pydantic_settings import BaseSettings, SettingsConfigDict
from pydantic import SecretStr
class AppSettings(BaseSettings):
model_config = SettingsConfigDict(
env_prefix="MYAPP_",
env_file=".env", # optional, for local dev
case_sensitive=False,
)
db_url: str = "sqlite:///local.db"
api_key: SecretStr = SecretStr("")
settings = AppSettings() # module-level, load once
flow.py
from prefect import flow, task
from config import settings
@task
def use_config():
# safely unwrap when you need the raw value
key = settings.api_key.get_secret_value()
return settings.db_url, bool(key)
@flow
def my_flow():
return use_config()
If you want Prefect-integrated env names (Pydantic v2 + Prefect 3.x)
- This lets you use PREFECT_MYAPP_* env vars and get automatic masking in UI/logs for secrets.
config.py
from pydantic import SecretStr
from prefect.settings.base import PrefectBaseSettings, build_settings_config
class AppSettings(PrefectBaseSettings):
# Enables PREFECT_MYAPP_* env vars (e.g., PREFECT_MYAPP_API_KEY)
model_config = build_settings_config(("myapp",))
api_key: SecretStr = SecretStr("")
db_url: str = "sqlite:///local.db"
settings = AppSettings()
Passing values in a deployment (Docker/K8s work pools)
prefect.yaml
deployments:
- name: prod
entrypoint: flows/main.py:my_flow
work_pool:
name: docker
job_variables:
env:
# Non-prefect prefix
MYAPP_DB_URL: "<postgresql://db/prod>"
# Or Prefect-integrated prefix (if using PrefectBaseSettings)
PREFECT_MYAPP_API_KEY: "{{ prefect.blocks.secret.api-key }}"
Secrets: blocks vs env
- For sensitive values, use SecretStr fields and either:
- Inject via env from a Prefect Secret block in prefect.yaml (example above), or
- Load a block at runtime in your flow/task.
- Env is simplest; blocks give better security and can be rotated without redeploying.
When to instantiate inside the flow
- Rarely needed. Only do it if you truly need to re-read env mid-run or after import time. Otherwise, module-level instantiation is preferred.
- If you need “reloadable” behavior, wrap settings in a small getter you can refresh:
from functools import lru_cache
@lru_cache(maxsize=1)
def get_settings():
from config import AppSettings
return AppSettings()
def refresh_settings():
get_settings.cache_clear()
Docs for deeper reference
- Settings & profiles: Prefect settings and profiles
- Blocks and secrets: Prefect blocks
- Deployments overview: Deployments
If you can share whether you’re on Docker/Kubernetes or local, and whether you need settings to change at runtime between tasks, I can tailor the pattern to your setup.