Build Your Objects in One Place at Startup

A module opens its database connection as soon as you import it:

# shop/db.py
import os
import sqlite3

connection = sqlite3.connect(os.environ["SHOP_DB"])

Importing shop.db now opens a file. On a machine without SHOP_DB set, which includes most test machines, the import fails with KeyError. Nothing ever closes the connection, and a test cannot swap in a different one without patching.

Create objects in main and pass them in

The composition root is the one place, usually main, where the program creates its objects and passes them to each other. Everything else receives what it needs as an argument, as in the previous article.

import sqlite3
from contextlib import closing


class OrderStore:
    def __init__(self, db: sqlite3.Connection) -> None:
        self._db = db

    def add(self, item: str) -> None:
        self._db.execute("insert into orders (item) values (?)", (item,))

    def count(self) -> int:
        (n,) = self._db.execute("select count(*) from orders").fetchone()
        return int(n)


def main(path: str) -> None:
    with closing(sqlite3.connect(path)) as db:
        db.execute("create table if not exists orders (item text)")
        store = OrderStore(db)
        store.add("book")
        print(store.count())


if __name__ == "__main__":
    main(":memory:")  # prints 1

Importing this module opens nothing. main creates the connection, so main closes it. OrderStore receives the connection, so it uses it and leaves it open. A test is a composition root too. It builds an OrderStore around its own in-memory database.

With several resources, use contextlib.ExitStack. It closes each one in reverse order, even when the block raises an exception:

import sqlite3
from concurrent.futures import ThreadPoolExecutor
from contextlib import ExitStack, closing

with ExitStack() as stack:
    db = stack.enter_context(closing(sqlite3.connect(":memory:")))
    pool = stack.enter_context(ThreadPoolExecutor(max_workers=4))
    # pass db and pool to the objects that need them
    print(pool.submit(len, "ok").result())  # 2

Register each resource right after you create it. Then a failure while creating the next one still closes the first.

Close what you create, and leave open what you receive

Code that creates a resource owns it and closes it. Code that receives a resource borrows it and leaves it open, because the caller still uses it. This function follows the rule for a thread pool:

from collections.abc import Iterator
from concurrent.futures import Executor, ThreadPoolExecutor
from contextlib import contextmanager


@contextmanager
def executor_scope(executor: Executor | None = None) -> Iterator[Executor]:
    if executor is None:
        with ThreadPoolExecutor(max_workers=2) as owned:
            yield owned
    else:
        yield executor


with ThreadPoolExecutor(max_workers=1) as shared:
    with executor_scope(shared) as borrowed:
        assert borrowed.submit(len, "hello").result() == 5
    assert shared.submit(len, "ok").result() == 2  # still open

When the caller passes shared, executor_scope borrows it and leaves it running. When the caller passes nothing, executor_scope creates a pool and owns it, and its own with block shuts the pool down. The exercise applies the same rule to a class.

Common mistakes

Closing a borrowed resource. If a class shuts down a pool that the caller passed in, the caller's next submit raises RuntimeError. Close only what you created, and say in the docstring which arguments you borrow.

Expecting with sqlite3.connect(...) to close the connection. A connection's with block commits or rolls back the transaction, and it leaves the connection open. Wrap the connection in contextlib.closing() to close it.

Going further

Libraries name their cleanup call differently. A SQLAlchemy engine has dispose(), which you can register with stack.callback(engine.dispose). See the contextlib.ExitStack and sqlite3 documentation.