Async with FastAPI

informix-driver has a native async API. Use it from FastAPI by creating the pool in the app’s lifespan and dependency-injecting connections per request.

App skeleton

from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends, HTTPException
from informix_db import aio

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.pool = await aio.create_pool(
        host="db.example.com",
        user="informix", password="...",
        database="mydb", server="informix",
        min_size=2, max_size=20,
    )
    yield
    await app.state.pool.close()

app = FastAPI(lifespan=lifespan)


async def get_conn():
    async with app.state.pool.connection() as conn:
        yield conn


@app.get("/users/{user_id}")
async def get_user(user_id: int, conn = Depends(get_conn)):
    cur = await conn.cursor()
    await cur.execute(
        "SELECT id, name, email FROM users WHERE id = ?",
        (user_id,),
    )
    row = await cur.fetchone()
    if row is None:
        raise HTTPException(404, "user not found")
    return {"id": row[0], "name": row[1], "email": row[2]}

Why this shape

• Lifespan-scoped pool: the pool lives for the lifetime of the app, login handshake amortized across all requests.
• Per-request connection via Depends: each request gets its own connection from the pool. The async generator pattern (yield conn) means the connection returns to the pool when the request finishes, including on exception.
• No run_in_executor: every call is awaitable natively. No event-loop blocking, no thread-pool tuning.

Cancellation

If a client disconnects mid-request, FastAPI cancels the task. informix-driver is cancellation-safe — the cancellation propagates cleanly, the in-flight worker is reaped, and the connection returns to the pool clean (transactions rolled back). You don’t need to wrap anything in try/finally.

Note

This is a Phase 27 invariant: async cancellation cannot leak running workers onto recycled connections. The earlier behavior was a High audit finding; the fix is a CI tripwire test that’s been green every commit since.

Connection-level transactions

For request-scoped transactions, use a context manager around the connection:

@app.post("/orders")
async def create_order(order: OrderIn, conn = Depends(get_conn)):
    async with conn.transaction():
        cur = await conn.cursor()
        await cur.execute(
            "INSERT INTO orders VALUES (?, ?, ?)",
            (order.id, order.customer_id, order.total),
        )
        await cur.execute(
            "UPDATE inventory SET qty = qty - ? WHERE sku = ?",
            (order.qty, order.sku),
        )
    return {"ok": True}

The transaction commits on normal exit and rolls back on any exception, including HTTPException.