Async Databases

Almost every web service spends most of its time waiting on a database. In an asyncio program that wait is dangerous in one specific way: the event loop runs on a single thread, so one blocking call from psycopg2, sqlite3 or the synchronous redis-py client freezes every other coroutine until it returns. The tests pass, the demo feels fast, and then the server stalls under real load. An async driver avoids this because it yields at network I/O, letting the loop serve other tasks while Postgres or Redis does its work.

Going async does not make a single query faster. A 200 ms query is still 200 ms; the gain is that many requests can wait on I/O together. That also moves the bottleneck: the database and its connection pool, not the loop, now cap your throughput. Much of this chapter is about that second half, covering how pools are sized, why 1000 tasks do not mean 1000 parallel queries, and what goes wrong when sessions, transactions or timeouts are handled carelessly.

By the end you will be able to talk to Postgres with asyncpg (connections, pools, prepared statements) and with the SQLAlchemy 2.0 AsyncSession, choose a transaction isolation level and retry safely, and spot N+1 and lazy-loading traps before they reach production. You will also size a pool against max_connections, use aiosqlite and redis.asyncio correctly, and put deadlines on every call so that a timeout or cancellation leaves connections and transactions clean.

An asyncio program runs every coroutine on a single thread, and the event loop decides who runs next. The loop can only switch tasks when the running task reaches an await. If a coroutine calls a blocking function such as cursor.fetchall() from psycopg2, sqlite3, or the synchronous redis-py client, the thread is stuck inside that call. Every other coroutine waits until it returns, including ones that have nothing to do with the database.

An async driver behaves differently. It sends the query over the socket and then awaits the reply. While it waits, the task is suspended and the loop is free to run something else. Only await hands control back to the loop, so a call that never awaits can never share the thread.

The program below makes the difference visible without needing a database. A background ticker adds one tick every 0.1 s. The blocking query uses time.sleep, which holds the thread like a sync driver does. The async query uses asyncio.sleep, which suspends like await conn.fetch(...) does. Both pretend to take 0.35 s.

The ticker stands in for every other request your server is handling. With the blocking call it gets no turns at all for 0.35 s. With the awaited call it keeps ticking. The sequence below is what happens each time you await a real driver call.

In a real handler it looks like the fragment below. The only change from a sync driver is that the call is awaited, and that single keyword is what lets the loop continue.

async def get_user(conn, uid):
    # conn is an asyncpg connection; awaiting suspends this task only
    return await conn.fetchrow('SELECT * FROM users WHERE id = $1', uid)

Fragment: needs a live asyncpg connection, so it is shown without running.

It is easy to expect async to make queries faster. It does not. A query that takes 200 ms inside Postgres still takes 200 ms when you await it. What async changes is what the rest of your program does during those 200 ms. Instead of one request occupying the thread, many requests can all be waiting on I/O at once.

The next program scales the numbers down to 0.1 s per query so it finishes quickly. A fake query awaits asyncio.sleep, standing in for the network round trip. We time one query, then ten awaited in a row, then ten started together with asyncio.gather.

A single query is no faster than before. The gain only appears when many requests overlap. If one endpoint is slow because its query is slow, the fix is the query, an index, or a cache. Changing the driver will not help.

Starting a thousand coroutines does not give you a thousand parallel queries. Postgres serves them over a limited number of connections, and an asyncpg or SQLAlchemy pool hands those out. If 1000 coroutines share 10 connections, 10 run and 990 wait on the pool. A pool behaves like a semaphore, so we can model it with one. Here 20 tasks share a pool of 5.

Twenty tasks at five per round take four rounds of 0.1 s. Waiting on the pool costs almost nothing, because a waiting coroutine uses no thread, so a small pool is normal. Growing it only helps while the database has spare capacity. Pool sizing is covered in its own section later in this chapter.

Each common database has a blocking driver and an async counterpart. Pick the async one for anything that runs inside async def.

The driver families differ in how they stay async. asyncpg and redis.asyncio speak their wire protocols over non-blocking sockets, so the loop does the waiting. aiosqlite is the odd one out. SQLite is an in-process library with no network, so aiosqlite runs the normal sqlite3 calls on a dedicated thread and awaits the result. The loop stays free in either case.

Sometimes the library you need has no async version, such as a legacy ORM or a vendor SDK. In that case await asyncio.to_thread(fn, *args) runs the call in a worker thread and suspends your coroutine until it finishes. The loop thread never blocks. The program below runs three 0.2 s blocking calls at the same time.

The three calls overlapped because each ran on its own thread, so the loop was free the whole time. This is an escape hatch, not a goal. The default thread pool is small, so a hot path that sends thousands of calls through to_thread will queue behind it. Use an async driver whenever one exists.

The same idea applies to other blocking calls. Here is what each one should become.

Blocking calls are hard to spot by reading code, so asyncio can report them for you. Run your program with asyncio.run(main(), debug=True), or set the environment variable PYTHONASYNCIODEBUG=1. In debug mode the loop logs a warning such as Executing <Task ...> took 0.212 seconds whenever one step of a task hogs the thread for more than 100 ms. The logged task shows where to look.

To catch shorter stalls, lower the threshold from inside the running loop with asyncio.get_running_loop().slow_callback_duration = 0.05. This only has an effect in debug mode, so use it in development and not in production.

The symptoms you see in production also point to the cause. Use this table to tell a blocked loop from an exhausted pool or a plain slow query.

asyncpg talks to Postgres over its own binary protocol and never blocks the event loop. The simplest way in is a single connection: await asyncpg.connect(dsn) opens one network connection to the server and gives you a connection object. That object is backed by a real Postgres backend process, so you must give it back with await conn.close(). Put the close in a finally block so it runs even when a query raises.

# uses the asyncpg package; this block only defines a function
async def count_users(dsn):
    conn = await asyncpg.connect(dsn)
    try:
        return await conn.fetchval('SELECT count(*) FROM users')
    finally:
        await conn.close()

An unclosed connection keeps its Postgres backend alive until the process exits.

Once you have a connection, four methods cover almost every query. They differ only in the shape of what they hand back, so pick the one that matches the result you expect.

asyncpg uses Postgres-style positional placeholders: $1, $2 and so on. The %s of psycopg2 and the ? of SQLite are not understood and give a syntax error. The values are passed as separate arguments after the SQL string and travel to the server apart from the SQL text, so the server never parses them as SQL. That is what blocks injection: a value like '; DROP TABLE users; -- is just data.

async def find_user(conn, uid, org):
    r = await conn.fetchrow(
        'SELECT * FROM users WHERE id = $1 AND org = $2',
        uid, org)
    if r is None:
        return None
    return r['name'], r[0], dict(r)

The same Record read by column name, by position and as a plain dict.

A Record behaves like a light mix of tuple and dict. You can read a column by name with r['name'], by position with r[0], or turn the whole row into a dictionary with dict(r). Values arrive as real Python objects, because asyncpg decodes the binary protocol directly rather than handing you strings.

Opening a connection costs a network round trip plus authentication, and Postgres only allows a limited number of them. A pool opens connections once and lends them out. You create it a single time with await asyncpg.create_pool(dsn, min_size=10, max_size=10); both sizes default to 10. Each unit of work then borrows a connection with async with pool.acquire() as conn: and the connection returns to the pool when the block ends, even if it raised.

Inside the async with block you own that connection exclusively, so run every query that belongs together there. When you only need one query, the shortcuts pool.fetch(), pool.fetchrow(), pool.fetchval() and pool.execute() do the whole cycle in one call: acquire, run the query, release. Reach for them first, and use acquire() only when several queries must share one connection or one transaction.

A borrowed connection may have been changed by the previous user, so the pool cleans it before handing it to anyone else. It runs RESET ALL to undo session settings, UNLISTEN to drop notification subscriptions, and closes any cursors left open. Separately, max_inactive_connection_lifetime (300 seconds by default) closes connections that sit idle that long, and the pool opens new ones when demand returns.

Some setup belongs to each connection rather than to a query, such as teaching asyncpg how to decode jsonb. Pass an init= coroutine to create_pool and it runs once on every new connection before the pool uses it. By default asyncpg returns jsonb as a string; the codec below makes it return Python objects instead.

import json

async def init(conn):
    await conn.set_type_codec(
        'jsonb',
        encoder=json.dumps,
        decoder=json.loads,
        schema='pg_catalog')

async def start(dsn):
    return await asyncpg.create_pool(dsn, init=init)

A codec set inside a query would be erased by RESET ALL on release, so install it through init.

A Postgres connection speaks a strict request and reply conversation, so a connection serves one query at a time, and every concurrent task needs its own connection. If two coroutines await queries on the same connection at the same moment, asyncpg raises InterfaceError: cannot perform operation: another operation is in progress. The program below uses stand-in classes instead of asyncpg, so it runs anywhere. FakeConn raises the same error when it is busy, and FakePool hands each task a free connection from a queue, which is what a real pool does.

In shared(), the first fetch marks the connection busy and suspends at its sleep. The second fetch then starts on the same object and hits the error. In pooled(), each call takes a different connection from the queue, so both run side by side and both succeed. The fix in real code is the same: give each task its own connection through pool.fetch() or its own pool.acquire(), and never pass one connection to several tasks.

Inserting rows one at a time in a loop costs a network round trip per row, and the waiting dominates the runtime. asyncpg offers two ways to cut that. executemany() sends one statement with many sets of arguments in a batch. copy_records_to_table() uses the Postgres COPY protocol, which streams all the rows in a single operation and is often 10x or more faster than row-by-row INSERT.

async def load_events(conn, rows):
    # rows is a list of tuples, e.g. [(1, 'login'), (2, 'logout')]
    async with conn.transaction():
        await conn.copy_records_to_table(
            'events', records=rows, columns=['user_id', 'kind'])

async def load_few(conn, rows):
    await conn.executemany(
        'INSERT INTO events(user_id, kind) VALUES($1, $2)', rows)

Pass columns=[...] to COPY into a subset of the table's columns.

Wrapping a load in conn.transaction() makes it all or nothing: if one row is bad, none of them are kept. COPY cannot express ON CONFLICT, so when rows might already exist, use executemany() with an upsert statement, or COPY into a staging table first and merge from there.

Every SQL string you send to Postgres has to be parsed, analysed and planned before any row is read. A prepared statement splits that work from the execution. The server parses and plans the query once, keeps the result under a name, and each later call only sends new values for the $1, $2 placeholders. For a query you run thousands of times a second, skipping the repeated parse and plan work is a real saving.

You rarely have to ask for this in asyncpg, because the driver prepares every query automatically. When you call conn.fetch(), fetchrow() or execute() with a SQL string, asyncpg prepares it, runs it, and remembers the prepared statement in a cache that belongs to that one connection. The next time the same SQL text goes through the same connection, asyncpg skips the parse step and just executes.

The cache holds statement_cache_size=100 statements by default and behaves as an LRU: when a 101st distinct query arrives, the statement that was used least recently is thrown away. Because the cache lives on the connection, a pool does not share it. Ten pooled connections means ten separate caches, and each one warms up on its own. The small simulation below shows the eviction rule with a cache of two entries.

Query B had to be prepared a second time. Adding C pushed out B, which was the least recently used entry at that moment. In a real application the same thing happens if you build more than 100 distinct SQL strings per connection, for example by pasting values into the text instead of using placeholders.

When you want to hold on to a statement yourself, for instance to run it in a tight loop, prepare it explicitly. The returned object has the same fetch methods as a connection, minus the SQL argument.

The second call sends only the value 7. Keep in mind that stmt is tied to the connection that created it. If you prepare on one connection from a pool and later run the object after releasing that connection, you are no longer in control of who else is using it, so prepare and use the statement inside a single async with pool.acquire() block.

Planning depends on the values in the query. Postgres can look at $1, compare it with table statistics, and build a plan for that exact value. That is a custom plan. A generic plan is built without looking at the value, so it can be reused for every call. Postgres tries custom plans for the first five executions of a prepared statement, then compares the generic plan's estimated cost with the average of those custom plans. If the generic plan looks about as cheap, it may switch to it from the sixth execution onward to save planning time.

The trouble starts when the data is lopsided. Imagine an orders table where status = 'rare' matches 50 rows and status = 'common' matches nine million. With the real value in view, the planner uses an index scan for 'rare' and a sequential scan for 'common'. A generic plan has no value to look at, so it commits to one shape for both. If that shape is the sequential scan, your rare lookups become slow; if it is the index scan, the common lookups crawl through millions of random reads.

The symptom is a query that runs fast for a while and then becomes slow after roughly the sixth call on a connection, only for some parameter values. The fix is to tell Postgres never to switch: set plan_cache_mode to force_custom_plan. With asyncpg you can send it as a server setting so that every pooled connection gets it when it connects.

You pay a little planning time on each call in exchange for a plan that matches the value. For queries on evenly distributed columns the default auto is fine and you can leave it alone.

Prepared statements live inside one Postgres server connection, which is why they clash with connection poolers. PgBouncer in transaction mode hands a server connection to a client only for the length of one transaction, then gives it to someone else. asyncpg, however, prepares a statement and later executes it by name, assuming the same server connection is still behind its own socket. After PgBouncer moves you, that name is unknown.

The message names a statement such as __asyncpg_stmt_1__, the internal name asyncpg gave it. It often appears only under load, since with a quiet pooler you may keep getting the same server connection and everything looks fine in testing.

The classic fix is to turn the client-side statement cache off, so asyncpg never relies on a name that must survive across transactions. Set statement_cache_size=0 when connecting.

Through SQLAlchemy there are two caches to switch off. One is asyncpg's own cache; the other is a cache the SQLAlchemy asyncpg dialect keeps of its prepared statement names. Pass both through connect_args, otherwise the error can still come back.

Turning the cache off has a cost: every call is parsed again. A newer option keeps the performance. PgBouncer 1.21 and later can track protocol-level prepared statements itself and replay them on whichever server connection a client lands on. Enable it by setting max_prepared_statements to a value above zero in pgbouncer.ini; once it is on you can leave asyncpg's statement cache at its default.

A cached prepared statement is a promise about the shape of the table at the time it was planned. If a deploy runs ALTER TABLE and changes a column type or adds a column that a SELECT * would now return, Postgres refuses to run the old plan and asyncpg raises InvalidCachedStatementError ("cached plan must not change result type").

What happens next depends on where the call was made. Outside a transaction, asyncpg catches the error, throws away the stale statement, prepares it again and retries once, so your code never notices. Inside a transaction that is not possible: statements earlier in the transaction may have already run, and replaying only the failing one would silently break atomicity. asyncpg therefore lets the exception reach you, and you must retry the whole transaction yourself.

The failed attempt invalidates the cached statement, so the second pass prepares it afresh against the new table definition. One retry is enough; if it fails again something else is wrong and the error should surface. Naming the columns in the SELECT instead of using * also makes the problem much less likely, because adding a column then no longer changes the result type.

SQLAlchemy 2.0 ships with a full asyncio layer, so you can keep the ORM you already know and still never block the event loop. Everything starts with an async engine. You build it with create_async_engine, and the URL names the async driver after the plus sign. postgresql+asyncpg tells SQLAlchemy to use the asyncpg dialect from the earlier sections, and pool_size=5 keeps five connections open in the engine's pool.

from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

engine = create_async_engine(
    "postgresql+asyncpg://user:pw@host/db",
    pool_size=5,
)

Session = async_sessionmaker(engine, expire_on_commit=False)

Both objects are created at module level or in startup code, not inside a request.

The engine is not a connection. It is the owner of the pool, and it only borrows a connection when a session actually needs to talk to the database. The second object, async_sessionmaker, is a session factory: calling Session() hands you a fresh AsyncSession bound to that engine. Build it once at startup and share it across the whole application, because it holds configuration and no per-request state.

The expire_on_commit=False option deserves attention. By default a commit marks every loaded object as expired, so the next read of user.name triggers a reload from the database. In a sync session that reload happens silently. In an async session it cannot be awaited from a plain attribute access, so it fails. Turning expiry off keeps the values you already loaded readable after the commit.

Three objects are involved, and they live for very different lengths of time. Mixing them up is the root of most async ORM bugs, so it helps to keep this table in mind.

Most of the time you open a session and a transaction together. The idiom async with Session() as s, s.begin(): does both in one line. The async with on the session guarantees it is closed and its connection returns to the pool. The s.begin() part wraps your work in a transaction: it commits when the block ends cleanly and rolls back if an exception escapes.

Inside the block you write queries in the 2.0 style: build a statement with select(), then pass it to execute(). The call returns a result object, and scalars() unwraps each row into the ORM object itself, so all() gives you a plain list of User instances. For a lookup by primary key, s.get() is shorter, and it checks the session's identity map before it sends any SQL.

from sqlalchemy import select

async def adults_and_first():
    async with Session() as s, s.begin():
        s.add(User(name="ada", age=36))          # sync: only stages the object
        res = await s.execute(select(User).where(User.age > 30))
        users = res.scalars().all()
        first = await s.get(User, 1)             # primary-key lookup
        return users, first

Assumes a mapped `User` class with `name` and `age` columns.

The result object has several ways to read rows, and the right one depends on how many rows you expect.

Not every session method is a coroutine, and forgetting an await is an easy mistake. add() only places an object in the session, so it is an ordinary call. Anything that may touch the network has to be awaited.

execute() fetches the whole result before it returns, which is a problem for a table with millions of rows. stream() is the alternative. It uses a server-side cursor, so rows arrive in batches and you consume them with async for. Memory stays flat no matter how large the result is.

It is worth knowing how the ORM manages to be async at all. The ORM core is the same synchronous code that SQLAlchemy 1.x had. A greenlet bridges it to the async driver: when the sync code needs the network, the greenlet switches out, the driver awaits the I/O on the event loop, and execution switches back. This only works inside calls that go through the bridge, such as await s.execute().

If your code triggers database I/O outside that bridge, there is nothing to switch back to, and SQLAlchemy raises MissingGreenlet. The classic cause is reading an expired attribute or a lazy relationship such as user.posts straight from a plain attribute access. The cure is to load what you need up front or to await the load explicitly, which the N+1 section covers in depth.

Some SQLAlchemy APIs are synchronous only and cannot be awaited, for example schema creation and database inspection. For those, open an async connection and hand the sync function to run_sync, which runs it on the bridge with a sync-style connection.

async def create_tables():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

DDL helpers like create_all expect a sync connection, so run_sync supplies one.

An AsyncSession tracks one unit of work: one transaction, one set of loaded objects, one connection. It is not safe for concurrent tasks. If two tasks run queries on the same session at the same time, their operations overlap inside a single transaction, and SQLAlchemy raises errors such as IllegalStateChangeError or the driver reports an operation already in progress. The rule is simple: share the engine and the sessionmaker, never the session.

When tasks really do need to run in parallel, give each task its own session. The async with inside the helper opens a session for that task alone and returns its connection to the pool when the task finishes.

In a web framework the same idea becomes a dependency that opens one session per request. With FastAPI, a generator dependency yields the session and closes it after the response is sent.

The last piece is shutdown. The engine keeps pooled connections open, so call await engine.dispose() when the application stops, or the database sees those connections drop abruptly. In FastAPI the natural place is the lifespan function: create tables on the way in, dispose on the way out.

When something goes wrong, the error name usually points to the cause.

A transaction groups several statements so they either all take effect or none do. In asyncpg you open one with conn.transaction() and use it as an async context manager. When the block exits cleanly, asyncpg sends COMMIT. If an exception escapes the block, it sends ROLLBACK and then re-raises the exception, so your error handling still sees it.

async def move_funds(conn, src, dst, amount):
    async with conn.transaction():
        await conn.execute(
            'UPDATE accounts SET bal = bal - $1 WHERE id = $2', amount, src)
        await conn.execute(
            'UPDATE accounts SET bal = bal + $1 WHERE id = $2', amount, dst)
    # COMMIT has run here; an exception above would have rolled both back

Both updates commit together or not at all

Transaction blocks can nest. A second conn.transaction() inside an open one does not start a new transaction, because Postgres has no such thing. asyncpg turns it into a SAVEPOINT instead. If the inner block fails, only the work since that savepoint is rolled back, and the outer transaction can carry on and commit what came before.

async def import_rows(conn, rows):
    async with conn.transaction():          # BEGIN
        for row in rows:
            try:
                async with conn.transaction():  # SAVEPOINT
                    await conn.execute(
                        'INSERT INTO items VALUES ($1, $2)', *row)
            except Exception:
                pass  # only this row's savepoint is rolled back
    # the good rows are committed together

One bad row does not undo the others

The transaction() call also takes options. isolation= picks the isolation level, readonly=True forbids writes, and deferrable=True is meaningful only together with serializable and read-only. A deferrable transaction waits until Postgres can hand it a snapshot that cannot be part of a serialization failure, then runs without risk of being aborted. That makes it a good fit for a long report.

async def big_report(conn, query):
    tx = conn.transaction(
        isolation='serializable',
        readonly=True,
        deferrable=True,
    )
    async with tx:
        return await conn.fetch(query)

A long read-only scan that will never fail with 40001

An isolation level decides which other transactions' changes your transaction can see while it runs. Postgres offers three behaviours, and you choose between them per transaction or per engine.

The default is READ COMMITTED. Each statement sees the data that was committed before that statement began. Two identical SELECTs in the same transaction can therefore return different rows if someone commits in between. REPEATABLE READ takes a single snapshot when the transaction starts and uses it for every statement. In Postgres this blocks non-repeatable reads and also phantoms. SERIALIZABLE uses Serializable Snapshot Isolation (SSI), which tracks read and write dependencies between transactions and blocks every anomaly. The price is that Postgres may abort a transaction with a serialization failure.

You can also ask for READ UNCOMMITTED, but Postgres treats it exactly like READ COMMITTED. There are no dirty reads in Postgres at any level.

The asterisk matters: the SQL standard allows phantoms at REPEATABLE READ, but Postgres does not, because its snapshot covers the whole transaction. Do not assume the same on other databases.

SERIALIZABLE blocks every anomaly by aborting a transaction when it detects a dangerous pattern. The error is a SerializationFailure with SQLSTATE 40001. The database has already rolled the transaction back, so the fix is to rerun the whole transaction from BEGIN, not just the statement that raised. Back off a little between tries so the competing transactions can finish.

The loop below uses a fake transfer() that fails twice, so you can watch the retry logic without a database. With asyncpg you would catch asyncpg.SerializationError and wrap async with conn.transaction(isolation='serializable') inside the try.

SQLAlchemy offers the same ideas. engine.execution_options(isolation_level='REPEATABLE READ') returns a copy of the engine, leaving the original unchanged, and every connection checked out through the copy uses that level. Inside a session, s.begin_nested() creates a savepoint. It needs an outer s.begin(), and if the nested block fails only the savepoint is rolled back.

async def audit_in_savepoint(engine, Session, audit):
    rr_engine = engine.execution_options(isolation_level='REPEATABLE READ')
    async with Session(bind=rr_engine) as s, s.begin():
        try:
            async with s.begin_nested():   # SAVEPOINT
                s.add(audit)
        except Exception:
            pass  # savepoint rolled back, outer work still commits

Level set on an engine copy, savepoint inside the outer transaction

Isolation levels govern snapshots; row locks let you claim rows explicitly. select(...).with_for_update() emits SELECT ... FOR UPDATE, which locks the chosen rows until the transaction ends, so no one else can change them meanwhile. For a job queue, add skip_locked=True: workers skip rows that another worker has already locked instead of waiting, so each worker claims a different job.

async def claim_job(Session, Job, select):
    async with Session() as s, s.begin():
        q = (select(Job)
             .where(Job.state == 'new')
             .limit(1)
             .with_for_update(skip_locked=True))
        job = (await s.execute(q)).scalar_one_or_none()
        if job is not None:
            job.state = 'running'
        return job

FOR UPDATE SKIP LOCKED through SQLAlchemy

The raw SQL version works the same way through asyncpg, as long as it runs inside a transaction so the lock lasts until COMMIT.

Because await hands control back to the event loop, it is easy to forget that the transaction stays open while your coroutine is suspended. If you await an HTTP call inside async with conn.transaction():, the database sits waiting for you. Any row locks you took are held the whole time, and the pooled connection is pinned, so other tasks queue for the pool.

async def bad(conn, http, url, upd, ins):
    async with conn.transaction():
        await conn.execute(upd)
        r = await http.get(url)      # locks held while we wait
        await conn.execute(ins, r)

async def good(conn, http, url, upd, ins):
    r = await http.get(url)          # outside the transaction
    async with conn.transaction():
        await conn.execute(upd)
        await conn.execute(ins, r)

Do external I/O first, then open a short transaction

To find leaks, query pg_stat_activity for sessions in that state. An old xact_start means a transaction has been open for a long time without running anything.

As a safety net, set idle_in_transaction_session_timeout as a server setting, for example '10s'. Postgres then closes any session that stays idle inside a transaction longer than that, and the pool discards the dead connection and opens a fresh one on the next checkout.

The N+1 problem starts with one query that loads N parent rows, for example 100 users. You then read user.posts on each parent, and every read fires its own query. That is 1 + N round trips to Postgres, here 101. The code looks innocent, because the extra queries come from a plain attribute access. You only see them in the SQL log or in a slow endpoint.

The example below fakes the database with a list of query strings, so you can count round trips without a server. The first loop is the N+1 shape: one query per user. The second asks for all the posts in one IN query, which is what an eager loader does for you.

With the synchronous Session, a lazy load quietly runs the extra query when you touch the attribute. That is convenient, and it is also how N+1 hides. In an AsyncSession the attribute access is plain synchronous Python, so it has no way to await the query. Instead of silently querying, SQLAlchemy raises MissingGreenlet. The error is annoying, but it turns a hidden performance bug into an immediate failure that points at the line.

async def broken(s, User, select):
    users = (await s.scalars(select(User))).all()
    return users[0].posts   # implicit lazy load -> MissingGreenlet

Fragment: s is an AsyncSession and User is a mapped class.

The cure for N+1 is to decide up front which relationships a query needs, and tell SQLAlchemy with a loader option. There are two main strategies. They differ in how many queries they run and in what shape the data comes back, so the right one depends on the relationship.

selectinload is the default choice for collections. It runs your main query first, then a second query that fetches all the children whose foreign key is in the list of parent ids it just loaded. Two queries cover any number of parents, and the parent rows are never repeated in the result.

async def users_with_posts(s, User, select, selectinload):
    stmt = select(User).options(selectinload(User.posts))
    users = (await s.scalars(stmt)).all()
    return [(u.name, len(u.posts)) for u in users]   # no extra queries

Fragment: the posts are already loaded, so reading u.posts is safe.

joinedload folds the related rows into the main query with a LEFT OUTER JOIN. That fits a many-to-one link such as a post's author, because each post row carries exactly one author and nothing is duplicated. If you use it on a collection, each parent appears once per child row, and SQLAlchemy refuses to hand you those duplicates silently. You must call result.unique() before reading the rows.

Eager loading only helps if you remember it everywhere. To catch the places you forgot, make unplanned lazy loads fail on every code path. Declare the relationship with relationship(lazy="raise"), or add options(raiseload('*')) to a query. Any access to a relationship you did not load explicitly then raises straight away. This is especially useful in tests, because an endpoint that would have been N+1 in production fails in the test suite instead.

Sometimes you do want a lazy load, for example to fetch posts for one user you already hold. Mix the AsyncAttrs mixin into your declarative base (or a model class), and every relationship gets an awaitable twin under awaitable_attrs. Awaiting it runs the query visibly, so nothing is hidden. If the relationship is already loaded and you want fresh data, await s.refresh(user, ["posts"]) reloads just that relationship.

async def load_on_purpose(s, User):
    user = await s.get(User, 1)
    posts = await user.awaitable_attrs.posts   # explicit, awaited query
    await s.refresh(user, ['posts'])           # reload one relationship
    return posts

Fragment: User must include the AsyncAttrs mixin.

Commit has a lazy-loading trap of its own. By default a session expires every object on commit, so the next attribute read tries to refresh the row from the database. That read is plain synchronous access again, so it raises MissingGreenlet. Create the factory with async_sessionmaker(engine, expire_on_commit=False) and the attributes you already loaded stay readable after commit. You trade away automatic freshness, so reload with refresh() when you need current values.

Plain asyncpg has no ORM and no lazy loading, but it has the same N+1 shape: a loop that awaits fetchrow() once per id. The fix is to send the whole list of ids in one query. Postgres accepts an array parameter, so you write WHERE id = ANY($1::int[]) and pass an ordinary Python list. One round trip returns every matching row.

async def users_by_ids(conn, ids):
    rows = await conn.fetch(
        'SELECT * FROM users WHERE id = ANY($1::int[])',
        ids,                      # a Python list, sent as one array
    )
    return {r['id']: r for r in rows}

Fragment: conn is an asyncpg connection.

A second trap looks modern rather than naive. It is tempting to avoid the loop by firing all the lookups at once with asyncio.gather. On a single AsyncSession that does not help, and it is not safe.

async def tempting_but_wrong(s, User, ids):
    return await asyncio.gather(*(s.get(User, i) for i in ids))

Fragment: N queries on one shared session.

This still sends N separate queries. Worse, one session wraps one connection and one transaction, and it is not task-safe, so running many operations on it at the same time can raise IllegalStateChangeError or interleave state. The right answer is the same as before: ask the database for all the ids in a single IN query.

Every Postgres connection is a separate backend process on the server, and each one costs several MB of RAM plus a share of CPU scheduling. That is why the server caps them: max_connections defaults to 100, and 3 of those slots are reserved for superusers, so your applications can really count on 97. Past that point the database refuses new connections, and well before it, extra connections only add memory pressure and contention.

Your side of the connection is managed by a pool. SQLAlchemy's QueuePool has four settings that matter, and their defaults are worth knowing by heart because the default error message quotes them.

So by default one process can hold at most 5 + 10 = 15 connections, and the 16th caller waits up to 30 seconds.

In a threaded server you might need roughly one connection per thread. In asyncio the situation is different: thousands of coroutines can be in flight, but a coroutine only needs a connection while a query or transaction is actually running. The pool behaves like a semaphore: every coroutine that wants a connection waits in line, and waiting costs almost nothing because the event loop simply serves other tasks meanwhile. A small pool is therefore normal, not a sign of a misconfiguration.

Start from demand. Little's law says the number of connections in use at any moment is about the arrival rate multiplied by the average time each request holds a connection. Use the time the connection is checked out, not the full request time: a request that spends 200 ms calling another service but only 20 ms inside a transaction holds a connection for 20 ms.

rps = 500          # requests per second
hold = 0.02        # seconds a connection is held per request
needed = rps * hold
print(f"connections needed: {round(needed)}")

500 requests per second at 20 ms each keep about 10 connections busy.

A second, independent check comes from the PostgreSQL wiki. As a starting heuristic, the useful number of connections is about (CPU cores × 2) + effective spindles. A database server with 8 cores and one SSD lands near 17. This surprises people, but a bigger pool often makes things slower: more queries compete for the same CPU and disk, and every extra backend adds context switching and lock contention. If Little's law gives 10 and the heuristic gives 17, a pool in that range is plenty.

The pool belongs to one process, but max_connections belongs to the whole database. The budget is: app instances × workers per instance × (pool_size + max_overflow) must stay below max_connections minus the headroom you keep for admin work such as psql sessions, migrations and cron jobs. Note that max_overflow counts, because a burst can use it on every process at once.

def total(instances, workers, pool_size, max_overflow):
    return instances * workers * (pool_size + max_overflow)

limit = 100 - 3 - 7   # max_connections, superuser slots, admin headroom
for name, args in [("3 x 4, 5+10", (3, 4, 5, 10)),
                   ("2 x 4, 3+2", (2, 4, 3, 2))]:
    used = total(*args)
    print(f"{name}: {used} of {limit} -> fits: {used < limit}")

The same function checks the default pool and a shrunken one.

Three instances with four workers each and the default pool could open 180 connections against 90 available. The fix is to shrink the per-process pool: with 3 + 2 per process, two instances of four workers need only 2 × 4 × (3 + 2) = 40 connections. If you autoscale, budget for the maximum instance count, and when trimming, reduce max_overflow before pool_size.

A pool also keeps connections open for a long time, and a connection can die while it sits idle: the database restarts, a firewall drops idle flows, or a load balancer or proxy silently closes the link. The pool does not notice until you try to use it, and then your query fails. Three settings address this, each with a different cost.

Pre-ping sends a lightweight check before handing out a connection and replaces it if the check fails, so a restart of the database costs you a moment instead of a burst of errors. Recycling replaces connections before they get old enough to be dropped, so set the value below the shortest idle timeout on the path to the database; 1800 seconds is a common choice. Together they give a robust configuration for a direct connection:

engine = create_async_engine(
    url,
    pool_size=5,
    max_overflow=10,
    pool_timeout=30,
    pool_pre_ping=True,
    pool_recycle=1800,
)

Fragment: needs SQLAlchemy and a real database URL.

When an external pooler such as PgBouncer sits in front of Postgres, or when you run on serverless where processes come and go, an in-app pool just stacks a second pool on top of the first and wastes slots. Use NullPool, so each checkout opens a connection and each release closes it, and let the external pooler own the pooling. In PgBouncer's transaction mode also turn off asyncpg's statement caches, as the prepared statements section explained.

engine = create_async_engine(
    url,
    poolclass=NullPool,
    connect_args={"statement_cache_size": 0},
)

Fragment: the pooler caps the connections, not the application.

When the pool runs dry, a task waits pool_timeout seconds and then fails with an error that quotes your own settings. This is the signature symptom of an undersized pool or of sessions that never give their connection back.

Because both causes produce the same message, check for leaks first. If every session is opened with async with, and no HTTP call or other slow external work sits inside a transaction, the leak is unlikely and the pool really is too small for the load. The chart below walks through the usual order of questions.

Two tools help you see what is going on. In Postgres, SELECT state, count(*) FROM pg_stat_activity GROUP BY state; shows how many connections are active, idle, or stuck as idle in transaction; a growing last group means leaked sessions or held transactions. In the application, engine.pool.status() reports how many connections are checked out and how much overflow is in use. Before you raise pool_size, cut the hold time: fewer queries per checkout and no external calls inside a transaction reduce the need under Little's law.

The standard sqlite3 module is synchronous, so calling it directly inside async def freezes the event loop for as long as the query runs. aiosqlite fixes this without rewriting SQLite. Each connection gets its own dedicated background thread, and that thread runs the real stdlib sqlite3 code. The async methods you call, such as execute and commit, only put a request on that thread's queue and then await the answer.

The sketch below is a stripped-down version of the same idea, built only from the standard library so you can run it. The worker thread owns the sqlite3 connection, the coroutine hands it work through a queue, and a future carries the result back to the loop. The real library adds cursors, transactions and error handling on top of this shape.

import asyncio, queue, sqlite3, threading

class TinyAsyncDB:
    def __init__(self):
        self._jobs = queue.Queue()
        self.last_thread = None
        self._thread = threading.Thread(target=self._run, name='sqlite-worker', daemon=True)
        self._thread.start()

    def _run(self):
        conn = sqlite3.connect(':memory:')
        while (job := self._jobs.get()) is not None:
            fut, loop, sql, params = job
            try:
                rows = conn.execute(sql, params).fetchall()
                conn.commit()
                self.last_thread = threading.current_thread().name
                loop.call_soon_threadsafe(fut.set_result, rows)
            except Exception as exc:
                loop.call_soon_threadsafe(fut.set_exception, exc)
        conn.close()

    async def execute(self, sql, params=()):
        loop = asyncio.get_running_loop()
        fut = loop.create_future()
        self._jobs.put((fut, loop, sql, params))
        return await fut

    def close(self):
        self._jobs.put(None)
        self._thread.join()

async def main():
    db = TinyAsyncDB()
    await db.execute('CREATE TABLE t (n INTEGER)')
    await db.execute('INSERT INTO t VALUES (?)', (7,))
    rows = await db.execute('SELECT n FROM t')
    print('rows:', rows)
    print('event loop thread:', threading.current_thread().name)
    print('sqlite thread:', db.last_thread)
    db.close()

asyncio.run(main())

A toy version of what aiosqlite does for every connection

The loop never touches SQLite itself; it only awaits a future. That is why other coroutines keep running while a query executes. It also means a single aiosqlite connection still runs one statement at a time, because its worker thread is a single queue.

In real code you open the connection with async with aiosqlite.connect("app.db") as db:, which starts the thread on entry and stops it on exit. Run statements with await db.execute(sql, params), always passing values as parameters rather than formatting them into the SQL. Writes are not durable until you call await db.commit(); leaving the block without committing discards them. Setting db.row_factory = aiosqlite.Row lets you read columns by name, as in row["name"], instead of by position.

async def list_adults(min_age):
    import aiosqlite

    async with aiosqlite.connect('app.db') as db:
        db.row_factory = aiosqlite.Row
        await db.execute('INSERT INTO users (name, age) VALUES (?, ?)', ('ada', 36))
        await db.commit()  # nothing is saved without this

        async with db.execute('SELECT name, age FROM users WHERE age >= ?', (min_age,)) as cur:
            rows = await cur.fetchall()
        return [(r['name'], r['age']) for r in rows]

Fragment: call it from a running event loop, for example with asyncio.run(list_adults(18))

SQLite lets only one writer hold the database at a time. When a second connection tries to write while the first is mid-transaction, it waits for the lock. Python's sqlite3.connect has a default timeout of 5.0 seconds, and that wait is what acts as a busy timeout. You only see database is locked once that wait runs out, or immediately if the timeout is set to 0. Under real load, 5 seconds of waiting is often not enough, and the error shows up as a flaky failure.

Two settings address this. PRAGMA journal_mode=WAL switches SQLite to write-ahead logging, so readers no longer block the writer and the writer no longer blocks readers. The setting is stored in the database file, so it persists. PRAGMA busy_timeout=5000 (or the timeout argument of connect) tunes how many milliseconds a connection waits for the write lock before giving up. This one is per connection, so set it every time you open one.

The demo below uses plain sqlite3 so it can run anywhere. One connection starts a write and holds it open, while a second connection with a short 0.1 s timeout tries to write. The second connection gives up with database is locked, yet it can still read, because WAL keeps readers out of the writer's way.

With aiosqlite you run the same two pragmas right after opening the connection. Keep write transactions short, because every millisecond you hold the lock is a millisecond other writers spend waiting.

async def open_db():
    import aiosqlite

    db = await aiosqlite.connect('app.db')
    await db.execute('PRAGMA journal_mode=WAL')
    await db.execute('PRAGMA busy_timeout=5000')
    return db

Fragment: close the returned connection with await db.close()

If your app uses SQLAlchemy 2.0, the URL sqlite+aiosqlite:///app.db runs the same AsyncSession code on a local file with no server to install. That makes it handy for unit tests and for a first run on a laptop. Treat it as a stand-in, though: SQLite and Postgres behave differently in ways that matter, and a test that passes on one can still fail on the other.

def make_test_engine():
    from sqlalchemy.ext.asyncio import create_async_engine

    return create_async_engine('sqlite+aiosqlite:///app.db')

Swap the URL for postgresql+asyncpg://... in production

The practical consequence is that lock-related bugs, such as deadlocks, SELECT ... FOR UPDATE behaviour and isolation levels, will not show up on SQLite. Run at least part of your test suite against a real Postgres before you trust it.

The old standalone aioredis library was merged into redis-py in version 4.2, and it now lives there as redis.asyncio. You get the same commands as the sync client, but each one is a coroutine. The usual way to create a client is redis.asyncio.Redis.from_url(url, decode_responses=True). With decode_responses=True, replies come back as str instead of bytes, which is what you want in most application code.

async def cache_demo():
    import redis.asyncio as redis

    r = redis.Redis.from_url('redis://localhost:6379/0', decode_responses=True)
    await r.set('k', 'v', ex=60)   # expires after 60 seconds
    value = await r.get('k')       # 'v' as a str
    await r.aclose()               # redis-py 5.0.1 and later
    return value

Fragment: run it inside an event loop with a Redis server available

The ex=60 argument gives the key a 60 second lifetime, and aclose() releases the client's connections. On redis-py versions before 5.0.1 the method was called close(), so check which one your installed version provides.

Every awaited command costs a network round trip. When you have several commands to send, a pipeline queues them on the client and sends them together. With transaction=True the group is also wrapped in MULTI/EXEC, so the server runs the commands as one unit. Inside the block, the commands are queued without being awaited; only execute() is awaited, and it returns the list of replies.

A client is cheap to use but not free to create, because each one owns connections. Build one ConnectionPool for the whole application with ConnectionPool.from_url(url, max_connections=50), wrap it in a single Redis client, and share that client everywhere. Creating a new client for every request opens a new set of sockets each time and can overwhelm the Redis server under load.

async def count_hit():
    from redis.asyncio import ConnectionPool, Redis

    pool = ConnectionPool.from_url('redis://localhost:6379/0', max_connections=50, decode_responses=True)
    r = Redis(connection_pool=pool)   # create once at startup, share it

    async with r.pipeline(transaction=True) as p:
        p.incr('hits')                # queued, not sent yet
        p.expire('hits', 60)          # queued, not sent yet
        results = await p.execute()   # one round trip, MULTI/EXEC

    await r.aclose()                  # at shutdown
    await pool.disconnect()           # a pool you passed in is not closed for you
    return results

Fragment: in a real app the pool and client live for the whole process, not for one call

Because you supplied the pool yourself, aclose() does not tear it down. Call pool.disconnect() as well during application shutdown, for example in the lifespan handler.

Redis can push messages to subscribers, so a consumer does not have to poll. Call r.pubsub() to get a subscription object, await ps.subscribe("ch") to join a channel, and then iterate with async for msg in ps.listen(). The loop runs for as long as the subscription lives. The first item you receive is not data but a confirmation with type set to subscribe, so filter on msg["type"] == "message" before reading msg["data"]. If you would rather poll with a deadline, get_message(ignore_subscribe_messages=True, timeout=1.0) returns the next real message or None.

async def listen(r):
    ps = r.pubsub()
    await ps.subscribe('ch')
    try:
        async for msg in ps.listen():
            if msg['type'] == 'message':   # skip the subscribe confirmation
                print(msg['data'])
    finally:
        await ps.aclose()

Fragment: r is the shared client from the previous page. Run it as its own task, because it never ends on its own

Suppose you read a balance, subtract from it and write it back. Another client could change the balance between your read and your write, and you would overwrite its update. Optimistic locking solves this without holding a lock. You watch the key, read it, then queue your write between multi() and execute(). If anyone else touched the watched key in the meantime, Redis refuses the transaction and execute() raises WatchError. Your job is then to retry the whole block from the read.

async def spend(r, amount):
    from redis.exceptions import WatchError

    async with r.pipeline() as p:
        for _ in range(5):                 # cap the retries
            try:
                await p.watch('bal')
                bal = int(await p.get('bal'))
                p.multi()                  # from here commands are queued
                p.set('bal', bal - amount)
                await p.execute()          # raises WatchError if bal changed
                return bal - amount
            except WatchError:
                continue                   # someone else won; redo the read
    raise RuntimeError('balance kept changing')

Fragment: r is the shared client

Ignoring WatchError means a lost update, which is the very bug the check exists to prevent. Under heavy contention on one key, though, retries pile up and most attempts fail. In that case move the logic into a Lua script. Redis runs a script atomically on the server, so there is no watch, no retry loop and only one round trip.

async def spend_atomic(r, amount):
    script = r.register_script("""
    local bal = tonumber(redis.call('GET', KEYS[1]) or '0')
    if bal < tonumber(ARGV[1]) then return -1 end
    return redis.call('DECRBY', KEYS[1], ARGV[1])
    """)
    return await script(keys=['bal'], args=[amount])   # -1 means insufficient funds

Fragment: the check and the update happen in one atomic step

A database call can hang for many reasons: a lock that never frees, a query with a bad plan, or a network that silently dropped. Without a deadline the coroutine awaits forever, and it holds a pooled connection the whole time. The cure is a deadline at every layer, so whichever one is closest to the problem gives up first.

The simplest deadline lives in your own code. On Python 3.11+ wrap the awaits in async with asyncio.timeout(5):. If the block runs longer than five seconds, asyncio cancels the current await and the block raises the builtin TimeoutError. The example uses a fake query so it runs anywhere.

On older Pythons use await asyncio.wait_for(coro, 5), which wraps a single awaitable. Before 3.11 it raises asyncio.TimeoutError, which is a different class from the builtin TimeoutError. The two only became the same class in 3.11, so code that must run on older versions should catch asyncio.TimeoutError. The timeout() block is usually nicer because it covers several awaits with one shared budget.

asyncpg has its own timeouts, so you often do not need to wrap each call yourself. Pass command_timeout= to asyncpg.connect() or create_pool() to set a default for every command on those connections. Pass timeout= to a single fetch() or execute() call to override it. When the limit passes, asyncpg does not just abandon the call. It sends a Postgres cancel request, so the server stops working on the query too.

pool = await asyncpg.create_pool(dsn, command_timeout=10)

# per-call override for one query
rows = await pool.fetch(q, timeout=2)

Pool-wide default of 10 s, with a tighter 2 s for one call.

Client deadlines only help while your process is alive and healthy. Postgres can enforce limits itself, and they keep working if the client crashes or stalls. You set them as server settings when each connection opens, so every pooled connection carries them.

With asyncpg pass them as server_settings= to create_pool(). With SQLAlchemy put the same dictionary under connect_args={'server_settings': {...}} in create_async_engine().

A timeout, a closed client connection or a failed sibling task all end the same way: asyncio cancels the task. The task is not killed on the spot. Instead a CancelledError is raised at the await where it is currently suspended, and from there it unwinds like any exception, running finally blocks and context-manager exits on the way out.

This is what makes database code mostly safe by default. If the task is inside async with conn.transaction():, the exception leaves the block, so the transaction issues a ROLLBACK. The pool then either resets the connection or discards it if its state looks doubtful, so the next user gets a clean one.

The example below imitates a transaction with a tiny context manager so you can watch the order of events. Notice that the rollback happens before our own cleanup message, and that the cancel still reaches the caller.

Since Python 3.8, CancelledError inherits from BaseException, not Exception. A plain except Exception: lets it through, which is the intent. But except BaseException: pass or a bare except: catches it and discards it, and the task carries on as if nothing happened. The timeout block above then never fires correctly, and the task cannot be stopped at shutdown. The rule is simple: do your cleanup, then re-raise.

A database connection is a conversation: you send a command and then read its reply from the socket. If a cancel lands after the command was sent but before the reply was fully read, the connection is left half-read. The next command on it may read the leftover bytes of the previous answer. Careful drivers detect this and close the connection, or run a reset before reuse. Reusing such a connection yourself is dangerous.

This is not hypothetical. In March 2023 redis-py had exactly this class of bug, tracked as CVE-2023-28858 and CVE-2023-28859. A cancelled command on a pooled async connection returned the connection to the pool with an unread reply, and a later request, possibly from a different user, received that reply. The fix is in the driver, but the lesson is for you too: keep drivers updated, never push a connection back into a pool by hand after a cancel, and if you are unsure of its state, close it.

Sometimes a short step must finish even if the caller is cancelled, such as a final commit or releasing a connection. asyncio.shield() wraps the awaitable in its own task. If the outer task is cancelled, the inner task keeps running. There is a catch: the outer await still raises CancelledError straight away, so shield protects the work, not the caller. Use it for short cleanup only, never to make a whole job uncancellable.

A TaskGroup cancels all the remaining tasks as soon as one of them fails, then raises the errors together as an exception group. That is a good fit for database work, because a failed query should not leave its siblings running. It also means every sibling must be cancel-safe, and it means they must not share a connection or an AsyncSession. One connection serves one command at a time, and a cancel in one task could leave the shared connection half-read for the others. Give each task its own pool.acquire() or its own session.

Once a deadline fires, the temptation is to just try again. That is only safe when running the work twice has the same effect as running it once. A transaction that failed with a serialization error was rolled back completely, so nothing happened and a retry is fine. A timed-out INSERT is different: the client gave up, but the server may already have committed the row. A blind retry then creates a duplicate.

The loop below retries a fake serialization failure. It sleeps a little longer after each failure, which is called exponential backoff, so a crowd of clients does not hammer the database in lockstep. It reruns the whole unit of work each time, never a single statement from the middle of it. In real code add random jitter to the delay and stop after about three tries.

This page collects the calls you reach for most often. Code that needs a live database is shown as a fragment: the call site only, with pool, dsn, uid and the other names assumed to exist. The runnable example is on the last page.

Create one pool per process at startup. Borrow a connection with async with pool.acquire(), which hands it back on exit. Then call one of four methods. Values go in as positional $1, $2 placeholders and travel separately from the SQL text, so %s and ? do not work.

pool = await asyncpg.create_pool(dsn)

async with pool.acquire() as c:
    n = await c.fetchval('SELECT count(*) FROM users WHERE org = $1', org)
    row = await c.fetchrow('SELECT * FROM users WHERE id = $1', uid)
    await c.execute('UPDATE users SET seen = now() WHERE id = $1', uid)

Fragment: assumes dsn, org and uid exist.

A connection runs one query at a time. If two tasks share one connection and both await it, you get an InterfaceError. Give each task its own connection with pool.fetch() (which borrows a connection just for that call) or its own pool.acquire().

async with c.transaction() commits on a clean exit and rolls back when an exception escapes. Pass isolation='serializable' to pick the level. A transaction block opened inside another one becomes a savepoint, so only the inner block rolls back.

async with c.transaction(isolation='serializable'):
    await c.execute(debit, src, amount)
    async with c.transaction():          # SAVEPOINT
        await c.execute(audit, src, amount)

Fragment: debit and audit are SQL strings.

In transaction mode PgBouncer may send the next statement to a different server connection. A prepared statement that asyncpg cached on the first connection then does not exist on the second, and you see prepared statement "__asyncpg_stmt_1__" does not exist. Turn the statement cache off. Through SQLAlchemy, two keys are needed.

# plain asyncpg
conn = await asyncpg.connect(dsn, statement_cache_size=0)

# through SQLAlchemy: both keys
engine = create_async_engine(
    url,
    connect_args={'statement_cache_size': 0, 'prepared_statement_cache_size': 0},
)

Fragment: url is a postgresql+asyncpg URL.

Build the engine from a postgresql+asyncpg:// URL and wrap it in an async_sessionmaker with expire_on_commit=False, so loaded attributes stay readable after a commit instead of raising MissingGreenlet. Create both once at startup. For each request or task, open a new session with async with Session() as s, s.begin():, which commits on success and rolls back on error. Never share a session between concurrent tasks.

engine = create_async_engine('postgresql+asyncpg://user:pw@host/db', pool_pre_ping=True)
Session = async_sessionmaker(engine, expire_on_commit=False)

async with Session() as s, s.begin():
    res = await s.execute(select(User).where(User.age > 30))
    users = res.scalars().all()

Fragment: User is a mapped class.

An AsyncSession cannot run a hidden query when you read user.posts, because attribute access cannot be awaited. Decide how each relationship loads when you write the query. Set lazy='raise' on relationships so a surprise load fails loudly in tests instead of silently costing a query per row.

stmt = select(User).options(selectinload(User.posts))
users = (await s.scalars(stmt)).all()

posts = await u.awaitable_attrs.posts   # needs the AsyncAttrs mixin

Fragment: eager-load before leaving the session.

Every process holds its own pool, and a pool can grow past pool_size by max_overflow. The total across your deployment must stay under the server's limit, with headroom left for migrations and psql:

SQLite allows one writer at a time, so with aiosqlite run PRAGMA journal_mode=WAL and PRAGMA busy_timeout=5000 on each connection, so waiting writers wait instead of failing with 'database is locked'. For Redis, make one shared connection pool per app, batch commands with a pipeline, and close the client at shutdown with await r.aclose().

import redis.asyncio as redis

pool = redis.ConnectionPool.from_url(url, max_connections=50)
r = redis.Redis(connection_pool=pool)

async with r.pipeline(transaction=True) as p:
    p.incr('hits')
    p.expire('hits', 60)
    await p.execute()                  # one round trip

await r.aclose()                       # at shutdown
await pool.disconnect()

Fragment: url is a redis:// URL. A pool you pass in is not closed for you.

A client-side asyncio.timeout stops your code from waiting, and asyncpg then sends a cancel request so the server stops working too. A server-side statement_timeout protects you when the client is gone or buggy. Use both, and set the server value slightly lower than the client one so the server's clear error arrives first.

async with asyncio.timeout(5):                      # client layer
    rows = await conn.fetch(q)

engine = create_async_engine(
    url,
    connect_args={'server_settings': {'statement_timeout': '4s', 'lock_timeout': '2s'}},
)                                                   # server layer

Fragment: server settings apply to every new pooled connection.

While a transaction is open it holds row locks and one pooled connection. If you await an HTTP call inside it, every other task that needs that row or connection waits for a network you do not control. Make the call first, then open the transaction and write the results. A leaked transaction shows up as idle in transaction in pg_stat_activity, and idle_in_transaction_session_timeout makes the server close it.

When a task is cancelled or a timeout fires, CancelledError is raised at the current await. It is a BaseException, so except Exception lets it pass, but except BaseException: pass buries it and the task refuses to stop. Do your cleanup, then re-raise. The example below runs without any database: a fake transaction fails twice with a serialization error, the whole thing is retried with backoff, and then a slow call is cut off by a deadline.

Notice that the retry wraps the whole unit of work, not a single statement, because a failed serializable transaction has already been rolled back. Retry only work that is safe to repeat: a serialization failure or a lost connection qualifies, but an INSERT that timed out may have committed, so use an idempotency key before repeating it.

Each question shows a short situation. Decide what happens before you open the answer.

• Async drivers (asyncpg, aiosqlite, redis.asyncio) let the loop serve other tasks while the database works. They add concurrency, not speed, and a blocking call in async def freezes everything.
• One asyncpg connection runs one query at a time. Share a pool, not a connection, and use $1 placeholders so values never touch the SQL text.
• Behind PgBouncer in transaction mode, turn the statement caches off (statement_cache_size=0, plus prepared_statement_cache_size=0 through SQLAlchemy).
• Use one AsyncSession per request or task, with expire_on_commit=False. Eager-load with selectinload or joinedload, and treat MissingGreenlet as a missed load.
• Keep transactions short and free of external I/O. Retry the whole transaction on SerializationFailure (40001).
• Size pools with instances x workers x (pool_size + max_overflow) below max_connections, and use Little's law instead of guessing big.
• Set deadlines on both sides (asyncio.timeout and statement_timeout), never swallow CancelledError, and retry only idempotent work.