Python psycopg async usage

When you need to run PostgreSQL queries from an asyncio application, psycopg3 provides a native async API. The psycopg.AsyncConnection and psycopg.AsyncCursor classes let you write non-blocking database code without wrapping synchronous calls in threads. This article covers the practical aspects of python psycopg async usage: setting up connections, executing queries, managing transactions, handling errors, and understanding the concurrency model.

What psycopg async offers

psycopg3 (the current major version) ships with an async implementation that integrates directly with asyncio. Unlike psycopg2, which is synchronous and requires you to manage threads or use run_in_executor for concurrency, psycopg3 exposes AsyncConnection and AsyncCursor. These classes mirror the synchronous API but use await for every operation that touches the network. The result is that your event loop stays free to handle other tasks while the database query is in flight.

The async API is not a separate library; it is part of the same psycopg package. You import AsyncConnection from psycopg and use it with await expressions. This makes it straightforward to convert existing synchronous code once you understand the differences.

Setting up an async connection

To create an async connection, use await psycopg.AsyncConnection.connect(). The connection parameters are the same as for the synchronous version: host, port, database, user, and password. Here is a minimal example:

import asyncio
import psycopg

async def main():
    conn = await psycopg.AsyncConnection.connect(
        host="localhost",
        port=5432,
        dbname="mydb",
        user="myuser",
        password="secret"
    )
    print("connected")
    await conn.close()

asyncio.run(main())

The connect method is a coroutine, so you must await it. Once connected, you can use the connection to create cursors and execute queries. The connection object is also an async context manager, which ensures the connection is closed even if an exception occurs:

async with await psycopg.AsyncConnection.connect(...) as conn:
    # use the connection
    pass

Note that connect itself is a coroutine, so you need await before it even inside an async with. This pattern is common and avoids forgetting to close the connection.

Executing queries with AsyncCursor

To run a query, create a cursor using await conn.cursor(). The cursor is also an async object. You can execute a statement and then fetch results using await cursor.fetchall(), await cursor.fetchone(), or iterate over the cursor asynchronously.

async with await psycopg.AsyncConnection.connect(...) as conn:
    async with conn.cursor() as cur:
        await cur.execute("SELECT id, name FROM users WHERE active = %s", (True,))
        async for row in cur:
            print(row)

Notice that conn.cursor() is a coroutine, so you need await before it. The cursor also supports async with to ensure it is closed. The execute method is awaited, and fetching rows is also asynchronous. The %s placeholder is used for parameters, just like in the synchronous API.

If you need to execute a query that returns no rows, such as an INSERT or UPDATE, you still await cur.execute(). The cursor will not have any rows to fetch, but you can check cur.rowcount to see how many rows were affected.

Managing transactions and commits

By default, psycopg does not autocommit. You must explicitly commit or roll back a transaction. In async code, this is done with await conn.commit() or await conn.rollback(). The connection's transaction state is the same as in synchronous psycopg: the first execute starts a transaction, and you commit when you are ready.

async with await psycopg.AsyncConnection.connect(...) as conn:
    async with conn.cursor() as cur:
        await cur.execute("INSERT INTO logs (message) VALUES (%s)", ("hello",))
    await conn.commit()

If an exception occurs before the commit, the transaction remains open. You can use try/except to roll back and re-raise the error. Alternatively, you can use the connection as a context manager that automatically commits on success and rolls back on failure. The async connection supports this pattern:

async with await psycopg.AsyncConnection.connect(...) as conn:
    async with conn.transaction():
        # operations are committed if no exception
        pass

The transaction() method returns an async context manager that handles commit and rollback for you. This is a cleaner way to manage transactions when you have multiple statements that should be atomic.

Handling errors in async code

Errors in async psycopg behave like any other exception in Python. The main difference is that they can be raised at await points. You should wrap your database operations in try/except blocks to catch psycopg.Error or more specific subclasses like psycopg.OperationalError or psycopg.DataError.

try:
    await cur.execute("SELECT * FROM missing_table")
except psycopg.errors.UndefinedTable:
    print("Table does not exist")

Because the exception is raised inside the coroutine, it propagates to the caller. If you are using asyncio.gather or other concurrency primitives, the exception will be raised when you await the task. This means you need to handle exceptions at the point where you await the database operation, not inside a separate callback.

Another common issue is that a connection is left in a broken state after an error. If a transaction fails, you may need to roll back before you can reuse the connection. The async with conn.transaction() pattern automatically rolls back on exception, so it is safer for long-running applications.

Performance and concurrency considerations

The primary benefit of async psycopg is that the event loop is not blocked while waiting for the database. This allows other tasks to proceed during I/O. However, each connection can only be used by one coroutine at a time. If you have multiple concurrent tasks that need to access the database, you must use a connection pool.

psycopg provides a connection pool in the psycopg_pool package, which supports async connections. You can create an AsyncConnectionPool and acquire connections from it. This prevents you from creating a new connection for every request, which would be expensive. The pool manages the lifecycle of connections and ensures that each connection is used by only one task at a time.

from psycopg_pool import AsyncConnectionPool

pool = AsyncConnectionPool("host=localhost dbname=mydb user=myuser password=secret", open=False)
await pool.open()

async with pool.connection() as conn:
    async with conn.cursor() as cur:
        await cur.execute("SELECT 1")

Using a pool is essential for production applications that handle many concurrent requests. Without a pool, you risk exhausting the database's connection limit or creating too many connections that each consume memory and file descriptors.

Common pitfalls and limitations

One of the most frequent mistakes is forgetting to await a coroutine. For example, calling conn.cursor() without await returns a coroutine object, not a cursor. This leads to confusing errors when you try to call methods on it. Always double-check that you have await before any operation that returns a coroutine.

Another pitfall is mixing synchronous and asynchronous code. If you call a synchronous psycopg function from an async context, it will block the event loop. Avoid using the synchronous psycopg.connect in an async application. Stick to the async API throughout.

A limitation of the async API is that you cannot use a single connection from multiple tasks simultaneously. If you try to execute two queries on the same connection without waiting for the first to finish, you will get an error. This is why a connection pool is necessary for concurrency.

Finally, note that psycopg's async support requires Python 3.8 or later, and you need to be running inside an asyncio event loop. The asyncio.run() function provides a simple way to start the loop for a script, but in a larger application you will likely integrate psycopg with your existing asyncio framework.

When to choose async over sync

The decision to use async psycopg depends on your application's concurrency model. If you are building an asyncio-based web server, API gateway, or real-time service, the async API is the natural fit. It allows you to handle many database operations concurrently without tying up threads.

If your application is primarily synchronous, using the sync API with a thread pool might be simpler. The async API adds complexity because you must manage the event loop and ensure that all database calls are awaited. However, if you are already using asyncio for other I/O, adding async psycopg avoids the overhead of switching between threads and the event loop.

For most new asyncio projects, the async API is the right choice. It keeps your code consistent and avoids the performance penalty of blocking the event loop. Just remember to use a connection pool and handle transactions explicitly to get the most out of it.