Keyboard shortcuts

Press ← or → to navigate between chapters

Press S or / to search in the book

Press ? to show this help

Press Esc to hide this help

Patterns & Style

What using this well looks like from Python, and the mistakes that read fine. The Rust book's Patterns & Style covers the ones that are about the engine; these are the ones that are about Python.

One configuration per subsystem

db = DynamicConfig(Database, key="db").file("config.toml").env("APP_")
cache = DynamicConfig(Cache, key="cache").file("config.toml").env("APP_")
flags = DynamicConfig(Values, key="flags").file("config.toml")

Three objects over one file, and they share nothing: three schemas, three reloads, three failures that do not touch each other. A broken flags section leaves the database's document serving.

The flag table has no schema on purpose. Its keys are a product decision, and a class that had to declare each one would be edited every time somebody added a flag — which is what Values is for.

Read current() where you use it

@app.get("/")
def index():
    db = config.current()          # here, not at import time
    return {"host": db.host}

current() is an attribute read on a cached instance — validation ran once, at the reload. A module-level DB = config.current() is a value that has stopped reloading, and it is the one mistake this library cannot stop you making.

In FastAPI, the dependency is the configuration object:

def database() -> Database:
    return config.current()

@app.get("/")
def index(db: Annotated[Database, Depends(database)]):
    ...

Let the schema be the schema you already have

Pydantic if the program uses Pydantic, a dataclass if it does not want a dependency, msgspec if the shape is hot, Values if there is nothing to declare. The engine does not care, and swapping one for another changes no other line. What a schema may be.

Declare secrets where the field is, not in a list somewhere else: SecretStr, field(metadata={"secret": True}), or msgspec.Meta(extra={"secret": True}). One declaration drives the redaction in the cache, in explain, and in a scrubbed validation error.

Hooks are for waking something, not for doing the work

config.on_reload(lambda previous, current: pool.resize(current.pool.max_size))

By default a hook runs inside the reload, on the thread that noticed the change — often the watcher's, and in an asyncio program that is not the event loop. Anything slow, and anything that awaits, should say so:

async def follow():
    async for db in config.changes():
        await pool.resize(db.pool.max_size)

@config.on_reload_async                       # or as a hook, on this loop
async def resize(previous, current):
    await pool.resize(current.pool.max_size)

Both run on the loop, and in both the reload was over before the work started. Which to pick: changes() when the work is the service's own loop, on_reload_async when it belongs next to the thing it rebuilds — and dispatch=Dispatch.EXECUTOR when the slow work is not async at all.

One lifecycle for the configurations that share one

group = ConfigGroup(db, cache, flags)

with group.running():
    serve()

Five configurations mean five init() calls, five watchers and five handles to stop in the right order — none of which is application logic. When a deployment moves them together, group.reload_atomic() is the call that refuses to leave two of them new and one of them old.

The group is lifecycle only. db.current() is still the read path, because nothing should sit between a program and its values.

One rule bounds what atomic means: the writes are all-or-nothing, the reads are not — two current() calls are two reads, and a reader landing between the group's installs can see new db beside old cache for an instant. Two values that must always be read together belong in one configuration, where a single snapshot carries both.

Testing without a filesystem

with config.overrides(rate_limit=1, mode="test"):
    assert something_that_reads() == "test at 1/s"

Three doors, and none of them writes a file: overrides for a block, load() for a candidate nobody installs, and defaults for the twelve fields a test does not care about. The shipped pytest plugin gives you a scratch directory and a clean environment as fixtures.

What to check in CI, and what at startup

QuestionWhere
Does the committed file still parse and validate?CI: DynamicConfig(...).check()
Does this deployment's configuration load?startup — init(), and let it raise
Is the store reachable?a health endpoint, not a startup gate
@app.get("/healthz")
def healthz():
    status = config.status()
    code = 200 if status.consecutive_failures == 0 else 503
    return JSONResponse(status.__dict__, status_code=code)

Type checking

The stubs are shipped, so mypy --strict sees your model through current(). Two habits keep that true:

  • Annotate the configuration object: config: DynamicConfig[Database] when it is a module-level name, so the generic parameter does not get lost.
  • try_current() when it may not be installed: it is Database | None, which is what the checker wants at a boundary where current() would raise.