Blocks: Config and Secrets

Every pipeline has a second codebase hiding inside it: the credentials, endpoints, bucket names, and warehouse handles that the real code needs but that don’t belong in the real code. Airflow files this under connections and variables — a connection is a bag of fields (host, login, password, a JSON extra for whatever didn’t fit), and a hook is the class that knows how to read that bag and talk to the thing. The split works, but the connection itself is barely typed, and the knowledge of what its fields mean lives somewhere else entirely. Prefect collapses that split. A block is the config and the client in one typed object.

A block is a class, not a bag

The simplest block is a secret. The bookshop’s warehouse password lives in Prefect, encrypted, and your flow loads it by name:

from prefect import task
from prefect.blocks.system import Secret

@task
def load_orders():
    password = Secret.load("bookshop-db-password").get()
    conn = connect(host="warehouse.internal", password=password)
    ...

Secret.load("bookshop-db-password") fetches the block from the Prefect API; .get() decrypts and returns the value. The password never appears in the flow file, never lands in git, and never shows in a log — it exists as an encrypted value server-side and arrives only at the moment .get() runs. That’s the same guarantee an Airflow connection’s password field gives you, with one difference: Secret is a class, load returns an instance of it, and the type is right there in the import. You’re not pulling a field out of a dict and hoping it’s the field you meant.

You create the block once, either in the UI or in code:

from prefect.blocks.system import Secret

Secret(value="hunter2").save("bookshop-db-password")

Run that once — from a REPL, a setup script, wherever — and the block is registered. From then on every flow loads it by name. Saving from the UI is the more common path for real secrets, precisely because it keeps the value out of any file you’d commit. Pass overwrite=True to .save() when you’re updating one that already exists.

Integration blocks carry the client, not just the config

Where blocks pull ahead of connections is the provider integrations. Prefect’s prefect-* collections ship blocks that bundle credentials and a working client for a specific system. Install prefect-aws and you get S3Bucket; install prefect-snowflake and you get SnowflakeConnector. The block isn’t a description of how to reach S3 — it is the S3 handle:

from prefect_aws import S3Bucket

@task
def stage_extract(rows: bytes):
    bucket = S3Bucket.load("bookshop-staging")
    bucket.write_path("orders/today.csv", content=rows)

S3Bucket.load(...) gives you an object with methods — write_path, read_path, list_objects — already wired to the right bucket and credentials. Compare the Airflow shape: a connection holding the AWS keys, plus an S3Hook you instantiate with that connection’s id, plus your memory of which hook method does what. In Prefect the credentials and the operations are the same object, and the type tells you what operations exist. The SnowflakeConnector block works the same way — load it, get a connection you can query, with the account, warehouse, and auth already baked in.

Because blocks are classes, they compose. An S3Bucket block holds an AwsCredentials block; a SnowflakeConnector holds a SnowflakeCredentials block. You configure the credentials once and point every resource block at them, instead of copying the same keys into a dozen connections. That nesting is not a special feature — it’s just a block field whose type is another block, and it’s the same mechanism you get when you write your own.

Writing your own block

The moment your system isn’t one of the two hundred things a prefect-* collection already covers, you subclass Block and declare your fields. This is where blocks stop being a nicer connection and start being something Airflow has no direct answer for. Say the bookshop talks to a small in-house “royalties” API that wants a base URL, a numeric timeout, and an API key. In Airflow you’d stuff the URL into a connection’s host, the key into password, and the timeout into the JSON extra — and then write a hook to dig them back out. In Prefect you write the shape down once:

from typing import Optional
import httpx
from prefect.blocks.core import Block
from pydantic import SecretStr


class RoyaltiesAPI(Block):
    """Client for the bookshop royalties service."""

    _block_type_name = "Royalties API"
    _logo_url = "https://bookshop.example/static/royalties.png"

    base_url: str
    api_key: SecretStr
    timeout_seconds: int = 30
    verify_tls: bool = True

    def _client(self) -> httpx.Client:
        return httpx.Client(
            base_url=self.base_url,
            headers={"Authorization": f"Bearer {self.api_key.get_secret_value()}"},
            timeout=self.timeout_seconds,
            verify=self.verify_tls,
        )

    def royalties_for(self, isbn: str) -> dict:
        with self._client() as client:
            resp = client.get(f"/titles/{isbn}/royalties")
            resp.raise_for_status()
            return resp.json()

The fields are ordinary pydantic type annotations, so validation is free: pass a string where timeout_seconds wants an int and the block refuses to save. base_url is required because it has no default; verify_tls defaults to True. _block_type_name is the human label the UI shows (without it Prefect derives one from the class name); _logo_url is the icon on the block’s card. Both are class attributes with a leading underscore, which is Prefect’s convention for “this is block metadata, not a savable field” — they don’t become part of the stored document.

The two methods are the whole point. _client is private plumbing; royalties_for is the operation a flow actually wants. Once this block is saved, a task never constructs an httpx client, never assembles an auth header, never remembers the base URL:

@task
def enrich_royalties(isbn: str) -> dict:
    api = RoyaltiesAPI.load("bookshop-royalties")
    return api.royalties_for(isbn)

That’s the shape every first-party integration block has — S3Bucket.write_path, SnowflakeConnector.get_connection — and now yours has it too, with the same load-by-name ergonomics and the same UI card.

Nesting blocks

Because a block field is just a typed attribute, a field can be another block, and that’s how the first-party integrations avoid repeating credentials. You can do the same in your own code. Suppose the royalties service and a second internal service share the same OAuth credentials — model the credentials as their own block and reference it:

class ServiceCredentials(Block):
    _block_type_name = "Bookshop Service Credentials"
    client_id: str
    client_secret: SecretStr


class RoyaltiesAPI(Block):
    _block_type_name = "Royalties API"
    base_url: str
    credentials: ServiceCredentials   # a nested block
    timeout_seconds: int = 30

Save a ServiceCredentials instance once, and every resource block that needs it points at the same document. When you load the outer block, Prefect resolves the nested one for you — RoyaltiesAPI.load("bookshop-royalties").credentials.client_secret is a live SecretStr, hydrated from its own encrypted record. Rotate the secret in one place and every consumer picks it up on its next run. That’s the composition that S3Bucket → AwsCredentials and SnowflakeConnector → SnowflakeCredentials are doing under the hood; you don’t so much learn a new feature as notice the field types already allow it.

SecretStr and SecretDict

Note that api_key is a SecretStr, not a str. This is the field-level version of the Secret block, and it does two jobs. First, the value is stored encrypted alongside the rest of the block. Second — and this is the part that saves you at 2 a.m. — pydantic refuses to render it. If a RoyaltiesAPI instance gets logged, printed, or shown in the UI, the key comes out as SecretStr('**********'). You have to call .get_secret_value() to see the real string, which is exactly why the _client method does so explicitly and nothing else can do it by accident. A plain str field would sit in your logs the first time someone prints the block during debugging.

When the sensitive thing is a whole document rather than a single string — a service-account JSON, a map of per-region keys — use SecretDict:

from prefect.blocks.core import Block
from pydantic import SecretStr
from prefect.types import SecretDict


class WarehouseCredentials(Block):
    account: str
    password: SecretStr
    connect_args: SecretDict  # every value obfuscated, keys visible

SecretDict obfuscates the values while leaving the keys readable, so the UI can show you that there’s a private_key entry without showing you the key itself. Mix secret and non-secret fields freely; only the ones you type as secret get the encryption-and-masking treatment.

Registering the type

A block type has to exist on the server before the UI can offer a form for it. First-party collections register themselves on install, but a block you wrote lives in your code, so you register it explicitly:

prefect block register -m bookshop.blocks

-m takes the module path; Prefect imports it, finds every Block subclass, and registers the type (its name, its field schema, its logo) with the API. After that the “Royalties API” card shows up in the UI’s block catalog and someone can fill in a bookshop-royalties instance by hand — no code required to create the value, only to define the shape. You register the type once; you save instances as often as you like. If you change the class — add a field, tighten a type — re-run register so the server’s schema matches the code.

Schema evolution is the part worth being deliberate about. When you add a field, give it a default so existing saved instances still validate against the new schema; a required field with no default will make older documents fail to load until someone fills it in. Prefect keeps track of a block type’s schema as it changes so the server can tell whether a stored instance matches the code that’s trying to load it, but — as the caveat later in this chapter stresses — that bookkeeping is not a rollback history you can browse to un-break a bad edit. Treat a block class the way you’d treat any shared interface: additive changes are cheap, renames and required-field additions are migrations, and the register step is when the server learns about them. In CI, a prefect block register -m bookshop.blocks after deploy keeps the registered schema honest without anyone remembering to run it by hand.

The CLI

Blocks have a small prefect block command surface that’s worth knowing before you reach for the UI:

prefect block type ls          # every registered block type
prefect block ls               # every saved block instance, with slug/name
prefect block inspect s3-bucket/bookshop-staging   # one instance's fields
prefect block delete s3-bucket/bookshop-staging    # remove it
prefect block register -m prefect_aws              # register a whole collection

The s3-bucket/bookshop-staging form — type-slug/instance-name — is the block’s stable identifier, and it’s what shows up in deployment YAML and error messages. inspect prints the non-secret fields and masks the secret ones, so it’s safe to run in a shared terminal. register -m prefect_aws is how you’d surface every AWS block type after installing the collection, if for some reason the auto-registration didn’t run.

Block, Variable, Secret, or parameter?

Four things in Prefect can hold a value your flow reads, and using the wrong one is the most common config mistake. Here’s the decision matrix:

Mechanism	Holds	Typed?	Encrypted?	Reach for it when
Parameter	Per-run input	Yes (flow signature)	No	The value changes every run — a date, a batch id, a filename
Variable (prefect.variables.Variable)	Small JSON config	Loosely (JSON)	No	A shared knob many flows read — a feature flag, a threshold, a region list
Secret block	One sensitive string	Yes	Yes	A bare credential with no client attached — a token, a password
Custom / integration block	Typed config + methods	Yes	Per-field	An external resource with behavior — a bucket, a warehouse, an API

The line people trip over is Variable versus block. prefect.variables.Variable is the true analogue of an Airflow Variable: a small JSON-ish value, stored in plaintext, meant for runtime knobs that many flows share and that ops folks tweak without a deploy. It is not the place for a credential (no encryption) and not the place for a large document or a client object (it has no type and no methods). Blocks are the opposite: typed, method-bearing, encrypted where you ask for it, and built to carry a whole resource. Variables get their own chapter precisely because the two are easy to conflate and shouldn’t be — if you find yourself putting a password in a Variable or JSON-encoding a config dict into one, you wanted a block. And parameters are the axis orthogonal to both: they’re the inputs that change per run, not the config that changes on a slower clock than the code.

The integrations catalog

“Airflow has more providers” is the honest first objection from anyone with a mature Airflow install, and it deserves a straight answer. Airflow’s provider ecosystem is genuinely enormous. Prefect’s is smaller in raw count but covers the systems most pipelines actually touch, and each integration ships blocks that do more than describe a connection. The survey:

• prefect-aws — S3Bucket, AwsCredentials, ECSTask, plus Secrets Manager and Batch helpers.
• prefect-gcp — GcsBucket, GcpCredentials, BigQueryWarehouse, Cloud Run and Vertex infrastructure.
• prefect-azure — Blob Storage, AzureContainerInstanceCredentials, ML and Cosmos helpers.
• prefect-snowflake — SnowflakeConnector, SnowflakeCredentials.
• prefect-databricks — credentials plus jobs blocks for triggering and polling Databricks runs.
• prefect-dbt — runs a dbt project from inside a flow, and it’s the one integration this track’s readers are most likely to reach for, so it gets code below rather than a bullet.
• prefect-kubernetes — credentials and job blocks, the backbone of the Kubernetes work pool.
• prefect-github — repository storage blocks for pulling flow code at run time.
• prefect-slack — a webhook/credentials block for notifications, wired into automations.

Installing one is the ordinary pip step, and the blocks it ships register themselves so they appear in the UI catalog immediately:

uv pip install prefect-aws prefect-snowflake

prefect-dbt, concretely

That bullet list is a catalog, and a catalog is not an answer. So here is the one integration this track actually needs, with code — because the bookshop we keep loading is the same bookshop the dbt series models, and “how do I run my dbt project from a flow?” is the first question a reader of both will ask.

Ignore the older material telling you to reach for a dbt block. The current API is a runner:

from prefect import flow
from prefect_dbt import PrefectDbtRunner, PrefectDbtSettings


@flow
def bookshop_transform():
    runner = PrefectDbtRunner(
        settings=PrefectDbtSettings(
            project_dir="/opt/dbt/bookshop",
            profiles_dir="/opt/dbt/bookshop",
        ),
        raise_on_failure=True,
    )
    runner.invoke(["build", "--select", "marts"])

invoke takes the dbt CLI arguments as a list — it is dbt build --select marts, run in-process through dbt’s programmatic entrypoint rather than shelled out. Which buys you the thing that matters: prefect-dbt reads dbt’s own event stream and turns each model, seed, and test into a tracked Prefect task run. You don’t get one opaque green square that means “dbt did something”; you get the dbt DAG, node by node, in the Prefect UI, with the failing model named. That is the same argument Cosmos makes on the Airflow side, arrived at from the other direction — and it’s why raise_on_failure=True matters: without it, a failed dbt node leaves your flow cheerfully green.

The comparison worth holding: Airflow’s Cosmos parses your manifest.json at parse time to build the DAG, which is why that series spends a chapter on parse-time cost. prefect-dbt doesn’t need to — the graph is discovered while it runs, because in Prefect the graph is always discovered while it runs. Same problem, and the dynamic-execution model dissolves it rather than solving it. If you’ve read the dbt-on-Airflow series, this is the paragraph that pays it off.

The catalog breadth matters, but the doubling matters more. In Airflow a provider gives you operators and hooks — things you call from task code. A Prefect block from the same integration is also your deployment storage (a GitHubRepository block tells a worker where to pull code), your result storage (an S3Bucket block is where task results get persisted, the subject of the next chapter), and your infrastructure config (an ECSTask or Kubernetes job block configures where a run executes). One typed object is credential, client, code source, and result store depending on where you point it. Concretely, the same S3Bucket block you write extracts into can be handed to a deployment as its result store:

from prefect import flow
from prefect_aws import S3Bucket

@flow(result_storage=S3Bucket.load("bookshop-staging"))
def nightly_load():
    ...

Nothing new was configured — the bucket credentials the pipeline already uses are now also where Prefect persists task results, because a block is a value you can pass wherever a resource is expected. So the provider-count comparison undersells the design: a single block type is load-bearing across the parts of Prefect that Airflow spreads across connections, extra_dag_run config, remote logging settings, and executor config. Fewer moving parts, each doing more.

The security model

Blocks are only as safe as where they’re stored, and the guarantee differs by deployment. On Prefect Cloud, secret fields are encrypted at rest and the values are write-only through the API surface — the UI and CLI mask them, and the plaintext is never returned except to a running flow that calls .get() / .get_secret_value(). On a self-hosted Prefect server, encryption depends on your database and the server’s configuration; the server encrypts secret fields, but you own the key management and the database’s own at-rest protection. The practical rule: on self-hosted, treat the Prefect database with the same care you’d give an Airflow metadata database that holds Fernet-encrypted connections, because that’s exactly the analogous asset. Either way, a SecretStr field is masked in transit through logs and UI — the obfuscation is a client-and-server behavior, not just a display trick — so the failure mode where a credential leaks into a task log is closed by construction, not by remembering to be careful.

There’s a corollary for who can read a secret. Anyone whose flow can .load() a block and call .get_secret_value() has the plaintext — that’s the whole point, the running flow needs it. So the boundary that actually protects a credential is the boundary around who can run code against your workspace and who can edit blocks in the UI, not the encryption itself. On Cloud that’s workspace membership and roles; on self-hosted it’s whatever authentication you’ve put in front of the API and the database. The mental model to carry over from Airflow is exact: a Fernet-encrypted connection is safe from someone reading the metadata database directly, and useless as protection against someone who can already trigger a DAG that uses it. Blocks inherit that shape. Encrypt at rest, mask in logs, and then spend your real attention on access to the control plane.

Loading blocks in async flows

Everything so far used the synchronous .load(). In an async flow, load is awaitable, and so are the block methods that do I/O:

from prefect import flow
from prefect_aws import S3Bucket

@flow
async def stage_async(rows: bytes):
    bucket = await S3Bucket.load("bookshop-staging")
    await bucket.awrite_path("orders/today.csv", content=rows)

await S3Bucket.load(...) fetches the block without blocking the event loop, and integration blocks expose async variants of their I/O methods (aread_path / awrite_path on S3Bucket, for instance) so a highly concurrent flow that fans out over hundreds of keys doesn’t serialize on the network. For your own blocks, if a method does async I/O, write it async def and load the block with await — the framework doesn’t force a choice, it follows yours.

Why this matters past the syntax

The real payoff is that your configuration becomes inspectable and editable without touching code. Every block shows up in the Prefect UI: you can see that bookshop-staging points at the right bucket, rotate the warehouse password on the Secret block, and correct a typo’d endpoint — all without a deploy, because the flow reads the block fresh at run time. That’s the same operational win Airflow connections give you, and it’s the right instinct: config that changes on a different schedule than code shouldn’t live in the code.

One operational caveat: treat block updates as replacements, not as a friendly undo stack. Prefect block schemas are versioned as types evolve, and .save(..., overwrite=True) is the explicit signal that you are replacing the document stored under that name, but you should not depend on a user-facing rollback history to rescue a bad credential edit. Manage important block definitions the same way you manage Airflow connections: document the intended fields, rotate through a controlled process, and keep the source of truth somewhere your team can audit. Keep secrets in Secret blocks, SecretStr fields, or credentials blocks — never in a flow parameter, never in a plain variable, never in the source. The rule is the same one you already follow with Airflow connections; blocks just give you a typed, extensible object to follow it with.

Final thoughts

Airflow’s connection-plus-hook split made sense for a system where the scheduler is generic and the integrations are bolted on: keep the credential dumb and let a hook interpret it. Prefect started from the assumption that config is code you didn’t want to write, and gave it types, classes, inheritance, methods, and a UI — so a bucket is an object that writes to a bucket, an API is an object that answers questions, and a secret is an object that hands you a secret only when you ask by name. It’s a small idea with a large blast radius: once your configuration is typed and loadable by name, the rest of your pipeline stops carrying strings it has to remember the meaning of.

Next: Retries, Timeouts, and the Failures Worth Repeating