psycopg
Psycopg 3 is the current PostgreSQL adapter for Python, a full rewrite of the psycopg2 everyone has been importing for fifteen years. It is a DB API 2.0 driver: connect, get a cursor, execute SQL with %s placeholders, fetch rows. What changed in version 3 is the machinery underneath. Parameters are sent to the server separately from the query text using the extended query protocol instead of being pasted into a string client-side, asyncio is a first-class API rather than a bolt-on, COPY is exposed as a normal Python context manager, and connection pooling moved into a separate psycopg-pool package with its own release cycle. The PyPI project named psycopg is version 3; psycopg2 still lives at its own name.
The default choice for new PostgreSQL work in Python: actively developed, sync and async in one package, and a much cleaner COPY and pooling story than psycopg2. Install it with the extras, because the bare package is the slow path that also needs a system libpq.
Use it if
- You are starting a new project against PostgreSQL and want the driver that is actually being developed rather than the one in maintenance mode
- You need both sync and async against the same database: AsyncConnection mirrors Connection method for method, so one codebase can do both
- You bulk load data and want COPY as a first-class API (cursor.copy() with write_row) instead of psycopg2 copy_expert string wrangling
- You want server-side parameter binding by default, which keeps query text and data separate on the wire and lets Postgres cache plans
- You want row factories, so a cursor can hand you dicts, namedtuples or your own dataclass instead of positional tuples
- Your team is deep in psycopg2 and the migration is not funded. The porting list is real: "with connection" now closes the connection at block exit instead of only committing, copy_from and copy_expert are gone, the pool moved to psycopg-pool, and server-side binding rejects patterns that used to work.
- You parameterize things that are not data. Because binding happens server-side, cur.execute("SET search_path = %s", [x]) and multiple semicolon-separated statements in one execute fail. You have to reach for ClientCursor or psycopg.sql composition, and finding that out costs an afternoon.
- You need maximum asyncio throughput. asyncpg implements the PostgreSQL wire protocol directly in Cython rather than going through libpq, and it stays measurably ahead on raw query rate.
- You are on PyPy, or on macOS before 14.0 with Apple silicon. Binary packages are not published for either, so you are back to installing libpq and build tools yourself.
- Copyleft is a problem for you. Psycopg is LGPL-3.0. Dynamic linking keeps most companies comfortable, but if your legal review bans the licence family outright, settle that before writing code.
- You wanted an ORM or a query builder. This is a driver. You still write SQL strings, and nothing here manages migrations, relationships or a schema.
Setup reality
The install command that works is pip install "psycopg[binary,pool]", and the trap is that everyone types pip install psycopg first. Bare psycopg is the pure Python implementation: the docs describe it as much slower, and it still needs the libpq client library present on the machine because it loads it dynamically through ctypes, so a slim container gives you an import error nobody expects from a pip install that succeeded. The [binary] extra bundles its own libpq and is what the docs recommend for most users, but it is not published for PyPy or for pre-14.0 macOS on ARM64, and the libpq version baked into the wheel is whatever the build runners had, which you can check with psycopg.pq.version(). The [c] extra builds against your system libpq and needs build tools plus headers. Pooling is a separate package behind [pool], and since psycopg_pool 3.2 a sync pool that relies on the implicit open warns you that the default will flip to False, while an async pool opened in its constructor raises a RuntimeWarning. Use the pool as a context manager and both problems disappear. Python 3.10 or newer is required.
Patterns
Connect, query, and let the block commitconnect-and-query
import psycopg
with psycopg.connect("postgresql://user:pw@localhost/app") as conn:
with conn.cursor() as cur:
cur.execute("INSERT INTO items (name, qty) VALUES (%s, %s)", ("widget", 3))
cur.execute("SELECT id, name FROM items WHERE qty > %s", (1,))
print(cur.fetchall())Unlike psycopg2, leaving the connection block closes the connection as well as committing it, so do not reuse conn afterwards. An exception inside the block rolls back instead.
Get dicts instead of tuplesdict-rows
from psycopg.rows import dict_row, class_row
from dataclasses import dataclass
with conn.cursor(row_factory=dict_row) as cur:
cur.execute("SELECT id, name FROM items")
print(cur.fetchone()) # {"id": 1, "name": "widget"}
@dataclass
class Item:
id: int
name: str
with conn.cursor(row_factory=class_row(Item)) as cur:
cur.execute("SELECT id, name FROM items")
item = cur.fetchone() # Item(id=1, name="widget")row_factory can be set on the connection instead of per cursor so every cursor inherits it. class_row matches by column name, so your SELECT column list has to line up with the field names.
Pass parameters, never format themparameters-safely
cur.execute("SELECT * FROM items WHERE name = %s AND qty > %s", ("widget", 2))
cur.execute(
"SELECT * FROM items WHERE name = %(name)s",
{"name": "widget"},
)
# IN lists work with a single placeholder and a sequence
cur.execute("SELECT * FROM items WHERE id = ANY(%s)", ([1, 2, 3],))Placeholders are always %s or %(name)s, never ? and never an f-string. A single parameter still needs a one-element tuple, and the classic bug is writing ("widget") which is just a string.
Interpolate a table or column name safelydynamic-identifiers
from psycopg import sql
query = sql.SQL("SELECT {fields} FROM {table} WHERE id = %s").format(
fields=sql.SQL(", ").join(map(sql.Identifier, ["id", "name"])),
table=sql.Identifier("public", "items"),
)
cur.execute(query, (1,))Parameters can only ever be values, so a variable table or column name has to go through sql.Identifier, which quotes and escapes it. String concatenation here is the classic injection hole.
Fall back to client-side bindingclient-side-binding
from psycopg import ClientCursor
with psycopg.connect(dsn, cursor_factory=ClientCursor) as conn:
with conn.cursor() as cur:
cur.execute("SET statement_timeout = %s", ("5s",))
print(cur.mogrify("SELECT %s", ("x",)))Server-side binding cannot parameterize non-data positions such as SET, LISTEN channel names, or several statements in one execute. ClientCursor restores the psycopg2 behaviour of building the query string in Python, which is also the only way to get mogrify back.
Control transactions explicitlytransaction-block
conn.autocommit = True # each statement commits on its own
with conn.transaction():
cur.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
with conn.transaction(): # nested, becomes a SAVEPOINT
cur.execute("INSERT INTO audit (msg) VALUES (%s)", ("transfer",))
cur.execute("UPDATE accounts SET balance = balance + 100 WHERE id = 2")conn.transaction() commits at the end of the block and rolls back on an exception. Nested blocks map onto SAVEPOINTs, so an inner failure can be caught without discarding the outer transaction.
Bulk load rows with COPYcopy-bulk-load
records = [(i, f"item-{i}") for i in range(100_000)]
with cur.copy("COPY items (id, name) FROM STDIN") as copy:
for record in records:
copy.write_row(record)
# and back out again
with cur.copy("COPY items TO STDOUT") as copy:
for row in copy.rows():
print(row)COPY is an order of magnitude faster than looped INSERTs and is the reason to pick this driver for loaders. copy_from and copy_expert from psycopg2 no longer exist; this context manager replaced both.
Stream a result set too large for memoryserver-side-cursor
with conn.cursor(name="big_scan") as cur:
cur.itersize = 10_000
cur.execute("SELECT * FROM events")
for row in cur:
handle(row)Naming the cursor makes it a server-side cursor, so rows arrive in batches instead of the whole result landing in the client. It has to live inside a transaction, and you cannot reuse the name while it is open.
Pool connectionsconnection-pool
from psycopg_pool import ConnectionPool
with ConnectionPool(dsn, min_size=2, max_size=10) as pool:
with pool.connection() as conn:
conn.execute("SELECT 1")Pooling is the separate psycopg-pool package, installed via psycopg[pool]. Use the pool as a context manager: since 3.2 a sync pool relying on the implicit open warns that the default will become False, and an async pool opened in its constructor raises a RuntimeWarning.
Query from asyncioasync-connection
import asyncio
import psycopg
async def main():
async with await psycopg.AsyncConnection.connect(dsn) as aconn:
async with aconn.cursor() as acur:
await acur.execute("SELECT id, name FROM items WHERE qty > %s", (1,))
async for row in acur:
print(row)
asyncio.run(main())Note the await inside async with: connect() is a coroutine that returns the context manager. Method names match the sync API exactly, so porting is mostly adding await.
Insert many rows and read what came backexecutemany-returning
cur.executemany(
"INSERT INTO items (name) VALUES (%s) RETURNING id",
[("a",), ("b",), ("c",)],
returning=True,
)
while True:
print(cur.fetchone())
if not cur.nextset():
breakreturning=True is opt-in because keeping every result set costs memory. executemany batches the round trips but is still slower than COPY for large loads.
Listen for Postgres notificationslisten-notify
conn = psycopg.connect(dsn, autocommit=True)
conn.execute("LISTEN job_queue")
for notify in conn.notifies():
print(notify.channel, notify.payload)
if notify.payload == "stop":
breakautocommit must be on, otherwise LISTEN sits inside an open transaction and you receive nothing. notifies() blocks the connection, so give it a connection of its own rather than borrowing one from the pool.
Alternatives
| Package | Registry | Pick it when |
|---|---|---|
| psycopg2-binary | PyPI | An existing codebase already targets psycopg2 and you only need it to keep running |
| asyncpg | PyPI | Pure asyncio service where raw query throughput matters more than DB API compatibility |
| sqlalchemy | PyPI | You want a Core query builder or an ORM; it drives psycopg 3 as one of its dialects |