# app/db.py
engine = create_engine(os.environ["DATABASE_URL"])
Three characters of convenience, and the following are now true of your project:
-
import app.dbfails withoutDATABASE_URL, so the test suite cannot collect without one, so CI needs a database to run the unit tests. - The engine is created once per process at import, before anything has had a chance to configure logging, set a pool size, or decide it is running in a dry-run mode.
- There is no way to have two of them — one for the primary and one for the replica — because the identity is a module attribute.
- Nothing can shut it down cleanly, because nothing owns it.
Every one of those is caused by when the object was built, not by what it is.
The rule
Exactly one place constructs the real object graph, and it runs when the
program runs — not when a module is imported. That place is conventionally
main(). Everything else receives what it needs as an argument.
def main(argv: Sequence[str]) -> int:
config = load_config(os.environ)
with ExitStack() as stack:
engine = stack.enter_context(create_engine(config.database_url))
http = stack.enter_context(httpx.Client(timeout=config.timeout))
rates = SqlRateSource(engine)
audit = HttpAuditSink(http)
return run(config, rates, audit)
Read what that bought. run takes its collaborators as parameters, so a test
passes fakes. Nothing is constructed at import, so import app is free and
side-effect-free. ExitStack unwinds in reverse order, so the HTTP client
closes before the engine disposes, and it does so on the exception path too.
ExitStack is the part people skip. Nested with statements do the same
thing when the count is fixed and small; ExitStack handles the case where
the number of resources depends on configuration, and it lets you build them
in a loop.
💡A test needs a real database and a fake payment gateway. With a composition root, where does the wiring for that live — and what must NOT change? click to reveal
In the test. It builds its own graph: the real engine, a FakePaymentGateway,
and calls the same run(config, rates, audit) that main() calls.
What must not change is run. If the test has to pass a flag, set an
environment variable, or monkeypatch a module attribute to get the fake in,
then the seam is not a parameter and the composition root is not doing its
job. The signature of the use case is the contract, and both main() and the
test are just two callers of it.
This is the practical test for whether you have a composition root at all:
can a test construct a different graph without touching production code? If
yes, you do. If it needs a if os.environ.get("TESTING") anywhere, you do
not.
Own what you create, borrow what you are given
The rule that keeps a library usable inside somebody else’s composition root. Consider a class that needs an executor:
class BatchProcessor:
def __init__(self, fn, *, executor: Executor | None = None) -> None:
self.owns_executor = executor is None
self.executor = ThreadPoolExecutor() if executor is None else executor
def __exit__(self, *exc: object) -> None:
if self.owns_executor:
self.executor.shutdown(wait=True)
The owns_executor flag is the whole design. If the caller passed one in, it
belongs to their composition root — they may be sharing it with three other
components and shutting it down at the end of main(). Shutting it down
inside your __exit__ is not tidy; it is you breaking their program from
inside a library.
If you created it, you must dispose of it, or you leak threads or processes for the life of the process.
The same flag pattern applies to every borrowable resource: an HTTP client, a connection pool, a logger handler, a temporary directory.
💡A library exposes Client(session: requests.Session | None = None) and always calls session.close() in its own close(). What breaks, and how would you notice?
click to reveal
An application that shares one Session across four clients — which is
exactly what connection pooling is for — has its session closed the first
time any one of them is closed. The other three then fail with
connection-pool or closed-session errors, at some unrelated later point in
the program.
You notice late and badly: the failure surfaces in a different component from the one that caused it, only when more than one client exists, and typically only in production where the sharing is real. In tests, where each test makes its own client, it never reproduces.
The fix is four lines: record whether you created the session, and close it only if you did. And document it — “if you pass a session, you own its lifetime” belongs in the docstring, because the caller cannot infer it.
What import time is for
A module body should define things and nothing else: functions, classes,
constants, type aliases. Not connections, not pools, not configuration
parsing, not logging.basicConfig(), not registration into a global
registry.
The test is mechanical and worth running on your own codebase: in a fresh
interpreter, import yourpackage with the environment empty. If it raises,
or if it opens anything, or if it takes 400 ms, the module body is doing work
that belongs in main().