异步操作

Psycopg Connection and Cursor have counterparts AsyncConnection and AsyncCursor supporting an asyncio interface.

The design of the asynchronous objects is pretty much the same of the sync ones: in order to use them you will only have to scatter the await keyword here and there.

async with await psycopg.AsyncConnection.connect(
        "dbname=test user=postgres") as aconn:
    async with aconn.cursor() as acur:
        await acur.execute(
            "INSERT INTO test (num, data) VALUES (%s, %s)",
            (100, "abc'def"))
        await acur.execute("SELECT * FROM test")
        await acur.fetchone()
        # will return (1, 100, "abc'def")
        async for record in acur:
            print(record)

Changed in version 3.1: AsyncConnection.connect() performs DNS name resolution in a non-blocking way.

Before version 3.1, AsyncConnection.connect() may still block on DNS name resolution. To avoid that you should set the hostaddr connection parameter, or use the resolve_hostaddr_async() to do it automatically.

On Windows, Psycopg is not compatible with the default ProactorEventLoop. Please use a different loop, for instance the SelectorEventLoop.

For instance, you can use, early in your program:

asyncio.set_event_loop_policy(
    asyncio.WindowsSelectorEventLoopPolicy()
)

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.

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)
    ...

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

The Diagnostic object received by the callback should not be used after the callback function terminates, because its data is deallocated after the callbacks have been processed. If you need to use the information later please extract the attributes requested and forward them instead of forwarding the whole Diagnostic object.

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().

You don’t need an AsyncConnection to handle notifications: a normal blocking Connection is perfectly valid.

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,)

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
        ...