Psycopg - PostgreSQL database adapter for Python - Psycopg documentation

Supported systems #

The Psycopg version documented here has official and tested support for:

• Python: from version 3.7 to 3.11

  • Python 3.6 supported before Psycopg 3.1

• PostgreSQL: from version 10 to 15

• OS: Linux, macOS, Windows

The tests to verify the supported systems run in Github workflows : anything that is not tested there is not officially supported. This includes:

• Unofficial Python distributions such as Conda;

• Alternative PostgreSQL implementation;

• macOS hardware and releases not available on Github workflows.

If you use an unsupported system, things might work (because, for instance, the database may use the same wire protocol as PostgreSQL) but we cannot guarantee the correct working or a smooth ride.

Binary installation #

The quickest way to start developing with Psycopg 3 is to install the binary packages by running:

pip install "psycopg[binary]"

This will install a self-contained package with all the libraries needed. You will need pip 20.3 at least : please run pip install --upgrade pip to update it beforehand.

The above package should work in most situations. It will not work in some cases though.

If your platform is not supported you should proceed to a local installation or a pure Python installation .

Local installation #

A "Local installation" results in a performing and maintainable library. The library will include the speed-up C module and will be linked to the system libraries ( libpq , libssl …) so that system upgrade of libraries will upgrade the libraries used by Psycopg 3 too. This is the preferred way to install Psycopg for a production site.

In order to perform a local installation you need some prerequisites:

• a C compiler,

• Python development headers (e.g. the python3-dev package).

• PostgreSQL client development headers (e.g. the libpq-dev package).

• The pg_config program available in the PATH .

You must be able to troubleshoot an extension build, for instance you must be able to read your compiler’s error message. If you are not, please don’t try this and follow the binary installation instead.

If your build prerequisites are in place you can run:

pip install "psycopg[c]"

Pure Python installation #

If you simply install:

pip install psycopg

without [c] or [binary] extras you will obtain a pure Python implementation. This is particularly handy to debug and hack, but it still requires the system libpq to operate (which will be imported dynamically via ctypes ).

In order to use the pure Python installation you will need the libpq installed in the system: for instance on Debian system you will probably need:

sudo apt install libpq5

If you are not able to fulfill this requirement please follow the binary installation .

Installing the connection pool #

The Psycopg connection pools are distributed in a separate package from the psycopg package itself, in order to allow a different release cycle.

In order to use the pool you must install the pool extra, using pip install "psycopg[pool]" , or install the psycopg_pool package separately, which would allow to specify the release to install more precisely.

Handling dependencies #

If you need to specify your project dependencies (for instance in a requirements.txt file, setup.py , pyproject.toml dependencies…) you should probably specify one of the following:

• If your project is a library, add a dependency on psycopg . This will make sure that your library will have the psycopg package with the right interface and leaves the possibility of choosing a specific implementation to the end user of your library.

• If your project is a final application (e.g. a service running on a server) you can require a specific implementation, for instance psycopg[c] , after you have made sure that the prerequisites are met (e.g. the depending libraries and tools are installed in the host machine).

In both cases you can specify which version of Psycopg to use using requirement specifiers .

If you want to make sure that a specific implementation is used you can specify the PSYCOPG_IMPL environment variable: importing the library will fail if the implementation specified is not available. See pq module implementations .

Main objects in Psycopg 3 #

Here is an interactive session showing some of the basic commands:

# Note: the module name is psycopg, not psycopg3
import psycopg

# Connect to an existing database
with psycopg.connect("dbname=test user=postgres") as conn:

    # Open a cursor to perform database operations
    with conn.cursor() as cur:

        # Execute a command: this creates a new table
        cur.execute("""
            CREATE TABLE test (
                id serial PRIMARY KEY,
                num integer,
                data text)
            """)

        # Pass data to fill a query placeholders and let Psycopg perform
        # the correct conversion (no SQL injections!)
        cur.execute(
            "INSERT INTO test (num, data) VALUES (%s, %s)",
            (100, "abc'def"))

        # Query the database and obtain data as Python objects.
        cur.execute("SELECT * FROM test")
        cur.fetchone()
        # will return (1, 100, "abc'def")

        # You can use `cur.fetchmany()`, `cur.fetchall()` to return a list
        # of several records, or even iterate on the cursor
        for record in cur:
            print(record)

        # Make the changes to the database persistent
        conn.commit()

In the example you can see some of the main objects and methods and how they relate to each other:

• The function connect() creates a new database session and returns a new Connection instance. AsyncConnection.connect() creates an asyncio connection instead.

• The Connection class encapsulates a database session. It allows to:

  • create new Cursor instances using the cursor() method to execute database commands and queries,

  • terminate transactions using the methods commit() or rollback() .

• The class Cursor allows interaction with the database:

  • send commands to the database using methods such as execute() and executemany() ,

  • retrieve data from the database, iterating on the cursor or using methods such as fetchone() , fetchmany() , fetchall() .

• Using these objects as context managers (i.e. using with ) will make sure to close them and free their resources at the end of the block (notice that this is different from psycopg2 ).

Shortcuts #

The pattern above is familiar to psycopg2 users. However, Psycopg 3 also exposes a few simple extensions which make the above pattern leaner:

• the Connection objects exposes an execute() method, equivalent to creating a cursor, calling its execute() method, and returning it.

  # In Psycopg 2
  cur = conn.cursor()
  cur.execute(...)
  
  # In Psycopg 3
  cur = conn.execute(...)
  

• The Cursor.execute() method returns self . This means that you can chain a fetch operation, such as fetchone() , to the execute() call:

  # In Psycopg 2
  cur.execute(...)
  record = cur.fetchone()
  
  cur.execute(...)
  for record in cur:
      ...
  
  # In Psycopg 3
  record = cur.execute(...).fetchone()
  
  for record in cur.execute(...):
      ...
  

Using them together, in simple cases, you can go from creating a connection to using a result in a single expression:

print(psycopg.connect(DSN).execute("SELECT now()").fetchone()[0])
# 2042-07-12 18:15:10.706497+01:00

Connection context #

Psycopg 3 Connection can be used as a context manager:

with psycopg.connect() as conn:
    ... # use the connection

# the connection is now closed

When the block is exited, if there is a transaction open, it will be committed. If an exception is raised within the block the transaction is rolled back. In both cases the connection is closed. It is roughly the equivalent of:

conn = psycopg.connect()
try:
    ... # use the connection
except BaseException:
    conn.rollback()
else:
    conn.commit()
finally:
    conn.close()

Note that, while the above pattern is what most people would use, connect() doesn’t enter a block itself, but returns an "un-entered" connection, so that it is still possible to use a connection regardless of the code scope and the developer is free to use (and responsible for calling) commit() , rollback() , close() as and where needed.

AsyncConnection can be also used as context manager, using async with , but be careful about its quirkiness: see with async connections for details.

Adapting pyscopg to your program #

The above pattern of use only shows the default behaviour of the adapter. Psycopg can be customised in several ways, to allow the smoothest integration between your Python program and your PostgreSQL database:

• If your program is concurrent and based on asyncio instead of on threads/processes, you can use async connections and cursors .

• If you want to customise the objects that the cursor returns, instead of receiving tuples, you can specify your row factories .

• If you want to customise how Python values and PostgreSQL types are mapped into each other, beside the basic type mapping , you can configure your types .

execute() arguments #

Passing parameters to a SQL statement happens in functions such as Cursor.execute() by using %s placeholders in the SQL statement, and passing a sequence of values as the second argument of the function. For example the Python function call:

cur.execute("""
    INSERT INTO some_table (id, created_at, last_name)
    VALUES (%s, %s, %s);
    """,
    (10, datetime.date(2020, 11, 18), "O'Reilly"))

is roughly equivalent to the SQL command:

INSERT INTO some_table (id, created_at, last_name)
VALUES (10, '2020-11-18', 'O''Reilly');

Note that the parameters will not be really merged to the query: query and the parameters are sent to the server separately: see Server-side binding for details.

Named arguments are supported too using %( name )s placeholders in the query and specifying the values into a mapping. Using named arguments allows to specify the values in any order and to repeat the same value in several places in the query:

cur.execute("""
    INSERT INTO some_table (id, created_at, updated_at, last_name)
    VALUES (%(id)s, %(created)s, %(created)s, %(name)s);
    """,
    {'id': 10, 'name': "O'Reilly", 'created': datetime.date(2020, 11, 18)})

Using characters % , ( , ) in the argument names is not supported.

When parameters are used, in order to include a literal % in the query you can use the %% string:

cur.execute("SELECT (%s % 2) = 0 AS even", (10,))       # WRONG
cur.execute("SELECT (%s %% 2) = 0 AS even", (10,))      # correct

While the mechanism resembles regular Python strings manipulation, there are a few subtle differences you should care about when passing parameters to a query.

• The Python string operator % must not be used : the execute() method accepts a tuple or dictionary of values as second parameter. Never use % or + to merge values into queries :

  cur.execute("INSERT INTO numbers VALUES (%s, %s)" % (10, 20)) # WRONG
  cur.execute("INSERT INTO numbers VALUES (%s, %s)", (10, 20))  # correct
  

• For positional variables binding, the second argument must always be a sequence , even if it contains a single variable (remember that Python requires a comma to create a single element tuple):

  cur.execute("INSERT INTO foo VALUES (%s)", "bar")    # WRONG
  cur.execute("INSERT INTO foo VALUES (%s)", ("bar"))  # WRONG
  cur.execute("INSERT INTO foo VALUES (%s)", ("bar",)) # correct
  cur.execute("INSERT INTO foo VALUES (%s)", ["bar"])  # correct
  

• The placeholder must not be quoted :

  cur.execute("INSERT INTO numbers VALUES ('%s')", ("Hello",)) # WRONG
  cur.execute("INSERT INTO numbers VALUES (%s)", ("Hello",))   # correct
  

• The variables placeholder must always be a %s , even if a different placeholder (such as a %d for integers or %f for floats) may look more appropriate for the type. You may find other placeholders used in Psycopg queries ( %b and %t ) but they are not related to the type of the argument: see Binary parameters and results if you want to read more:

  cur.execute("INSERT INTO numbers VALUES (%d)", (10,))   # WRONG
  cur.execute("INSERT INTO numbers VALUES (%s)", (10,))   # correct
  

• Only query values should be bound via this method: it shouldn’t be used to merge table or field names to the query. If you need to generate SQL queries dynamically (for instance choosing a table name at runtime) you can use the functionalities provided in the psycopg.sql module:

  cur.execute("INSERT INTO %s VALUES (%s)", ('numbers', 10))  # WRONG
  cur.execute(                                                # correct
      SQL("INSERT INTO {} VALUES (%s)").format(Identifier('numbers')),
      (10,))
  

Danger: SQL injection #

The SQL representation of many data types is often different from their Python string representation. The typical example is with single quotes in strings: in SQL single quotes are used as string literal delimiters, so the ones appearing inside the string itself must be escaped, whereas in Python single quotes can be left unescaped if the string is delimited by double quotes.

Because of the difference, sometimes subtle, between the data types representations, a naïve approach to query strings composition, such as using Python strings concatenation, is a recipe for terrible problems:

SQL = "INSERT INTO authors (name) VALUES ('%s')" # NEVER DO THIS
data = ("O'Reilly", )
cur.execute(SQL % data) # THIS WILL FAIL MISERABLY
# SyntaxError: syntax error at or near "Reilly"

If the variables containing the data to send to the database come from an untrusted source (such as data coming from a form on a web site) an attacker could easily craft a malformed string, either gaining access to unauthorized data or performing destructive operations on the database. This form of attack is called SQL injection and is known to be one of the most widespread forms of attack on database systems. Before continuing, please print this page as a memo and hang it onto your desk.

Psycopg can automatically convert Python objects to SQL values : using this feature your code will be more robust and reliable. We must stress this point:

The correct way to pass variables in a SQL command is using the second argument of the Cursor.execute() method:

SQL = "INSERT INTO authors (name) VALUES (%s)"  # Note: no quotes
data = ("O'Reilly", )
cur.execute(SQL, data)  # Note: no % operator

Binary parameters and results #

PostgreSQL has two different ways to transmit data between client and server: TEXT , always available, and BINARY , available most of the times but not always. Usually the binary format is more efficient to use.

Psycopg can support both formats for each data type. Whenever a value is passed to a query using the normal %s placeholder, the best format available is chosen (often, but not always, the binary format is picked as the best choice).

If you have a reason to select explicitly the binary format or the text format for a value you can use respectively a %b placeholder or a %t placeholder instead of the normal %s . execute() will fail if a Dumper for the right data type and format is not available.

The same two formats, text or binary, are used by PostgreSQL to return data from a query to the client. Unlike with parameters, where you can choose the format value-by-value, all the columns returned by a query will have the same format. Every type returned by the query should have a Loader configured, otherwise the data will be returned as unparsed str (for text results) or buffer (for binary results).

Because not every PostgreSQL type supports binary output, by default, the data will be returned in text format. In order to return data in binary format you can create the cursor using Connection.cursor (binary=True) or execute the query using Cursor.execute (binary=True) . A case in which requesting binary results is a clear winner is when you have large binary data in the database, such as images:

cur.execute(
    "SELECT image_data FROM images WHERE id = %s", [image_id], binary=True)
data = cur.fetchone()[0]

Booleans adaptation #

Python bool values True and False are converted to the equivalent PostgreSQL boolean type :

>>> cur.execute("SELECT %s, %s", (True, False))
# equivalent to "SELECT true, false"

Numbers adaptation #

• Python int values can be converted to PostgreSQL smallint , integer , bigint , or numeric , according to their numeric value. Psycopg will choose the smallest data type available, because PostgreSQL can automatically cast a type up (e.g. passing a smallint where PostgreSQL expect an integer is gladly accepted) but will not cast down automatically (e.g. if a function has an integer argument, passing it a bigint value will fail, even if the value is 1).

• Python float values are converted to PostgreSQL float8 .

• Python Decimal values are converted to PostgreSQL numeric .

On the way back, smaller types ( int2 , int4 , float4 ) are promoted to the larger Python counterpart.

Strings adaptation #

Python str are converted to PostgreSQL string syntax, and PostgreSQL types such as text and varchar are converted back to Python str :

conn = psycopg.connect()
conn.execute(
    "INSERT INTO menu (id, entry) VALUES (%s, %s)",
    (1, "Crème Brûlée at 4.99€"))
conn.execute("SELECT entry FROM menu WHERE id = 1").fetchone()[0]
'Crème Brûlée at 4.99€'

PostgreSQL databases have an encoding , and the session has an encoding too, exposed in the Connection.info. encoding attribute. If your database and connection are in UTF-8 encoding you will likely have no problem, otherwise you will have to make sure that your application only deals with the non-ASCII chars that the database can handle; failing to do so may result in encoding/decoding errors:

# The encoding is set at connection time according to the db configuration
conn.info.encoding
'utf-8'

# The Latin-9 encoding can manage some European accented letters
# and the Euro symbol
conn.execute("SET client_encoding TO LATIN9")
conn.execute("SELECT entry FROM menu WHERE id = 1").fetchone()[0]
'Crème Brûlée at 4.99€'

# The Latin-1 encoding doesn't have a representation for the Euro symbol
conn.execute("SET client_encoding TO LATIN1")
conn.execute("SELECT entry FROM menu WHERE id = 1").fetchone()[0]
# Traceback (most recent call last)
# ...
# UntranslatableCharacter: character with byte sequence 0xe2 0x82 0xac
# in encoding "UTF8" has no equivalent in encoding "LATIN1"

In rare cases you may have strings with unexpected encodings in the database. Using the SQL_ASCII client encoding will disable decoding of the data coming from the database, which will be returned as bytes :

conn.execute("SET client_encoding TO SQL_ASCII")
conn.execute("SELECT entry FROM menu WHERE id = 1").fetchone()[0]
b'Cr\xc3\xa8me Br\xc3\xbbl\xc3\xa9e at 4.99\xe2\x82\xac'

Alternatively you can cast the unknown encoding data to bytea to retrieve it as bytes, leaving other strings unaltered: see Binary adaptation

Note that PostgreSQL text cannot contain the 0x00 byte. If you need to store Python strings that may contain binary zeros you should use a bytea field.

Binary adaptation #

Python types representing binary objects ( bytes , bytearray , memoryview ) are converted by default to bytea fields. By default data received is returned as bytes .

If you are storing large binary data in bytea fields (such as binary documents or images) you should probably use the binary format to pass and return values, otherwise binary data will undergo ASCII escaping , taking some CPU time and more bandwidth. See Binary parameters and results for details.

Date/time types adaptation #

• Python date objects are converted to PostgreSQL date .

• Python datetime objects are converted to PostgreSQL timestamp (if they don’t have a tzinfo set) or timestamptz (if they do).

• Python time objects are converted to PostgreSQL time (if they don’t have a tzinfo set) or timetz (if they do).

• Python timedelta objects are converted to PostgreSQL interval .

PostgreSQL timestamptz values are returned with a timezone set to the connection TimeZone setting , which is available as a Python ZoneInfo object in the Connection.info . timezone attribute:

>>> conn.info.timezone
zoneinfo.ZoneInfo(key='Europe/London')

>>> conn.execute("select '2048-07-08 12:00'::timestamptz").fetchone()[0]
datetime.datetime(2048, 7, 8, 12, 0, tzinfo=zoneinfo.ZoneInfo(key='Europe/London'))

>>> conn.execute("SET TIMEZONE to 'Europe/Rome'")  # UTC+2 in summer

>>> conn.execute("SELECT '2042-07-01 12:00Z'::timestamptz").fetchone()[0]  # UTC input
datetime.datetime(2042, 7, 1, 14, 0, tzinfo=zoneinfo.ZoneInfo(key='Europe/Rome'))

JSON adaptation #

Psycopg can map between Python objects and PostgreSQL json/jsonb types , allowing to customise the load and dump function used.

Because several Python objects could be considered JSON (dicts, lists, scalars, even date/time if using a dumps function customised to use them), Psycopg requires you to wrap the object to dump as JSON into a wrapper: either psycopg.types.json.Json or Jsonb .

from psycopg.types.json import Jsonb

thing = {"foo": ["bar", 42]}
conn.execute("INSERT INTO mytable VALUES (%s)", [Jsonb(thing)])

By default Psycopg uses the standard library json.dumps and json.loads functions to serialize and de-serialize Python objects to JSON. If you want to customise how serialization happens, for instance changing serialization parameters or using a different JSON library, you can specify your own functions using the psycopg.types.json.set_json_dumps() and set_json_loads() functions, to apply either globally or to a specific context (connection or cursor).

from functools import partial
from psycopg.types.json import Jsonb, set_json_dumps, set_json_loads
import ujson

# Use a faster dump function
set_json_dumps(ujson.dumps)

# Return floating point values as Decimal, just in one connection
set_json_loads(partial(json.loads, parse_float=Decimal), conn)

conn.execute("SELECT %s", [Jsonb({"value": 123.45})]).fetchone()[0]
# {'value': Decimal('123.45')}

If you need an even more specific dump customisation only for certain objects (including different configurations in the same query) you can specify a dumps parameter in the Json / Jsonb wrapper, which will take precedence over what is specified by set_json_dumps() .

from uuid import UUID, uuid4

class UUIDEncoder(json.JSONEncoder):
    """A JSON encoder which can dump UUID."""
    def default(self, obj):
        if isinstance(obj, UUID):
            return str(obj)
        return json.JSONEncoder.default(self, obj)

uuid_dumps = partial(json.dumps, cls=UUIDEncoder)
obj = {"uuid": uuid4()}
cnn.execute("INSERT INTO objs VALUES %s", [Json(obj, dumps=uuid_dumps)])
# will insert: {'uuid': '0a40799d-3980-4c65-8315-2956b18ab0e1'}

Lists adaptation #

Python list objects are adapted to PostgreSQL arrays and back. Only lists containing objects of the same type can be dumped to PostgreSQL (but the list may contain None elements).

>>> conn.execute("SELECT * FROM mytable WHERE id IN %s", [[10,20,30]])
Traceback (most recent call last):
  File "", line 1, in 
psycopg.errors.SyntaxError: syntax error at or near "$1"
LINE 1: SELECT * FROM mytable WHERE id IN $1
                                          ^

>>> conn.execute("SELECT * FROM mytable WHERE id = ANY(%s)", [[10,20,30]])

UUID adaptation #

Python uuid.UUID objects are adapted to PostgreSQL UUID type and back:

>>> conn.execute("select gen_random_uuid()").fetchone()[0]
UUID('97f0dd62-3bd2-459e-89b8-a5e36ea3c16c')

>>> from uuid import uuid4
>>> conn.execute("select gen_random_uuid() = %s", [uuid4()]).fetchone()[0]
False  # long shot

Network data types adaptation #

Objects from the ipaddress module are converted to PostgreSQL network address types :

• IPv4Address , IPv4Interface objects are converted to the PostgreSQL inet type. On the way back, inet values indicating a single address are converted to IPv4Address , otherwise they are converted to IPv4Interface

• IPv4Network objects are converted to the cidr type and back.

• IPv6Address , IPv6Interface , IPv6Network objects follow the same rules, with IPv6 inet and cidr values.

>>> conn.execute("select '192.168.0.1'::inet, '192.168.0.1/24'::inet").fetchone()
(IPv4Address('192.168.0.1'), IPv4Interface('192.168.0.1/24'))

>>> conn.execute("select '::ffff:1.2.3.0/120'::cidr").fetchone()[0]
IPv6Network('::ffff:102:300/120')

Enum adaptation #

Psycopg can adapt Python Enum subclasses into PostgreSQL enum types (created with the CREATE TYPE ... AS ENUM (...) command).

In order to set up a bidirectional enum mapping, you should get information about the PostgreSQL enum using the EnumInfo class and register it using register_enum() . The behaviour of unregistered and registered enums is different.

• If the enum is not registered with register_enum() :

  • Pure Enum classes are dumped as normal strings, using their member names as value. The unknown oid is used, so PostgreSQL should be able to use this string in most contexts (such as an enum or a text field).

    Changed in version 3.1: In previous version dumping pure enums is not supported and raise a "cannot adapt" error.

  • Mix-in enums are dumped according to their mix-in type (because a class MyIntEnum(int, Enum) is more specifically an int than an Enum , so it’s dumped by default according to int rules).

  • PostgreSQL enums are loaded as Python strings. If you want to load arrays of such enums you will have to find their OIDs using types.TypeInfo.fetch() and register them using register() .

• If the enum is registered (using EnumInfo .fetch() and register_enum() ):

  • Enums classes, both pure and mixed-in, are dumped by name.

  • The registered PostgreSQL enum is loaded back as the registered Python enum members.

class psycopg.types.enum. EnumInfo ( name , oid , array_oid , labels ) #

Manage information about an enum type.

EnumInfo is a subclass of TypeInfo : refer to the latter’s documentation for generic usage, especially the fetch() method.

labels #

After fetch() , it contains the labels defined in the PostgreSQL enum type.

enum #

After register_enum() is called, it will contain the Python type mapping to the registered enum.

psycopg.types.enum. register_enum ( info , context = None , enum = None , * , mapping = None ) #

Register the adapters to load and dump a enum type.

Parameters :

• info ( EnumInfo ) - The object with the information about the enum to register.

• context ( Optional [ AdaptContext ]) - The context where to register the adapters. If None , register it globally.

• enum ( Optional [ Type [ TypeVar ( E , bound= Enum )]]) - Python enum type matching to the PostgreSQL one. If None , a new enum will be generated and exposed as EnumInfo.enum .

• mapping ( Union [ Mapping [ TypeVar ( E , bound= Enum ), str ], Sequence [ Tuple [ TypeVar ( E , bound= Enum ), str ]], None ]) - Override the mapping between enum members and info labels.

After registering, fetching data of the registered enum will cast PostgreSQL enum labels into corresponding Python enum members.

If no enum is specified, a new Enum is created based on PostgreSQL enum labels.

Example:

>>> from enum import Enum, auto
>>> from psycopg.types.enum import EnumInfo, register_enum

>>> class UserRole(Enum):
...     ADMIN = auto()
...     EDITOR = auto()
...     GUEST = auto()

>>> conn.execute("CREATE TYPE user_role AS ENUM ('ADMIN', 'EDITOR', 'GUEST')")

>>> info = EnumInfo.fetch(conn, "user_role")
>>> register_enum(info, conn, UserRole)

>>> some_editor = info.enum.EDITOR
>>> some_editor


>>> conn.execute(
...     "SELECT pg_typeof(%(editor)s), %(editor)s",
...     {"editor": some_editor}
... ).fetchone()
('user_role', )

>>> conn.execute(
...     "SELECT ARRAY[%s, %s]",
...     [UserRole.ADMIN, UserRole.GUEST]
... ).fetchone()
[, ]

If the Python and the PostgreSQL enum don’t match 1:1 (for instance if members have a different name, or if more than one Python enum should map to the same PostgreSQL enum, or vice versa), you can specify the exceptions using the mapping parameter.

mapping should be a dictionary with Python enum members as keys and the matching PostgreSQL enum labels as values, or a list of (member, label) pairs with the same meaning (useful when some members are repeated). Order matters: if an element on either side is specified more than once, the last pair in the sequence will take precedence:

# Legacy roles, defined in medieval times.
>>> conn.execute(
...     "CREATE TYPE abbey_role AS ENUM ('ABBOT', 'SCRIBE', 'MONK', 'GUEST')")

>>> info = EnumInfo.fetch(conn, "abbey_role")
>>> register_enum(info, conn, UserRole, mapping=[
...     (UserRole.ADMIN, "ABBOT"),
...     (UserRole.EDITOR, "SCRIBE"),
...     (UserRole.EDITOR, "MONK")])

>>> conn.execute("SELECT '{ABBOT,SCRIBE,MONK,GUEST}'::abbey_role[]").fetchone()[0]
[<UserRole.ADMIN: 1>,
 <UserRole.EDITOR: 2>,
 <UserRole.EDITOR: 2>,
 <UserRole.GUEST: 3>]

>>> conn.execute("SELECT %s::text[]", [list(UserRole)]).fetchone()[0]
['ABBOT', 'MONK', 'GUEST']

A particularly useful case is when the PostgreSQL labels match the values of a str -based Enum. In this case it is possible to use something like {m: m.value for m in enum} as mapping:

>>> class LowercaseRole(str, Enum):
...     ADMIN = "admin"
...     EDITOR = "editor"
...     GUEST = "guest"

>>> conn.execute(
...     "CREATE TYPE lowercase_role AS ENUM ('admin', 'editor', 'guest')")

>>> info = EnumInfo.fetch(conn, "lowercase_role")
>>> register_enum(
...     info, conn, LowercaseRole, mapping={m: m.value for m in LowercaseRole})

>>> conn.execute("SELECT 'editor'::lowercase_role").fetchone()[0]

Composite types casting #

Psycopg can adapt PostgreSQL composite types (either created with the CREATE TYPE command or implicitly defined after a table row type) to and from Python tuples, namedtuple , or any other suitable object configured.

Before using a composite type it is necessary to get information about it using the CompositeInfo class and to register it using register_composite() .

class psycopg.types.composite. CompositeInfo ( name , oid , array_oid , * , regtype = '' , field_names , field_types ) #

Manage information about a composite type.

CompositeInfo is a TypeInfo subclass: check its documentation for the generic usage, especially the fetch() method.

python_type #

After register_composite() is called, it will contain the python type mapping to the registered composite.

psycopg.types.composite. register_composite ( info , context = None , factory = None ) #

Register the adapters to load and dump a composite type.

Parameters :

• info ( CompositeInfo ) - The object with the information about the composite to register.

• context ( Optional [ AdaptContext ]) - The context where to register the adapters. If None , register it globally.

• factory ( Optional [ Callable [ ... , Any ]]) - Callable to convert the sequence of attributes read from the composite into a Python object.

Note

Registering the adapters doesn’t affect objects already created, even if they are children of the registered context. For instance, registering the adapter globally doesn’t affect already existing connections.

After registering, fetching data of the registered composite will invoke factory to create corresponding Python objects.

If no factory is specified, a namedtuple is created and used to return data.

If the factory is a type (and not a generic callable), then dumpers for that type are created and registered too, so that passing objects of that type to a query will adapt them to the registered type.

Example:

>>> from psycopg.types.composite import CompositeInfo, register_composite

>>> conn.execute("CREATE TYPE card AS (value int, suit text)")

>>> info = CompositeInfo.fetch(conn, "card")
>>> register_composite(info, conn)

>>> my_card = info.python_type(8, "hearts")
>>> my_card
card(value=8, suit='hearts')

>>> conn.execute(
...     "SELECT pg_typeof(%(card)s), (%(card)s).suit", {"card": my_card}
...     ).fetchone()
('card', 'hearts')

>>> conn.execute("SELECT (%s, %s)::card", [1, "spades"]).fetchone()[0]
card(value=1, suit='spades')

Nested composite types are handled as expected, provided that the type of the composite components are registered as well:

>>> conn.execute("CREATE TYPE card_back AS (face card, back text)")

>>> info2 = CompositeInfo.fetch(conn, "card_back")
>>> register_composite(info2, conn)

>>> conn.execute("SELECT ((8, 'hearts'), 'blue')::card_back").fetchone()[0]
card_back(face=card(value=8, suit='hearts'), back='blue')

Range adaptation #

PostgreSQL range types are a family of data types representing a range of values between two elements. The type of the element is called the range subtype . PostgreSQL offers a few built-in range types and allows the definition of custom ones.

All the PostgreSQL range types are loaded as the Range Python type, which is a Generic type and can hold bounds of different types.

class psycopg.types.range. Range ( lower = None , upper = None , bounds = '[)' , empty = False ) #

Python representation for a PostgreSQL range type.

Parameters :

• lower ( Optional [ TypeVar ( T )]) - lower bound for the range. None means unbound

• upper ( Optional [ TypeVar ( T )]) - upper bound for the range. None means unbound

• bounds ( str ) - one of the literal strings () , [) , (] , [] , representing whether the lower or upper bounds are included

• empty ( bool ) - if True , the range is empty

This Python type is only used to pass and retrieve range values to and from PostgreSQL and doesn’t attempt to replicate the PostgreSQL range features: it doesn’t perform normalization and doesn’t implement all the operators supported by the database.

PostgreSQL will perform normalisation on Range objects used as query parameters, so, when they are fetched back, they will be found in the normal form (for instance ranges on integers will have [) bounds).

Range objects are immutable, hashable, and support the in operator (checking if an element is within the range). They can be tested for equivalence. Empty ranges evaluate to False in a boolean context, nonempty ones evaluate to True .

Range objects have the following attributes:

isempty #

True if the range is empty.

lower #

The lower bound of the range. None if empty or unbound.

upper #

The upper bound of the range. None if empty or unbound.

lower_inc #

True if the lower bound is included in the range.

upper_inc #

True if the upper bound is included in the range.

lower_inf #

True if the range doesn’t have a lower bound.

upper_inf #

True if the range doesn’t have an upper bound.

The built-in range objects are adapted automatically: if a Range objects contains date bounds, it is dumped using the daterange OID, and of course daterange values are loaded back as Range[date] .

If you create your own range type you can use RangeInfo and register_range() to associate the range type with its subtype and make it work like the builtin ones.

class psycopg.types.range. RangeInfo ( name , oid , array_oid , * , regtype = '' , subtype_oid ) #

Manage information about a range type.

RangeInfo is a TypeInfo subclass: check its documentation for generic details, especially the fetch() method.

psycopg.types.range. register_range ( info , context = None ) #

Register the adapters to load and dump a range type.

Parameters :

• info ( RangeInfo ) - The object with the information about the range to register.

• context ( Optional [ AdaptContext ]) - The context where to register the adapters. If None , register it globally.

Register loaders so that loading data of this type will result in a Range with bounds parsed as the right subtype.

Note

Registering the adapters doesn’t affect objects already created, even if they are children of the registered context. For instance, registering the adapter globally doesn’t affect already existing connections.

Example:

>>> from psycopg.types.range import Range, RangeInfo, register_range

>>> conn.execute("CREATE TYPE strrange AS RANGE (SUBTYPE = text)")
>>> info = RangeInfo.fetch(conn, "strrange")
>>> register_range(info, conn)

>>> conn.execute("SELECT pg_typeof(%s)", [Range("a", "z")]).fetchone()[0]
'strrange'

>>> conn.execute("SELECT '[a,z]'::strrange").fetchone()[0]
Range('a', 'z', '[]')

Multirange adaptation #

Since PostgreSQL 14, every range type is associated with a multirange , a type representing a disjoint set of ranges. A multirange is automatically available for every range, built-in and user-defined.

All the PostgreSQL range types are loaded as the Multirange Python type, which is a mutable sequence of Range elements.

class psycopg.types.multirange. Multirange ( items = () ) #

Python representation for a PostgreSQL multirange type.

Parameters :

items ( Iterable [ Range [ TypeVar ( T )]]) - Sequence of ranges to initialise the object.

This Python type is only used to pass and retrieve multirange values to and from PostgreSQL and doesn’t attempt to replicate the PostgreSQL multirange features: overlapping items are not merged, empty ranges are not discarded, the items are not ordered, the behaviour of multirange operators is not replicated in Python.

PostgreSQL will perform normalisation on Multirange objects used as query parameters, so, when they are fetched back, they will be found ordered, with overlapping ranges merged, etc.

Multirange objects are a MutableSequence and are totally ordered: they behave pretty much like a list of Range . Like Range, they are Generic on the subtype of their range, so you can declare a variable to be Multirange[date] and mypy will complain if you try to add it a Range[Decimal] .

Like for Range , built-in multirange objects are adapted automatically: if a Multirange object contains Range with date bounds, it is dumped using the datemultirange OID, and datemultirange values are loaded back as Multirange[date] .

If you have created your own range type you can use MultirangeInfo and register_multirange() to associate the resulting multirange type with its subtype and make it work like the builtin ones.

class psycopg.types.multirange. MultirangeInfo ( name , oid , array_oid , * , regtype = '' , range_oid , subtype_oid ) #

Manage information about a multirange type.

MultirangeInfo is a TypeInfo subclass: check its documentation for generic details, especially the fetch() method.

psycopg.types.multirange. register_multirange ( info , context = None ) #

Register the adapters to load and dump a multirange type.

Parameters :

• info ( MultirangeInfo ) - The object with the information about the range to register.

• context ( Optional [ AdaptContext ]) - The context where to register the adapters. If None , register it globally.

Register loaders so that loading data of this type will result in a Range with bounds parsed as the right subtype.

Note

Registering the adapters doesn’t affect objects already created, even if they are children of the registered context. For instance, registering the adapter globally doesn’t affect already existing connections.

Example:

>>> from psycopg.types.multirange import \
...     Multirange, MultirangeInfo, register_multirange
>>> from psycopg.types.range import Range

>>> conn.execute("CREATE TYPE strrange AS RANGE (SUBTYPE = text)")
>>> info = MultirangeInfo.fetch(conn, "strmultirange")
>>> register_multirange(info, conn)

>>> rec = conn.execute(
...     "SELECT pg_typeof(%(mr)s), %(mr)s",
...     {"mr": Multirange([Range("a", "q"), Range("l", "z")])}).fetchone()

>>> rec[0]
'strmultirange'
>>> rec[1]
Multirange([Range('a', 'z', '[)')])

Hstore adaptation #

The hstore data type is a key-value store embedded in PostgreSQL. It supports GiST or GIN indexes allowing search by keys or key/value pairs as well as regular BTree indexes for equality, uniqueness etc.

Psycopg can convert Python dict objects to and from hstore structures. Only dictionaries with string keys and values are supported. None is also allowed as value but not as a key.

In order to use the hstore data type it is necessary to load it in a database using:

=# CREATE EXTENSION hstore;

Because hstore is distributed as a contrib module, its oid is not well known, so it is necessary to use TypeInfo . fetch() to query the database and get its oid. The resulting object can be passed to register_hstore() to configure dumping dict to hstore and parsing hstore back to dict , in the context where the adapter is registered.

psycopg.types.hstore. register_hstore ( info , context = None ) #

Register the adapters to load and dump hstore.

Parameters :

• info ( TypeInfo ) - The object with the information about the hstore type.

• context ( Optional [ AdaptContext ]) - The context where to register the adapters. If None , register it globally.

Note

Registering the adapters doesn’t affect objects already created, even if they are children of the registered context. For instance, registering the adapter globally doesn’t affect already existing connections.

Example:

>>> from psycopg.types import TypeInfo
>>> from psycopg.types.hstore import register_hstore

>>> info = TypeInfo.fetch(conn, "hstore")
>>> register_hstore(info, conn)

>>> conn.execute("SELECT pg_typeof(%s)", [{"a": "b"}]).fetchone()[0]
'hstore'

>>> conn.execute("SELECT 'foo => bar'::hstore").fetchone()[0]
{'foo': 'bar'}

Geometry adaptation using Shapely #

When using the PostGIS extension, it can be useful to retrieve geometry values and have them automatically converted to Shapely instances. Likewise, you may want to store such instances in the database and have the conversion happen automatically.

Since PostgGIS is an extension, the geometry type oid is not well known, so it is necessary to use TypeInfo . fetch() to query the database and find it. The resulting object can be passed to register_shapely() to configure dumping shape instances to geometry columns and parsing geometry data back to shape instances, in the context where the adapters are registered.

psycopg.types.shapely. register_shapely ( ) #

Register Shapely dumper and loaders.

After invoking this function on an adapter, the queries retrieving PostGIS geometry objects will return Shapely’s shape object instances both in text and binary mode.

Similarly, shape objects can be sent to the database.

This requires the Shapely library to be installed.

Parameters :

• info - The object with the information about the geometry type.

• context - The context where to register the adapters. If None , register it globally.

Note

Registering the adapters doesn’t affect objects already created, even if they are children of the registered context. For instance, registering the adapter globally doesn’t affect already existing connections.

Example:

>>> from psycopg.types import TypeInfo
>>> from psycopg.types.shapely import register_shapely
>>> from shapely.geometry import Point

>>> info = TypeInfo.fetch(conn, "geometry")
>>> register_shapely(info, conn)

>>> conn.execute("SELECT pg_typeof(%s)", [Point(1.2, 3.4)]).fetchone()[0]
'geometry'

>>> conn.execute("""
... SELECT ST_GeomFromGeoJSON('{
...     "type":"Point",
...     "coordinates":[-48.23456,20.12345]}')
... """).fetchone()[0]

Notice that, if the geometry adapters are registered on a specific object (a connection or cursor), other connections and cursors will be unaffected:

>>> conn2 = psycopg.connect(CONN_STR)
>>> conn2.execute("""
... SELECT ST_GeomFromGeoJSON('{
...     "type":"Point",
...     "coordinates":[-48.23456,20.12345]}')
... """).fetchone()[0]
'0101000020E61000009279E40F061E48C0F2B0506B9A1F3440'

Autocommit transactions #

The manual commit requirement can be suspended using autocommit , either as connection attribute or as connect() parameter. This may be required to run operations that cannot be executed inside a transaction, such as CREATE DATABASE , VACUUM , CALL on stored procedures using transaction control.

With an autocommit transaction, the above sequence of operation results in:

with psycopg.connect(autocommit=True) as conn:

    cur = conn.cursor()

    cur.execute("SELECT count(*) FROM my_table")
    # This function call now only executes:
    # - SELECT count(*) FROM my_table
    # and no transaction starts.

    cur.execute("INSERT INTO data VALUES (%s)", ("Hello",))
    # The result of this statement is persisted immediately by the database

# The connection is closed at the end of the block but, because it is not
# in a transaction state, no COMMIT is executed.

An autocommit transaction behaves more as someone coming from psql would expect. This has a beneficial performance effect, because less queries are sent and less operations are performed by the database. The statements, however, are not executed in an atomic transaction; if you need to execute certain operations inside a transaction, you can achieve that with an autocommit connection too, using an explicit transaction block .

Transaction contexts #

A more transparent way to make sure that transactions are finalised at the right time is to use with Connection.transaction() to create a transaction context. When the context is entered, a transaction is started; when leaving the context the transaction is committed, or it is rolled back if an exception is raised inside the block.

Continuing the example above, if you want to use an autocommit connection but still wrap selected groups of commands inside an atomic transaction, you can use a transaction() context:

with psycopg.connect(autocommit=True) as conn:

    cur = conn.cursor()

    cur.execute("SELECT count(*) FROM my_table")
    # The connection is autocommit, so no BEGIN executed.

    with conn.transaction():
        # BEGIN is executed, a transaction started

        cur.execute("INSERT INTO data VALUES (%s)", ("Hello",))
        cur.execute("INSERT INTO times VALUES (now())")
        # These two operation run atomically in the same transaction

    # COMMIT is executed at the end of the block.
    # The connection is in idle state again.

# The connection is closed at the end of the block.

Note that connection blocks can also be used with non-autocommit connections: in this case you still need to pay attention to eventual transactions started automatically. If an operation starts an implicit transaction, a transaction() block will only manage a savepoint sub-transaction , leaving the caller to deal with the main transaction, as explained in Transactions management :

conn = psycopg.connect()

cur = conn.cursor()

cur.execute("SELECT count(*) FROM my_table")
# This function call executes:
# - BEGIN
# - SELECT count(*) FROM my_table
# So now a transaction has started.

with conn.transaction():
    # The block starts with a transaction already open, so it will execute
    # - SAVEPOINT

    cur.execute("INSERT INTO data VALUES (%s)", ("Hello",))

# The block was executing a sub-transaction so on exit it will only run:
# - RELEASE SAVEPOINT
# The transaction is still on.

conn.close()
# No COMMIT was sent: the INSERT was discarded.

If a transaction() block starts when no transaction is active then it will manage a proper transaction. In essence, a transaction context tries to leave a connection in the state it found it, and leaves you to deal with the wider context.

with conn.transaction() as tx1:
    num_ok = 0
    for operation in operations:
        try:
            with conn.transaction() as tx2:
                unreliable_operation(conn, operation)
        except Exception:
            logger.exception(f"{operation} failed")
        else:
            num_ok += 1

    save_number_of_successes(conn, num_ok)

from psycopg import Rollback

with conn.transaction() as outer_tx:
    for command in commands():
        with conn.transaction() as inner_tx:
            if isinstance(command, CancelCommand):
                raise Rollback(outer_tx)
            process_command(command)

# If `Rollback` is raised, it would propagate only up to this block,
# and the program would continue from here with no exception.

Transaction characteristics #

You can set transaction parameters for the transactions that Psycopg handles. They affect the transactions started implicitly by non-autocommit transactions and the ones started explicitly by Connection.transaction() for both autocommit and non-autocommit transactions. Leaving these parameters as None will use the server’s default behaviour (which is controlled by server settings such as default_transaction_isolation ).

In order to set these parameters you can use the connection attributes isolation_level , read_only , deferrable . For async connections you must use the equivalent set_isolation_level() method and similar. The parameters can only be changed if there isn’t a transaction already active on the connection.

psycopg2.errors.SerializationFailure: could not serialize access
due to concurrent update

Two-Phase Commit protocol support #

Psycopg exposes the two-phase commit features available in PostgreSQL implementing the two-phase commit extensions proposed by the DBAPI.

The DBAPI model of two-phase commit is inspired by the XA specification , according to which transaction IDs are formed from three components:

• a format ID (non-negative 32 bit integer)

• a global transaction ID (string not longer than 64 bytes)

• a branch qualifier (string not longer than 64 bytes)

For a particular global transaction, the first two components will be the same for all the resources. Every resource will be assigned a different branch qualifier.

According to the DBAPI specification, a transaction ID is created using the Connection.xid() method. Once you have a transaction id, a distributed transaction can be started with Connection.tpc_begin() , prepared using tpc_prepare() and completed using tpc_commit() or tpc_rollback() . Transaction IDs can also be retrieved from the database using tpc_recover() and completed using the above tpc_commit() and tpc_rollback() .

PostgreSQL doesn’t follow the XA standard though, and the ID for a PostgreSQL prepared transaction can be any string up to 200 characters long. Psycopg’s Xid objects can represent both XA-style transactions IDs (such as the ones created by the xid() method) and PostgreSQL transaction IDs identified by an unparsed string.

The format in which the Xids are converted into strings passed to the database is the same employed by the PostgreSQL JDBC driver : this should allow interoperation between tools written in Python and in Java. For example a recovery tool written in Python would be able to recognize the components of transactions produced by a Java program.

For further details see the documentation for the Two-Phase Commit support methods .

Writing data row-by-row #

Using a copy operation you can load data into the database from any Python iterable (a list of tuples, or any iterable of sequences): the Python values are adapted as they would be in normal querying. To perform such operation use a COPY ... FROM STDIN with Cursor.copy() and use write_row() on the resulting object in a with block. On exiting the block the operation will be concluded:

records = [(10, 20, "hello"), (40, None, "world")]

with cursor.copy("COPY sample (col1, col2, col3) FROM STDIN") as copy:
    for record in records:
        copy.write_row(record)

If an exception is raised inside the block, the operation is interrupted and the records inserted so far are discarded.

In order to read or write from Copy row-by-row you must not specify COPY options such as FORMAT CSV , DELIMITER , NULL : please leave these details alone, thank you :)

Reading data row-by-row #

You can also do the opposite, reading rows out of a COPY ... TO STDOUT operation, by iterating on rows() . However this is not something you may want to do normally: usually the normal query process will be easier to use.

PostgreSQL, currently, doesn’t give complete type information on COPY TO , so the rows returned will have unparsed data, as strings or bytes, according to the format.

with cur.copy("COPY (VALUES (10::int, current_date)) TO STDOUT") as copy:
    for row in copy.rows():
        print(row)  # return unparsed data: ('10', '2046-12-24')

You can improve the results by using set_types() before reading, but you have to specify them yourself.

with cur.copy("COPY (VALUES (10::int, current_date)) TO STDOUT") as copy:
    copy.set_types(["int4", "date"])
    for row in copy.rows():
        print(row)  # (10, datetime.date(2046, 12, 24))

Copying block-by-block #

If data is already formatted in a way suitable for copy (for instance because it is coming from a file resulting from a previous COPY TO operation) it can be loaded into the database using Copy.write() instead.

with open("data", "r") as f:
    with cursor.copy("COPY data FROM STDIN") as copy:
        while data := f.read(BLOCK_SIZE):
            copy.write(data)

In this case you can use any COPY option and format, as long as the input data is compatible with what the operation in copy() expects. Data can be passed as str , if the copy is in FORMAT TEXT , or as bytes , which works with both FORMAT TEXT and FORMAT BINARY .

In order to produce data in COPY format you can use a COPY ... TO STDOUT statement and iterate over the resulting Copy object, which will produce a stream of bytes objects:

with open("data.out", "wb") as f:
    with cursor.copy("COPY table_name TO STDOUT") as copy:
        for data in copy:
            f.write(data)

Binary copy #

Binary copy is supported by specifying FORMAT BINARY in the COPY statement. In order to import binary data using write_row() , all the types passed to the database must have a binary dumper registered; this is not necessary if the data is copied block-by-block using write() .

Asynchronous copy support #

Asynchronous operations are supported using the same patterns as above, using the objects obtained by an AsyncConnection . For instance, if f is an object supporting an asynchronous read() method returning COPY data, a fully-async copy operation could be:

async with cursor.copy("COPY data FROM STDIN") as copy:
    while data := await f.read():
        await copy.write(data)

The AsyncCopy object documentation describes the signature of the asynchronous methods and the differences from its sync Copy counterpart.

Example: copying a table across servers #

In order to copy a table, or a portion of a table, across servers, you can use two COPY operations on two different connections, reading from the first and writing to the second.

with psycopg.connect(dsn_src) as conn1, psycopg.connect(dsn_tgt) as conn2:
    with conn1.cursor().copy("COPY src TO STDOUT (FORMAT BINARY)") as copy1:
        with conn2.cursor().copy("COPY tgt FROM STDIN (FORMAT BINARY)") as copy2:
            for data in copy1:
                copy2.write(data)

Using FORMAT BINARY usually gives a performance boost, but it only works if the source and target schema are perfectly identical . If the tables are only compatible (for example, if you are copying an integer field into a bigint destination field) you should omit the BINARY option and perform a text-based copy. See Binary copy for details.

The same pattern can be adapted to use async objects in order to perform an async copy .

Server-side binding #

Psycopg 3 sends the query and the parameters to the server separately, instead of merging them on the client side. Server-side binding works for normal SELECT and data manipulation statements ( INSERT , UPDATE , DELETE ), but it doesn’t work with many other statements. For instance, it doesn’t work with SET or with NOTIFY :

>>> conn.execute("SET TimeZone TO %s", ["UTC"])
Traceback (most recent call last):
...
psycopg.errors.SyntaxError: syntax error at or near "$1"
LINE 1: SET TimeZone TO $1
                        ^

>>> conn.execute("NOTIFY %s, %s", ["chan", 42])
Traceback (most recent call last):
...
psycopg.errors.SyntaxError: syntax error at or near "$1"
LINE 1: NOTIFY $1, $2
               ^

and with any data definition statement:

>>> conn.execute("CREATE TABLE foo (id int DEFAULT %s)", [42])
Traceback (most recent call last):
...
psycopg.errors.UndefinedParameter: there is no parameter $1
LINE 1: CREATE TABLE foo (id int DEFAULT $1)
                                         ^

Sometimes, PostgreSQL offers an alternative: for instance the set_config() function can be used instead of the SET statement, the pg_notify() function can be used instead of NOTIFY :

>>> conn.execute("SELECT set_config('TimeZone', %s, false)", ["UTC"])

>>> conn.execute("SELECT pg_notify(%s, %s)", ["chan", "42"])

If this is not possible, you must merge the query and the parameter on the client side. You can do so using the psycopg.sql objects:

>>> from psycopg import sql

>>> cur.execute(sql.SQL("CREATE TABLE foo (id int DEFAULT {})").format(42))

or creating a client-side binding cursor such as ClientCursor :

>>> cur = ClientCursor(conn)
>>> cur.execute("CREATE TABLE foo (id int DEFAULT %s)", [42])

If you need ClientCursor often, you can set the Connection.cursor_factory to have them created by default by Connection.cursor() . This way, Psycopg 3 will behave largely the same way of Psycopg 2.

Note that, both server-side and client-side, you can only specify values as parameters (i.e. the strings that go in single quotes ). If you need to parametrize different parts of a statement (such as a table name), you must use the psycopg.sql module:

>>> from psycopg import sql

# This will quote the user and the password using the right quotes
# e.g.: ALTER USER "foo" SET PASSWORD 'bar'
>>> conn.execute(
...     sql.SQL("ALTER USER {} SET PASSWORD {}")
...     .format(sql.Identifier(username), password))

Multiple statements in the same query #

As a consequence of using server-side bindings , when parameters are used, it is not possible to execute several statements in the same execute() call, separating them by semicolon:

>>> conn.execute(
...     "INSERT INTO foo VALUES (%s); INSERT INTO foo VALUES (%s)",
...     (10, 20))
Traceback (most recent call last):
...
psycopg.errors.SyntaxError: cannot insert multiple commands into a prepared statement

One obvious way to work around the problem is to use several execute() calls.

There is no such limitation if no parameters are used . As a consequence, you can compose a multiple query on the client side and run them all in the same execute() call, using the psycopg.sql objects:

>>> from psycopg import sql
>>> conn.execute(
...     sql.SQL("INSERT INTO foo VALUES ({}); INSERT INTO foo values ({})"
...     .format(10, 20))

or a client-side binding cursor :

>>> cur = psycopg.ClientCursor(conn)
>>> cur.execute(
...     "INSERT INTO foo VALUES (%s); INSERT INTO foo VALUES (%s)",
...     (10, 20))

>>> conn.autocommit = True
>>> conn.execute("CREATE DATABASE foo; SELECT 1")
Traceback (most recent call last):
...
psycopg.errors.ActiveSqlTransaction: CREATE DATABASE cannot run inside a transaction block

Multiple results returned from multiple statements #

If more than one statement returning results is executed in psycopg2, only the result of the last statement is returned:

>>> cur_pg2.execute("SELECT 1; SELECT 2")
>>> cur_pg2.fetchone()
(2,)

In Psycopg 3 instead, all the results are available. After running the query, the first result will be readily available in the cursor and can be consumed using the usual fetch*() methods. In order to access the following results, you can use the Cursor.nextset() method:

>>> cur_pg3.execute("SELECT 1; SELECT 2")
>>> cur_pg3.fetchone()
(1,)

>>> cur_pg3.nextset()
True
>>> cur_pg3.fetchone()
(2,)

>>> cur_pg3.nextset()
None  # no more results

Remember though that you cannot use server-side bindings to execute more than one statement in the same query .

Different cast rules #

In rare cases, especially around variadic functions, PostgreSQL might fail to find a function candidate for the given data types:

>>> conn.execute("SELECT json_build_array(%s, %s)", ["foo", "bar"])
Traceback (most recent call last):
...
psycopg.errors.IndeterminateDatatype: could not determine data type of parameter $1

This can be worked around specifying the argument types explicitly via a cast:

>>> conn.execute("SELECT json_build_array(%s::text, %s::text)", ["foo", "bar"])

You cannot use IN %s with a tuple #

IN cannot be used with a tuple as single parameter, as was possible with psycopg2 :

>>> conn.execute("SELECT * FROM foo WHERE id IN %s", [(10,20,30)])
Traceback (most recent call last):
...
psycopg.errors.SyntaxError: syntax error at or near "$1"
LINE 1: SELECT * FROM foo WHERE id IN $1
                                      ^

What you can do is to use the = ANY() construct and pass the candidate values as a list instead of a tuple, which will be adapted to a PostgreSQL array:

>>> conn.execute("SELECT * FROM foo WHERE id = ANY(%s)", [[10,20,30]])

Note that ANY() can be used with psycopg2 too, and has the advantage of accepting an empty list of values too as argument, which is not supported by the IN operator instead.

Different adaptation system #

The adaptation system has been completely rewritten, in order to address server-side parameters adaptation, but also to consider performance, flexibility, ease of customization.

The default behaviour with builtin data should be what you would expect . If you have customised the way to adapt data, or if you are managing your own extension types, you should look at the new adaptation system .

Copy is no longer file-based #

psycopg2 exposes a few copy methods to interact with PostgreSQL COPY . Their file-based interface doesn’t make it easy to load dynamically-generated data into a database.

There is now a single copy() method, which is similar to psycopg2 copy_expert() in accepting a free-form COPY command and returns an object to read/write data, block-wise or record-wise. The different usage pattern also enables COPY to be used in async interactions.

with connection #

In psycopg2 , using the syntax with connection , only the transaction is closed, not the connection. This behaviour is surprising for people used to several other Python classes wrapping resources, such as files.

In Psycopg 3, using with connection will close the connection at the end of the with block, making handling the connection resources more familiar.

In order to manage transactions as blocks you can use the Connection.transaction() method, which allows for finer control, for instance to use nested transactions.

callproc() is gone #

cursor.callproc() is not implemented. The method has a simplistic semantic which doesn’t account for PostgreSQL positional parameters, procedures, set-returning functions… Use a normal execute() with SELECT function_name(...) or CALL procedure_name(...) instead.

client_encoding is gone #

Psycopg automatically uses the database client encoding to decode data to Unicode strings. Use ConnectionInfo.encoding if you need to read the encoding. You can select an encoding at connection time using the client_encoding connection parameter and you can change the encoding of a connection by running a SET client_encoding statement… But why would you?

No default infinity dates handling #

PostgreSQL can represent a much wider range of dates and timestamps than Python. While Python dates are limited to the years between 1 and 9999 (represented by constants such as datetime.date.min and max ), PostgreSQL dates extend to BC dates and past the year 10K. Furthermore PostgreSQL can also represent symbolic dates "infinity", in both directions.

In psycopg2, by default, infinity dates and timestamps map to ‘date.max’ and similar constants. This has the problem of creating a non-bijective mapping (two Postgres dates, infinity and 9999-12-31, both map to the same Python date). There is also the perversity that valid Postgres dates, greater than Python date.max but arguably lesser than infinity, will still overflow.

In Psycopg 3, every date greater than year 9999 will overflow, including infinity. If you would like to customize this mapping (for instance flattening every date past Y10K on date.max ) you can subclass and adapt the appropriate loaders: take a look at this example to see how.

What’s new in Psycopg 3 #

• Asynchronous support

• Server-side parameters binding

• Prepared statements

• Binary communication

• Python-based COPY support

• Support for static typing

• A redesigned connection pool

• Direct access to the libpq functionalities

with async connections #

As seen in the basic usage , connections and cursors can act as context managers, so you can run:

with psycopg.connect("dbname=test user=postgres") as conn:
    with conn.cursor() as cur:
        cur.execute(...)
    # the cursor is closed upon leaving the context
# the transaction is committed, the connection closed

For asynchronous connections it’s almost what you’d expect, but not quite. Please note that connect() and cursor() don’t return a context : they are both factory methods which return an object which can be used as a context . That’s because there are several use cases where it’s useful to handle the objects manually and only close() them when required.

As a consequence you cannot use async with connect() : you have to do it in two steps instead, as in

aconn = await psycopg.AsyncConnection.connect()
async with aconn:
    async with aconn.cursor() as cur:
        await cur.execute(...)

which can be condensed into async with await :

async with await psycopg.AsyncConnection.connect() as aconn:
    async with aconn.cursor() as cur:
        await cur.execute(...)

…but no less than that: you still need to do the double async thing.

Note that the AsyncConnection.cursor() function is not an async function (it never performs I/O), so you don’t need an await on it; as a consequence you can use the normal async with context manager.

Interrupting async operations using Ctrl-C #

If a long running operation is interrupted by a Ctrl-C on a normal connection running in the main thread, the operation will be cancelled and the connection will be put in error state, from which can be recovered with a normal rollback() .

If the query is running in an async connection, a Ctrl-C will be likely intercepted by the async loop and interrupt the whole program. In order to emulate what normally happens with blocking connections, you can use asyncio’s add_signal_handler() , to call Connection.cancel() :

import asyncio
import signal

async with await psycopg.AsyncConnection.connect() as conn:
    loop.add_signal_handler(signal.SIGINT, conn.cancel)
    ...

Server messages #

PostgreSQL can send, together with the query results, informative messages about the operation just performed, such as warnings or debug information. Notices may be raised even if the operations are successful and don’t indicate an error. You are probably familiar with some of them, because they are reported by psql :

$ psql
=# ROLLBACK;
WARNING:  there is no transaction in progress
ROLLBACK

Messages can be also sent by the PL/pgSQL ‘RAISE’ statement (at a level lower than EXCEPTION, otherwise the appropriate DatabaseError will be raised). The level of the messages received can be controlled using the client_min_messages setting.

By default, the messages received are ignored. If you want to process them on the client you can use the Connection.add_notice_handler() function to register a function that will be invoked whenever a message is received. The message is passed to the callback as a Diagnostic instance, containing all the information passed by the server, such as the message text and the severity. The object is the same found on the diag attribute of the errors raised by the server:

>>> import psycopg

>>> def log_notice(diag):
...     print(f"The server says: {diag.severity} - {diag.message_primary}")

>>> conn = psycopg.connect(autocommit=True)
>>> conn.add_notice_handler(log_notice)

>>> cur = conn.execute("ROLLBACK")
The server says: WARNING - there is no transaction in progress
>>> print(cur.statusmessage)
ROLLBACK

Asynchronous notifications #

Psycopg allows asynchronous interaction with other database sessions using the facilities offered by PostgreSQL commands LISTEN and NOTIFY . Please refer to the PostgreSQL documentation for examples about how to use this form of communication.

Because of the way sessions interact with notifications (see NOTIFY documentation), you should keep the connection in autocommit mode if you wish to receive or send notifications in a timely manner.

Notifications are received as instances of Notify . If you are reserving a connection only to receive notifications, the simplest way is to consume the Connection.notifies generator. The generator can be stopped using close() .

The following example will print notifications and stop when one containing the "stop" message is received.

import psycopg
conn = psycopg.connect("", autocommit=True)
conn.execute("LISTEN mychan")
gen = conn.notifies()
for notify in gen:
    print(notify)
    if notify.payload == "stop":
        gen.close()
print("there, I stopped")

If you run some NOTIFY in a psql session:

=# NOTIFY mychan, 'hello';
NOTIFY
=# NOTIFY mychan, 'hey';
NOTIFY
=# NOTIFY mychan, 'stop';
NOTIFY

You may get output from the Python process such as:

Notify(channel='mychan', payload='hello', pid=961823)
Notify(channel='mychan', payload='hey', pid=961823)
Notify(channel='mychan', payload='stop', pid=961823)
there, I stopped

Alternatively, you can use add_notify_handler() to register a callback function, which will be invoked whenever a notification is received, during the normal query processing; you will be then able to use the connection normally. Please note that in this case notifications will not be received immediately, but only during a connection operation, such as a query.

conn.add_notify_handler(lambda n: print(f"got this: {n}"))

# meanwhile in psql...
# =# NOTIFY mychan, 'hey';
# NOTIFY

print(conn.execute("SELECT 1").fetchone())
# got this: Notify(channel='mychan', payload='hey', pid=961823)
# (1,)

Detecting disconnections #

Sometimes it is useful to detect immediately when the connection with the database is lost. One brutal way to do so is to poll a connection in a loop running an endless stream of SELECT 1 … Don’t do so: polling is so out of fashion. Besides, it is inefficient (unless what you really want is a client-server generator of ones), it generates useless traffic and will only detect a disconnection with an average delay of half the polling time.

A more efficient and timely way to detect a server disconnection is to create an additional connection and wait for a notification from the OS that this connection has something to say: only then you can run some checks. You can dedicate a thread (or an asyncio task) to wait on this connection: such thread will perform no activity until awaken by the OS.

In a normal (non asyncio) program you can use the selectors module. Because the Connection implements a fileno() method you can just register it as a file-like object. You can run such code in a dedicated thread (and using a dedicated connection) if the rest of the program happens to have something else to do too.

import selectors

sel = selectors.DefaultSelector()
sel.register(conn, selectors.EVENT_READ)
while True:
    if not sel.select(timeout=60.0):
        continue  # No FD activity detected in one minute

    # Activity detected. Is the connection still ok?
    try:
        conn.execute("SELECT 1")
    except psycopg.OperationalError:
        # You were disconnected: do something useful such as panicking
        logger.error("we lost our database!")
        sys.exit(1)

In an asyncio program you can dedicate a Task instead and do something similar using add_reader :

import asyncio

ev = asyncio.Event()
loop = asyncio.get_event_loop()
loop.add_reader(conn.fileno(), ev.set)

while True:
    try:
        await asyncio.wait_for(ev.wait(), 60.0)
    except asyncio.TimeoutError:
        continue  # No FD activity detected in one minute

    # Activity detected. Is the connection still ok?
    try:
        await conn.execute("SELECT 1")
    except psycopg.OperationalError:
        # Guess what happened
        ...

Generic types #

Psycopg Connection and Cursor objects are Generic objects and support a Row parameter which is the type of the records returned.

By default methods such as Cursor.fetchall() return normal tuples of unknown size and content. As such, the connect() function returns an object of type psycopg.Connection[Tuple[Any, ...]] and Connection.cursor() returns an object of type psycopg.Cursor[Tuple[Any, ...]] . If you are writing generic plumbing code it might be practical to use annotations such as Connection[Any] and Cursor[Any] .

conn = psycopg.connect() # type is psycopg.Connection[Tuple[Any, ...]]

cur = conn.cursor()      # type is psycopg.Cursor[Tuple[Any, ...]]

rec = cur.fetchone()     # type is Optional[Tuple[Any, ...]]

recs = cur.fetchall()    # type is List[Tuple[Any, ...]]

Type of rows returned #

If you want to use connections and cursors returning your data as different types, for instance as dictionaries, you can use the row_factory argument of the connect() and the cursor() method, which will control what type of record is returned by the fetch methods of the cursors and annotate the returned objects accordingly. See Row factories for more details.

dconn = psycopg.connect(row_factory=dict_row)
# dconn type is psycopg.Connection[Dict[str, Any]]

dcur = conn.cursor(row_factory=dict_row)
dcur = dconn.cursor()
# dcur type is psycopg.Cursor[Dict[str, Any]] in both cases

drec = dcur.fetchone()
# drec type is Optional[Dict[str, Any]]

Example: returning records as Pydantic models #

Using Pydantic it is possible to enforce static typing at runtime. Using a Pydantic model factory the code can be checked statically using Mypy and querying the database will raise an exception if the rows returned is not compatible with the model.

The following example can be checked with mypy --strict without reporting any issue. Pydantic will also raise a runtime error in case the Person is used with a query that returns incompatible data.

from datetime import date
from typing import Optional

import psycopg
from psycopg.rows import class_row
from pydantic import BaseModel

class Person(BaseModel):
    id: int
    first_name: str
    last_name: str
    dob: Optional[date]

def fetch_person(id: int) -> Person:
    with psycopg.connect() as conn:
        with conn.cursor(row_factory=class_row(Person)) as cur:
            cur.execute(
                """
                SELECT id, first_name, last_name, dob
                FROM (VALUES
                    (1, 'John', 'Doe', '2000-01-01'::date),
                    (2, 'Jane', 'White', NULL)
                ) AS data (id, first_name, last_name, dob)
                WHERE id = %(id)s;
                """,
                {"id": id},
            )
            obj = cur.fetchone()

            # reveal_type(obj) would return 'Optional[Person]' here

            if not obj:
                raise KeyError(f"person {id} not found")

            # reveal_type(obj) would return 'Person' here

            return obj

for id in [1, 2]:
    p = fetch_person(id)
    if p.dob:
        print(f"{p.first_name} was born in {p.dob.year}")
    else:
        print(f"Who knows when {p.first_name} was born")

Checking literal strings in queries #

The execute() method and similar should only receive a literal string as input, according to PEP 675 . This means that the query should come from a literal string in your code, not from an arbitrary string expression.

For instance, passing an argument to the query should be done via the second argument to execute() , not by string composition:

def get_record(conn: psycopg.Connection[Any], id: int) -> Any:
    cur = conn.execute("SELECT * FROM my_table WHERE id = %s" % id)  # BAD!
    return cur.fetchone()

# the function should be implemented as:

def get_record(conn: psycopg.Connection[Any], id: int) -> Any:
    cur = conn.execute("select * FROM my_table WHERE id = %s", (id,))
    return cur.fetchone()

If you are composing a query dynamically you should use the sql.SQL object and similar to escape safely table and field names. The parameter of the SQL() object should be a literal string:

def count_records(conn: psycopg.Connection[Any], table: str) -> int:
    query = "SELECT count(*) FROM %s" % table  # BAD!
    return conn.execute(query).fetchone()[0]

# the function should be implemented as:

def count_records(conn: psycopg.Connection[Any], table: str) -> int:
    query = sql.SQL("SELECT count(*) FROM {}").format(sql.Identifier(table))
    return conn.execute(query).fetchone()[0]

At the time of writing, no Python static analyzer implements this check ( mypy doesn’t implement it , Pyre does, but doesn’t work with psycopg yet ). Once the type checkers support will be complete, the above bad statements should be reported as errors.

Creating new row factories #

A row factory is a callable that accepts a Cursor object and returns another callable, a row maker , which takes raw data (as a sequence of values) and returns the desired object.

The role of the row factory is to inspect a query result (it is called after a query is executed and properties such as description and pgresult are available on the cursor) and to prepare a callable which is efficient to call repeatedly (because, for instance, the names of the columns are extracted, sanitised, and stored in local variables).

Formally, these objects are represented by the RowFactory and RowMaker protocols.

RowFactory objects can be implemented as a class, for instance:

from typing import Any, Sequence
from psycopg import Cursor

class DictRowFactory:
    def __init__(self, cursor: Cursor[Any]):
        self.fields = [c.name for c in cursor.description]

    def __call__(self, values: Sequence[Any]) -> dict[str, Any]:
        return dict(zip(self.fields, values))

or as a plain function:

def dict_row_factory(cursor: Cursor[Any]) -> RowMaker[dict[str, Any]]:
    fields = [c.name for c in cursor.description]

    def make_row(values: Sequence[Any]) -> dict[str, Any]:
        return dict(zip(fields, values))

    return make_row

These can then be used by specifying a row_factory argument in Connection.connect() , Connection.cursor() , or by setting the Connection.row_factory attribute.

conn = psycopg.connect(row_factory=DictRowFactory)
cur = conn.execute("SELECT first_name, last_name, age FROM persons")
person = cur.fetchone()
print(f"{person['first_name']} {person['last_name']}")

Pool life cycle #

A simple way to use the pool is to create a single instance of it, as a global object, and to use this object in the rest of the program, allowing other functions, modules, threads to use it:

# module db.py in your program
from psycopg_pool import ConnectionPool

pool = ConnectionPool(conninfo, **kwargs)
# the pool starts connecting immediately.

# in another module
from .db import pool

def my_function():
    with pool.connection() as conn:
        conn.execute(...)

Ideally you may want to call close() when the use of the pool is finished. Failing to call close() at the end of the program is not terribly bad: probably it will just result in some warnings printed on stderr. However, if you think that it’s sloppy, you could use the atexit module to have close() called at the end of the program.

If you want to avoid starting to connect to the database at import time, and want to wait for the application to be ready, you can create the pool using open=False , and call the open() and close() methods when the conditions are right. Certain frameworks provide callbacks triggered when the program is started and stopped (for instance FastAPI startup/shutdown events ): they are perfect to initiate and terminate the pool operations:

pool = ConnectionPool(conninfo, open=False, **kwargs)

@app.on_event("startup")
def open_pool():
    pool.open()

@app.on_event("shutdown")
def close_pool():
    pool.close()

Creating a single pool as a global variable is not the mandatory use: your program can create more than one pool, which might be useful to connect to more than one database, or to provide different types of connections, for instance to provide separate read/write and read-only connections. The pool also acts as a context manager and is open and closed, if necessary, on entering and exiting the context block:

from psycopg_pool import ConnectionPool

with ConnectionPool(conninfo, **kwargs) as pool:
    run_app(pool)

# the pool is now closed

When the pool is open, the pool’s background workers start creating the requested min_size connections, while the constructor (or the open() method) returns immediately. This allows the program some leeway to start before the target database is up and running. However, if your application is misconfigured, or the network is down, it means that the program will be able to start, but the threads requesting a connection will fail with a PoolTimeout only after the timeout on connection() is expired. If this behaviour is not desirable (and you prefer your program to crash hard and fast, if the surrounding conditions are not right, because something else will respawn it) you should call the wait() method after creating the pool, or call open(wait=True) : these methods will block until the pool is full, or will raise a PoolTimeout exception if the pool isn’t ready within the allocated time.

Connections life cycle #

The pool background workers create connections according to the parameters conninfo , kwargs , and connection_class passed to ConnectionPool constructor, invoking something like connection_class ( conninfo , ** kwargs ) . Once a connection is created it is also passed to the configure() callback, if provided, after which it is put in the pool (or passed to a client requesting it, if someone is already knocking at the door).

If a connection expires (it passes max_lifetime ), or is returned to the pool in broken state, or is found closed by check() ), then the pool will dispose of it and will start a new connection attempt in the background.

Using connections from the pool #

The pool can be used to request connections from multiple threads or concurrent tasks - it is hardly useful otherwise! If more connections than the ones available in the pool are requested, the requesting threads are queued and are served a connection as soon as one is available, either because another client has finished using it or because the pool is allowed to grow (when max_size > min_size ) and a new connection is ready.

The main way to use the pool is to obtain a connection using the connection() context, which returns a Connection or subclass:

with my_pool.connection() as conn:
    conn.execute("what you want")

The connection() context behaves like the Connection object context: at the end of the block, if there is a transaction open, it will be committed, or rolled back if the context is exited with as exception.

At the end of the block the connection is returned to the pool and shouldn’t be used anymore by the code which obtained it. If a reset() function is specified in the pool constructor, it is called on the connection before returning it to the pool. Note that the reset() function is called in a worker thread, so that the thread which used the connection can keep its execution without being slowed down by it.

Pool connection and sizing #

A pool can have a fixed size (specifying no max_size or max_size = min_size ) or a dynamic size (when max_size > min_size ). In both cases, as soon as the pool is created, it will try to acquire min_size connections in the background.

If an attempt to create a connection fails, a new attempt will be made soon after, using an exponential backoff to increase the time between attempts, until a maximum of reconnect_timeout is reached. When that happens, the pool will call the reconnect_failed() function, if provided to the pool, and just start a new connection attempt. You can use this function either to send alerts or to interrupt the program and allow the rest of your infrastructure to restart it.

If more than min_size connections are requested concurrently, new ones are created, up to max_size . Note that the connections are always created by the background workers, not by the thread asking for the connection: if a client requests a new connection, and a previous client terminates its job before the new connection is ready, the waiting client will be served the existing connection. This is especially useful in scenarios where the time to establish a connection dominates the time for which the connection is used (see this analysis , for instance).