<@ULVA73B9P> how to call a flow as a normal python...
# ask-marvin
j
@Marvin how to call a flow as a normal python function, without "flow" functionality, but still connect to Prefect cloud for pulling secrets
m
thought for 261.4 seconds
Great question. In Prefect 3.x you can run your flow’s underlying Python function directly (bypassing orchestration) and still talk to Prefect Cloud to fetch variables/secrets. Two complementary patterns: 1) Call the flow’s underlying function with .fn() - Any
@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):
Copy code
prefect cloud login --key <YOUR_API_KEY> --workspace <account_handle>/<workspace_handle>
- Or set environment variables:
Copy code
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:
Copy 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:
Copy code
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):
Copy code
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)?
j
@Marvin How to use Prefect's logger if there is no flow context?
m
thought for 54.4 seconds
Short answer: use
prefect.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):
Copy code
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:
Copy code
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.:
Copy code
export PREFECT_LOGGING_LEVEL=INFO