3 · Reliability and delivery · 30 MIN

FastAPI dependencies and cleanup

A yielded dependency owns a resource’s request lifetime.

A dependency can acquire a resource, yield it to an endpoint and release it in finally. This example provides a small in-memory SQLite connection for each request and closes it afterwards. Each connection gets an independent demo dataset; nothing is durable across requests. A production service should use a configured database pool and transaction policy. The synchronous route and dependency keep blocking SQLite work out of an async function; async def does not make blocking database calls nonblocking.

Verify parameters are bound rather than interpolated into SQL.

Read the example

# main.py
import sqlite3
from fastapi import Depends, FastAPI
app = FastAPI()
def database():
    connection = sqlite3.connect(":memory:", check_same_thread=False)
    try:
        connection.execute("CREATE TABLE courses(id INTEGER, title TEXT)")
        connection.execute("INSERT INTO courses VALUES(?, ?)", (1, "React"))
        yield connection
    finally:
        connection.close()
@app.get("/courses")
def courses(db=Depends(database)):
    return [{"id": row[0], "title": row[1]} for row in db.execute("SELECT id, title FROM courses ORDER BY id")]

Every GET /courses returns the demo React row; the per-request connection is closed afterwards.

Your challenge

Override the dependency in tests with controlled data, inject a query failure and verify resource cleanup still runs.

Solution cost: O(p) rows returned plus query cost. time · O(p) response rows and this tiny per-request database. space

Common trap

check_same_thread=False permits thread access; it is not a complete concurrency or transaction strategy.

# main.py
import sqlite3
from fastapi import Depends, FastAPI
app = FastAPI()
def database():
    connection = sqlite3.connect(":memory:", check_same_thread=False)
    try:
        connection.execute("CREATE TABLE courses(id INTEGER, title TEXT)")
        connection.execute("INSERT INTO courses VALUES(?, ?)", (1, "React"))
        yield connection
    finally:
        connection.close()
@app.get("/courses")
def courses(db=Depends(database)):
    return [{"id": row[0], "title": row[1]} for row in db.execute("SELECT id, title FROM courses ORDER BY id")]

Further reading: Official documentation