FastAPI with Async PostgreSQL: Pools, Transactions, and Real Errors

Why This Combination

FastAPI’s own tutorial covers routing, request bodies and validation thoroughly, so this lesson assumes you’ve seen @app.get and a pydantic model before. What the tutorial keeps deliberately small is the database layer. This lesson connects FastAPI to a real PostgreSQL 17 server with asyncpg: one connection pool for the whole process, validation errors that stop bad input before SQL runs, constraint errors turned into proper HTTP status codes, a multi-statement write that is safe to fail halfway, and a measurement of what the pool does under load.

Start PostgreSQL in Docker and install the Python side:

docker run -d --name pg -p 5432:5432 -e POSTGRES_PASSWORD=secret -e POSTGRES_DB=shop postgres:17
pip install fastapi "uvicorn[standard]" asyncpg httpx

Every output below came from fastapi 0.143.0, starlette 1.7.0, pydantic 2.14.0, uvicorn 0.54.0, asyncpg 0.32.0 and httpx 0.28.1 on Python 3.12. The PostgreSQL track covers the database side in depth.

One Pool per Process, Opened in Lifespan

The schema lives in schema.sql. The CHECK constraints are deliberate: they are what the API will hit later.

CREATE TABLE IF NOT EXISTS products (
    id    serial PRIMARY KEY,
    sku   text UNIQUE NOT NULL,
    name  text NOT NULL,
    price numeric(10, 2) NOT NULL CHECK (price > 0),
    stock int NOT NULL DEFAULT 0 CHECK (stock >= 0)
);
CREATE TABLE IF NOT EXISTS orders (
    id         serial PRIMARY KEY,
    product_id int NOT NULL REFERENCES products (id),
    quantity   int NOT NULL
);

db.py creates the pool in FastAPI’s lifespan hook and hands each request a borrowed connection through a dependency:

import os
from contextlib import asynccontextmanager
import asyncpg
from fastapi import FastAPI, Request

DSN = os.environ.get("DATABASE_URL", "postgresql://postgres:secret@localhost:5432/shop")

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.pool = await asyncpg.create_pool(DSN, min_size=2, max_size=5)
    async with app.state.pool.acquire() as conn:
        await conn.execute(open("schema.sql").read())   # idempotent: CREATE ... IF NOT EXISTS
    print("pool open:", app.state.pool.get_size(), "connections")
    yield
    await app.state.pool.close()
    print("pool closed")

async def get_conn(request: Request):
    async with request.app.state.pool.acquire() as conn:   # borrowed for one request
        yield conn                                           # returned to the pool after the response

app.py holds the models and the first two routes:

from decimal import Decimal
from fastapi import Depends, FastAPI, HTTPException
from pydantic import BaseModel, Field
from db import get_conn, lifespan

app = FastAPI(lifespan=lifespan)

class ProductIn(BaseModel):
    sku: str = Field(min_length=3, max_length=20)
    name: str
    price: Decimal = Field(gt=0, max_digits=10, decimal_places=2)
    stock: int = Field(default=0, ge=0)

class Product(ProductIn):
    id: int

@app.post("/products", status_code=201, response_model=Product)
async def create_product(p: ProductIn, conn=Depends(get_conn)):
    row = await conn.fetchrow(
        "INSERT INTO products (sku, name, price, stock) VALUES ($1, $2, $3, $4) RETURNING *",
        p.sku, p.name, p.price, p.stock)
    return dict(row)

@app.get("/products/{product_id}", response_model=Product)
async def get_product(product_id: int, conn=Depends(get_conn)):
    row = await conn.fetchrow("SELECT * FROM products WHERE id = $1", product_id)
    if row is None:
        raise HTTPException(404, f"product {product_id} not found")
    return dict(row)

Start the server, then ask PostgreSQL from another terminal how many connections the app holds before any request has arrived:

$ uvicorn app:app
INFO:     Started server process [1]
INFO:     Waiting for application startup.
pool open: 2 connections
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

$ docker exec pg psql -U postgres -d shop -c "SELECT count(*) AS app_connections FROM pg_stat_activity WHERE datname = 'shop' AND pid <> pg_backend_pid()"
 app_connections 
-----------------
               2
(1 row)

pool open printed before Application startup complete, so uvicorn didn’t accept traffic until the schema existed and two connections were ready. Those are min_size=2: real PostgreSQL backends, opened once and kept. The pool grows towards max_size=5 only when concurrent requests need more.

The alternative is await asyncpg.connect(...) inside every handler. connect_cost.py measures what that costs, against borrowing from a pool, over 20 requests each:

import asyncio
import os
import time
import asyncpg

DSN = os.environ.get("DATABASE_URL", "postgresql://postgres:secret@localhost:5432/shop")

async def main():
    start = time.perf_counter()
    for _ in range(20):                      # a new connection per "request"
        conn = await asyncpg.connect(DSN)
        await conn.fetchval("SELECT 1")
        await conn.close()
    per_connect = (time.perf_counter() - start) / 20 * 1000

    pool = await asyncpg.create_pool(DSN, min_size=2, max_size=5)
    start = time.perf_counter()
    for _ in range(20):                      # borrow an already-open connection
        async with pool.acquire() as conn:
            await conn.fetchval("SELECT 1")
    per_acquire = (time.perf_counter() - start) / 20 * 1000
    await pool.close()

    print(f"connect + query + close: {per_connect:6.2f} ms per request")
    print(f"pool acquire + query:    {per_acquire:6.2f} ms per request")

asyncio.run(main())

$ python connect_cost.py
connect + query + close:   6.73 ms per request
pool acquire + query:      0.33 ms per request

A new connection cost about 20 times as much as a pooled one, even with PostgreSQL on the same machine. Each connect pays a TCP handshake, PostgreSQL forking a new backend process, and SCRAM password authentication. Two more runs gave 6.20 / 0.44 ms and 7.82 / 0.27 ms. Over a real network every round trip adds to the first number, and connecting per request also puts no upper bound on how many connections a traffic spike opens.

Validation Runs Before Any SQL

Create a product, read it back, and ask for one that doesn’t exist and one whose id isn’t a number:

$ curl -s -w '\nHTTP %{http_code}\n' localhost:8000/products -H 'content-type: application/json' -d '{"sku": "KB-101", "name": "Mechanical keyboard", "price": 89.90, "stock": 12}'
{"sku":"KB-101","name":"Mechanical keyboard","price":"89.90","stock":12,"id":1}
HTTP 201
$ curl -s -w '\nHTTP %{http_code}\n' localhost:8000/products/1
{"sku":"KB-101","name":"Mechanical keyboard","price":"89.90","stock":12,"id":1}
HTTP 200
$ curl -s -w '\nHTTP %{http_code}\n' localhost:8000/products/99
{"detail":"product 99 not found"}
HTTP 404
$ curl -s -w '\nHTTP %{http_code}\n' localhost:8000/products/abc
{"detail":[{"type":"int_parsing","loc":["path","product_id"],"msg":"Input should be a valid integer, unable to parse string as an integer","input":"abc"}]}
HTTP 422

The request sent 89.90 as a JSON number, and the response sent back "89.90", a string. Pydantic v2 serializes Decimal as a string so the value survives the trip exactly, and asyncpg already returned PostgreSQL’s numeric as a Decimal. /products/abc never reached the handler: product_id: int failed to parse, and FastAPI answered 422 with the location ["path", "product_id"].

Now two bodies that break the model’s rules:

$ curl -s -w '\nHTTP %{http_code}\n' localhost:8000/products -H 'content-type: application/json' -d '{"sku": "K1", "price": -5, "stock": 3}'
{"detail":[{"type":"string_too_short","loc":["body","sku"],"msg":"String should have at least 3 characters","input":"K1","ctx":{"min_length":3}},{"type":"missing","loc":["body","name"],"msg":"Field required","input":{"sku":"K1","price":-5,"stock":3}},{"type":"greater_than","loc":["body","price"],"msg":"Input should be greater than 0","input":-5,"ctx":{"gt":0}}]}
HTTP 422
$ curl -s -w '\nHTTP %{http_code}\n' localhost:8000/products -H 'content-type: application/json' -d '{"sku": "MS-202", "name": "Mouse", "price": 19.999}'
{"detail":[{"type":"decimal_max_places","loc":["body","price"],"msg":"Decimal input should have no more than 2 decimal places","input":19.999,"ctx":{"decimal_places":2}}]}
HTTP 422

The first 422 lists all three problems in one response, not just the first one found, so a client can fix everything in one round trip. The second is the useful one for money: numeric(10, 2) would have silently rounded 19.999 to 20.00 on insert, but decimal_places=2 rejects it before the INSERT runs. Having the same rule in the model and the column means the API refuses what the database would otherwise quietly change.

The Failure: A Constraint Error Becomes a 500

Validation checks the shape of one request. It can’t know that KB-101 is already taken, because only the database knows that:

$ curl -s -w '\nHTTP %{http_code}\n' localhost:8000/products -H 'content-type: application/json' -d '{"sku": "KB-101", "name": "Another keyboard", "price": 49}'
Internal Server Error
HTTP 500

The server terminal shows why:

$ uvicorn app:app
...
INFO:     127.0.0.1:33382 - "POST /products HTTP/1.1" 500 Internal Server Error
ERROR:    Exception in ASGI application
...
asyncpg.exceptions.UniqueViolationError: duplicate key value violates unique constraint "products_sku_key"
DETAIL:  Key (sku)=(KB-101) already exists.

The second ... replaces about 40 lines of traceback through starlette, fastapi and asyncpg. The only frame from your code is File "/app/app.py", line 19, in create_product, the fetchrow call. The database did its job, but the client got a 500 and no explanation, so it can’t tell “you sent a duplicate” from “the server is broken”, and monitoring will count it as a server error. Every asyncpg constraint error (unique, check, foreign key, not-null) subclasses IntegrityConstraintViolationError, so one handler covers all of them. Append it to app.py:

import asyncpg
from fastapi import Request
from fastapi.responses import JSONResponse

@app.exception_handler(asyncpg.IntegrityConstraintViolationError)
async def constraint_violation(request: Request, exc: asyncpg.IntegrityConstraintViolationError):
    return JSONResponse(status_code=409,
                        content={"detail": exc.detail or exc.message, "constraint": exc.constraint_name})

$ curl -s -w '\nHTTP %{http_code}\n' localhost:8000/products -H 'content-type: application/json' -d '{"sku": "KB-101", "name": "Another keyboard", "price": 49}'
{"detail":"Key (sku)=(KB-101) already exists.","constraint":"products_sku_key"}
HTTP 409

409 Conflict says the request was well-formed but conflicts with the current state, and the body names the constraint. That works better than checking SELECT ... WHERE sku = $1 before inserting: two concurrent requests can both pass that check, and only the constraint is race-free.

The Failure: Half an Order

An order inserts a row into orders and takes the quantity off stock. Here is the first version, appended to app.py:

class OrderIn(BaseModel):
    product_id: int
    quantity: int = Field(gt=0)

@app.post("/orders", status_code=201)
async def create_order(o: OrderIn, conn=Depends(get_conn)):
    order_id = await conn.fetchval(
        "INSERT INTO orders (product_id, quantity) VALUES ($1, $2) RETURNING id",
        o.product_id, o.quantity)
    await conn.execute(
        "UPDATE products SET stock = stock - $1 WHERE id = $2", o.quantity, o.product_id)
    return {"order_id": order_id}

Order 5 keyboards (stock is 12), then 20, then look at the tables:

$ curl -s -w '\nHTTP %{http_code}\n' localhost:8000/orders -H 'content-type: application/json' -d '{"product_id": 1, "quantity": 5}'
{"order_id":1}
HTTP 201
$ curl -s -w '\nHTTP %{http_code}\n' localhost:8000/orders -H 'content-type: application/json' -d '{"product_id": 1, "quantity": 20}'
{"detail":"Failing row contains (1, KB-101, Mechanical keyboard, 89.90, -13).","constraint":"products_stock_check"}
HTTP 409
$ docker exec pg psql -U postgres -d shop -c "SELECT o.id, o.quantity, p.stock FROM orders o JOIN products p ON p.id = o.product_id ORDER BY o.id"
 id | quantity | stock 
----+----------+-------
  1 |        5 |     7
  2 |       20 |     7
(2 rows)

The client was told the second order failed, with a 409 from products_stock_check, but order 2 is in the table anyway. A warehouse reading orders will ship 20 keyboards that were never taken off stock. asyncpg runs each statement in its own autocommit transaction unless you say otherwise, so the INSERT was committed before the UPDATE failed. Wrap both statements in one transaction:

class OrderIn(BaseModel):
    product_id: int
    quantity: int = Field(gt=0)

@app.post("/orders", status_code=201)
async def create_order(o: OrderIn, conn=Depends(get_conn)):
    async with conn.transaction():          # COMMIT on success, ROLLBACK on any exception
        order_id = await conn.fetchval(
            "INSERT INTO orders (product_id, quantity) VALUES ($1, $2) RETURNING id",
            o.product_id, o.quantity)
        await conn.execute(
            "UPDATE products SET stock = stock - $1 WHERE id = $2", o.quantity, o.product_id)
    return {"order_id": order_id}

Reset the data with DELETE FROM orders; UPDATE products SET stock = 12 WHERE id = 1, restart uvicorn, and repeat the same two orders:

$ curl -s -w '\nHTTP %{http_code}\n' localhost:8000/orders -H 'content-type: application/json' -d '{"product_id": 1, "quantity": 5}'
{"order_id":3}
HTTP 201
$ curl -s -w '\nHTTP %{http_code}\n' localhost:8000/orders -H 'content-type: application/json' -d '{"product_id": 1, "quantity": 20}'
{"detail":"Failing row contains (1, KB-101, Mechanical keyboard, 89.90, -13).","constraint":"products_stock_check"}
HTTP 409
$ docker exec pg psql -U postgres -d shop -c "SELECT o.id, o.quantity, p.stock FROM orders o JOIN products p ON p.id = o.product_id ORDER BY o.id"
 id | quantity | stock 
----+----------+-------
  3 |        5 |     7
(1 row)

The client sees the same 409, but now the database agrees with it: no orphan order. The CheckViolationError escaped the async with block, asyncpg sent ROLLBACK, and the exception handler from the previous section still turned it into a 409. The new order is 3, not 1, because DELETE doesn’t reset a serial sequence. The transactions and locking lesson covers what PostgreSQL does when two of these transactions update the same row at once.

What the Pool Does Under Load

max_size=5 is a hard ceiling. To see what happens beyond it, append a route whose query takes one second and returns the PostgreSQL backend process id that served it:

@app.get("/report")
async def report(conn=Depends(get_conn)):
    pid = await conn.fetchval("SELECT pg_backend_pid() FROM pg_sleep(1)")   # a 1-second query
    return {"backend_pid": pid}

load.py fires ten requests at once and records when each one finished:

import asyncio
import time
import httpx

async def one(client, start):
    r = await client.get("/report")
    return round(time.perf_counter() - start, 2), r.json()["backend_pid"]

async def main():
    async with httpx.AsyncClient(base_url="http://localhost:8000", timeout=30) as client:
        start = time.perf_counter()
        results = await asyncio.gather(*(one(client, start) for _ in range(10)))
    for seconds, pid in sorted(results):
        print(f"done at {seconds:.2f}s  backend_pid={pid}")
    print("distinct connections used:", len({pid for _, pid in results}))

asyncio.run(main())

$ python load.py
done at 1.09s  backend_pid=111
done at 1.09s  backend_pid=112
done at 1.09s  backend_pid=114
done at 1.10s  backend_pid=113
done at 1.10s  backend_pid=115
done at 2.09s  backend_pid=111
done at 2.10s  backend_pid=112
done at 2.10s  backend_pid=113
done at 2.10s  backend_pid=114
done at 2.10s  backend_pid=115
distinct connections used: 5

No request failed. Five ran immediately, and the other five waited inside pool.acquire() until a connection came back, then reused the same five backends (111-115). That queue is the pool protecting PostgreSQL, whose default max_connections is 100 for every client combined. Size it from the database’s budget, not from the request rate: four uvicorn workers with max_size=5 each can hold 20 connections. Queued requests still add latency, as the 2.10s rows show, so a pool that is regularly full means the queries need to get faster or the database needs more connections.

Practice

1. With the transactional /orders route in place, predict the order_id of the next successful order before you place one. Then order 1 keyboard and check.

$ curl -s -w '\nHTTP %{http_code}\n' localhost:8000/orders -H 'content-type: application/json' -d '{"product_id": 1, "quantity": 1}'
{"order_id":5}
HTTP 201

2. Add GET /products?min_stock=N that lists products with at least N in stock and rejects a negative N with a 422. Create an out-of-stock mouse first so the filter has something to exclude.

from fastapi import Query

@app.get("/products", response_model=list[Product])
async def list_products(min_stock: int = Query(default=0, ge=0), conn=Depends(get_conn)):
    rows = await conn.fetch("SELECT * FROM products WHERE stock >= $1 ORDER BY id", min_stock)
    return [dict(r) for r in rows]

$ curl -s -w '\nHTTP %{http_code}\n' localhost:8000/products -H 'content-type: application/json' -d '{"sku": "MS-202", "name": "Mouse", "price": 19.99, "stock": 0}'
{"sku":"MS-202","name":"Mouse","price":"19.99","stock":0,"id":4}
HTTP 201
$ curl -s -w '\nHTTP %{http_code}\n' 'localhost:8000/products?min_stock=1'
[{"sku":"KB-101","name":"Mechanical keyboard","price":"89.90","stock":6,"id":1}]
HTTP 200
$ curl -s -w '\nHTTP %{http_code}\n' 'localhost:8000/products?min_stock=-1'
{"detail":[{"type":"greater_than_equal","loc":["query","min_stock"],"msg":"Input should be greater than or equal to 0","input":"-1","ctx":{"ge":0}}]}
HTTP 422

3. Add DELETE /products/{product_id} that returns 204 on success and 404 if nothing was deleted. Delete the mouse twice, then try to delete the keyboard. Which status does the keyboard get, and which code produces it?

@app.delete("/products/{product_id}", status_code=204)
async def delete_product(product_id: int, conn=Depends(get_conn)):
    status = await conn.execute("DELETE FROM products WHERE id = $1", product_id)
    if status == "DELETE 0":
        raise HTTPException(404, f"product {product_id} not found")

$ curl -s -w '\nHTTP %{http_code}\n' -X DELETE localhost:8000/products/4

HTTP 204
$ curl -s -w '\nHTTP %{http_code}\n' -X DELETE localhost:8000/products/4
{"detail":"product 4 not found"}
HTTP 404
$ curl -s -w '\nHTTP %{http_code}\n' -X DELETE localhost:8000/products/1
{"detail":"Key (id)=(1) is still referenced from table \"orders\".","constraint":"orders_product_id_fkey"}
HTTP 409

4. Change the pool to max_size=10 in db.py, restart uvicorn, and run load.py again. Predict the finish times first.

$ python load.py
done at 1.09s  backend_pid=120
done at 1.10s  backend_pid=121
done at 1.10s  backend_pid=122
done at 1.10s  backend_pid=124
done at 1.10s  backend_pid=125
done at 1.10s  backend_pid=128
done at 1.11s  backend_pid=123
done at 1.11s  backend_pid=126
done at 1.11s  backend_pid=127
done at 1.11s  backend_pid=129
distinct connections used: 10