Joshua Caskie
10/22/2025, 1:14 PMMarvin
10/22/2025, 1:15 PMMarvin
10/22/2025, 1:19 PM@flow-decorated function has the original callable under .fn. Calling it runs your code “like normal Python” — no flow run, no orchestration/state — while you can still use Prefect’s Blocks/Variables to fetch secrets.
2) Recommended structure: separate business logic from orchestration
- Keep your core logic in a plain function, then wrap it with a small @flow if/when you want orchestration. This makes “normal Python” the default, and Prefect just a thin wrapper.
How to connect to Prefect Cloud for secrets (outside flows)
- Authenticate once using either:
- CLI profile (non-interactive environments can pass flags):
prefect cloud login --key <YOUR_API_KEY> --workspace <account_handle>/<workspace_handle>
- Or set environment variables:
export PREFECT_API_KEY=<YOUR_API_KEY>
export PREFECT_API_URL=<https://api.prefect.cloud/api/accounts/<ACCOUNT_ID>/workspaces/<WORKSPACE_ID>>
- Then load variables and secret blocks anywhere in your code:
from prefect.variables import Variable
from prefect.blocks.system import Secret
# Variables
region = Variable.get("aws_region", default="us-east-1")
# Secret block (created in UI or via code)
db_password = Secret.load("my-db-password").get()
Complete example
- Pure business logic that uses Prefect Cloud for secrets:
from prefect.variables import Variable
from prefect.blocks.system import Secret
def business_logic(x: int, y: int) -> int:
# Pull a variable and secret from Cloud (works without a running flow)
region = Variable.get("aws_region", default="us-east-1")
api_key = Secret.load("service-api-key").get()
# Do your normal work
# (Use region/api_key however you need)
return x + y
- Optional flow wrapper (orchestration when you want it):
from prefect import flow
@flow(name="my-orchestrated-flow")
def my_flow(x: int, y: int) -> int:
return business_logic(x, y)
if __name__ == "__main__":
# Run as “normal Python” (no flow run)
print("Pure:", business_logic(1, 2))
# Orchestrated run (tracked in Prefect)
print("Orchestrated:", my_flow(1, 2))
# You can also bypass orchestration via `.fn`
print("Manual bypass:", my_flow.fn(1, 2))
Important notes
- When you call my_flow.fn(...) or business_logic(...), Prefect runtime context (e.g., get_run_logger, prefect.runtime.*) is not available. Keep logging and runtime-dependent code inside the orchestrated wrapper, or guard it.
- Loading Blocks/Variables works fine without an active flow as long as you’re authenticated to Cloud (via profile/env vars).
Docs for reference
- Variables: Variables
- Secrets via Blocks: Store secrets
- Blocks overview: Blocks
- Connect to Cloud: Connect to Cloud
- API Keys: Create API keys
Would you like me to tailor this to your exact flow or show an example with a specific secret provider (e.g., AWS creds)?Joshua Caskie
10/22/2025, 1:22 PMMarvin
10/22/2025, 1:22 PMMarvin
10/22/2025, 1:22 PMprefect.logging.get_logger() when there’s no flow/task context. get_run_logger() only works inside a running flow/task.
Practical patterns
- Basic (no flow context):
from prefect.logging import get_logger
logger = get_logger(__name__)
<http://logger.info|logger.info>("This will log to your console with Prefect formatting.")
- Write-once logger that works both inside and outside flows:
from prefect.logging import get_logger, get_run_logger
def get_prefect_logger(name: str | None = None):
try:
# Works only when a flow/task run context is active
return get_run_logger()
except Exception:
# Fallback for scripts / `flow.fn(...)` / tests, etc.
return get_logger(name)
# Use everywhere
logger = get_prefect_logger(__name__)
<http://logger.info|logger.info>("Hello from anywhere")
Notes
- When you bypass orchestration (e.g., calling flow.fn(...)), there’s no run context, so get_run_logger() won’t be available. Use the fallback pattern above.
- Without a flow/task run, logs won’t be attached to a run in the Prefect UI. They’ll go to your console using Prefect’s formatting/redaction. If you need logs in Cloud tied to a run, execute within a flow/task.
- You can adjust log verbosity via environment variables, e.g.:
export PREFECT_LOGGING_LEVEL=INFO