The log output from INFO level when using the connection pool is quite verbose. Each query handled by the pool generates 3 log entries, that gets rather large with even a modest amount of query activity. Would these 3 messages be more appropriate for the DEBUG level?

2021-03-29 17:50:40,458 INFO psycopg3.pool.pool Thread-2 : connection requested from 'pool-1'
2021-03-29 17:50:40,458 INFO psycopg3.pool.pool Thread-2 : connection given by 'pool-1'
2021-03-29 17:50:40,588 INFO psycopg3.pool.pool Thread-2 : returning connection to 'pool-1'

Version tested

psycopg3 @ git+https://github.com/psycopg/psycopg3.git@b270a0fe2f66c870e7eb2e1baa79b65330c3d752#subdirectory=psycopg3

await AsyncConnection.connect("host=example.com") blocks on the name resolution.

If psycopg has access to an async name resolution, the conninfo string can be rewritten adding "hostaddr", which would result in non blocking.

I've made a couple of tests:

• using the dnspython library it works, but the solution is clunky because the /etc/hosts file is not taken into account. See _dns.py in the async-dns-dnspython branch. hosts is not currently taken into accout except for the localhost special case, hardcoded to 127.0.0.1, which I'm not 100% sure is a good idea.
• calling gethostbyname() in a separate thread works for me, but the workflow fails on Python 3.6 and 3.7 with a resource leak. Curiously, exactly the same test passes locally (using python:3.7.11 in Docker on Linux - instruction to reproduce available in the tests readme(.

So I'm tempted to leave the feature for later, and only commit the basic framework (merging only the connection subclassing hook)

Such backend is a pretty important real world test case. The idea is to be able to run the entire Django test suite.

In progress at https://github.com/dvarrazzo/django/tree/psycopg3

According to the documentation here, windows OS is supported. But it fail to install.

System Information:

• OS : Windows 10 21H1 19043.1237
• Python : 3.10.0
• Pip : 21.3 from C:\Users\Yandi Banyu Karima W\AppData\Roaming\Python\Python310\site-packages\pip (python 3.10)
• Postgresql : 13

Steps to reproduce:

1. Open powershell
2. run pip install --pre psycopg[binary]

Output:

Collecting psycopg[binary]
  Using cached psycopg-3.0b1-py3-none-any.whl (131 kB)
ERROR: Could not find a version that satisfies the requirement psycopg-binary==3.0.beta1; extra == "binary" (from psycopg[binary]) (from versions: none)
ERROR: No matching distribution found for psycopg-binary==3.0.beta1; extra == "binary"

Is it planed for this package work in gevent based projects or will it be totally async-await based like asyncpg?

Good day!

When importing the psycopg3 the following error occurs:

  File "d:\****\main.py", line 2, in <module>
    import psycopg3
  File "D:\Python\lib\site-packages\psycopg3\__init__.py", line 9, in <module>
    from . import pq
  File "D:\Python\lib\site-packages\psycopg3\pq\__init__.py", line 103, in <module>
    import_from_libpq()
  File "D:\Python\lib\site-packages\psycopg3\pq\__init__.py", line 95, in import_from_libpq
    raise ImportError(
ImportError: no pq wrapper available.
Attempts made:
- error importing 'c' wrapper: No module named 'psycopg3_c'
- error importing 'binary' wrapper: No module named 'psycopg3_binary'
- error importing 'python' wrapper: libpq library not found

The problem occurs on Windows 10 with PostreSQL 13 installed. According to the plan, libpq should be present...

According to https://pypi.org/project/psycopg2-binary/ latest version is 2.8.6 however there is no here any verion tags.
Is it possible to add at least last version tag and keep adding those tags on each new release?

Draft design at https://www.psycopg.org/articles/2021/01/17/pool-design/

Branch: https://github.com/psycopg/psycopg3/tree/connection-pool

Currently missing:

• async interface
• documentation
• dynamic change of minconn/maxconn
• implement max_lifetime
• implement get_info()
• add connection cleanup hook
• if a connection is returned broken, empty the pool and validate all connections added pool.check() to for a cleanup
• periodically check state of connections in the pool better done with a probe and pool.check()
• maybe active probe to detect disconnection? Not needed, can use something like this technique

Is there any secure connection with server. I can find any documentation for it.

So, can anybody help me ?

Ciao Daniele,

I noticed the below error when testing psycopg3. It worked in psycopg2. The database I'm connecting to is CockroachDB.

fabio@mac: ~/projects/payments $ /usr/local/bin/python3
Python 3.8.5 (default, Jul 21 2020, 10:48:26) 
[Clang 11.0.3 (clang-1103.0.32.62)] on darwin
Type "help", "copyright", "credits" or "license" for more information.
>>> import psycopg
>>> psycopg.__version__
'3.0.beta1'
>>> dburl = 'postgresql://root@localhost:26257/payments?sslmode=disable'
>>> 
>>> conn = psycopg.connect(dburl)
>>> conn
<psycopg.Connection [IDLE] (host=localhost port=26257 user=root database=payments) at 0x10e131d90>
>>> 
>>> conn2 = psycopg.connect(dburl)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/Users/fabio/Library/Python/3.8/lib/python/site-packages/psycopg/connection.py", line 578, in connect
    rv = cls._wait_conn(
  File "/Users/fabio/Library/Python/3.8/lib/python/site-packages/psycopg/connection.py", line 783, in _wait_conn
    return waiting.wait_conn(gen, timeout=timeout)
  File "/Users/fabio/Library/Python/3.8/lib/python/site-packages/psycopg/waiting.py", line 94, in wait_conn
    fileno, s = gen.send(ready)
  File "/Users/fabio/Library/Python/3.8/lib/python/site-packages/psycopg/connection.py", line 414, in _connect_gen
    pgconn = yield from connect(conninfo)
  File "psycopg_binary/_psycopg/generators.pyx", line 48, in connect
psycopg.OperationalError: connection failed: Connection refused
could not send startup packet: Connection refused
>>> 
>>> #######################################################################################
>>> 
>>> import psycopg2
>>> psycopg2.__version__
'2.8.6 (dt dec pq3 ext lo64)'
>>> conn = psycopg2.connect(dburl)
>>> conn2 = psycopg2.connect(dburl)
>>> 
>>> conn
<connection object at 0x10aeebd60; dsn: 'postgresql://root@localhost:26257/payments?sslmode=disable', closed: 0>
>>> conn2
<connection object at 0x10aeebeb0; dsn: 'postgresql://root@localhost:26257/payments?sslmode=disable', closed: 0>
>>> 

From sqlalchemy/sqlalchemy#6842 (comment)

I've a question, more like an use case, maybe it's better as an issue in psycopg tracker.

In the dialect I wanted to register hstore and user provided json dumps/loads globally when requiresed provided, but I think it's not advisable, since it causes issues both in the tests suite that necessarily try different combinations, and because doing so would mean that an user cannot have two engines with different configurations. So I think I'll move the registration per-connection, like in the other dialects.

Looking at how the registration happens in psycopg, my question/use case was: would it be possible for psycopg to have a "Context" class concept that could be used to register these adapters. Then in the connect call I could pass the context and the connection inherits from it? The idea being that each dialect (so basically each engine) would have its own context, I would do the registration only once, not on each connection. Something like this:

ctx = psycopg.context()
psycopg.types.json.set_json_loads(my_loads, ctx)
psycopg.types.hstore.register_hstore(hstore_info, ctx)
# other register stuff with ctx
conn = psycopg.connect(..., context=ctx)

Note that none of this is blocking or even required, just an use case that I though of while working on the dialect, and maybe also useful for direct users of psycopg, like a library that uses it could have its own customization in a context, avoiding polluting the global state.

@dvarrazzo has already provided an implementation at https://github.com/psycopg/psycopg/compare/connection-custom-context

I was hoping there might be an example how to create an AsyncPool with Dict_Row as the row parser + calling fetchone() / fetchall() directly on the connection..... something like

with config.pool.connection() as conn:
  ostream = await conn.execute("QUERY").fetchall()

or

with config.pool.connection(row_parser=dict_row) as conn:
  ostream = await conn.execute("QUERY").fetchall()

To side track for a second, this request derives from the following problem:

When using CockroachDB, an increasingly popular DB provider grounded in serializable transactions, psycopg3 invokes transactions on the database on all cursors.....iIn async mode, for a machine w/ a small # of cores, any print() statements will cause many async functions to stop progress

Given the time it takes to print to the console and then resume coroutines, we end up extending beyond the database serializable transaction timeout limit causing an error

I'm hoping to overcome this problem by having Psycopg3 execute a SELECT / INSERT operation without a cursor akin to sqlx / asyncpg. Unfortunately, I don't think this is possible.

However, at the least, it would be great to have a shorter API where one can call execute() + fetchall() returning a list of dict_rows on an AsyncPool interface

--pre is required for beta versions.

There are now dumper/loader to adapt Python Decimal to Postgres numeric in binary format, implemented both in C and in Cython.

These methods however are about the 30% slower than the text format adapters (at least the dumper). The problem is that, even in C, the binary dumper uses the Decimal.as_tuple() method, which is relatively slow.

An API to access the Decimal C values should be available in Python 3.10: https://bugs.python.org/issue41324 The C adapters should use it.

There is a draft of a branch where I was thinking to use the mpdec directly, but this would complicate the build at it requires the libmpdec-dev package. So it's probably a bad idea.

Implement RFC 2782 lookup for servers to use a DNS to resolve SRV records and ports.

Resolution might start if the host name is in the form _Service._Proto.Name. Systems such as Consul also allow more lax rules to define services. Maybe, if we wanted to cover them too, we could allow the port to be defined as SRV.

The right place to do it is in Connection._get_connection_params(), which has both sync and async versions. The method allows to manipulate the conninfo params: if RFC2782-looking hosts are found, or a SRV port is found, replace these items with the list of items found in the SRV record, then let the libpq to weed out the servers that don't work. For async connections, also resolve the host list into hostaddr (see #69 for some partial work).

Add file-like interface to large objects

Unlike psycopg2, don't use the libpq functions; can simply use the server-side functions that the libpq calls, which would allow to give it an async interface too.

Hello, this is more a background question than an actual issue - apologies if this isn't the right place for this sort of thing.

In psycopg2, if I select a large number of rows, psycopg2 will internally buffer them prior to my being able to consume them (i.e. by iterating over the output cursor). This has caused some major problems when selecting back many millions of rows. The workaround I've found is to create a named server-side cursor and tune down the itersize parameter, but this seems to have some negative implications from a performance perspective.

So my question is, can we expect improvements in this area in psycopg3? Is the current implementation necessary due to some Postgres wire protocol limitation, or is this something that can potentially be made better?

Thanks for any insights you can provide here, and for your work on this project. I'm excited to see what you do with it, regardless of whether or not this scenario happens to be solved.

I was leveraging the cur.query and cur.params attribute as a sanity check. I was doing some digging and noticed the changes to the Cursor API. I'm curious why NOT provide this information? I would love to be able to leverage this information as a sanity check to make sure the correct statement is being generated and being sent down the pipe.

I may be missing something obvious here but I couldn't find the equivalent as it is. I shifted to use the cur.statusmessage but it doesn't give me the level of granularity that I would like to see.

Add support for streaming replication as in psycopg2

Using the awesome new Copy.rows method works great while the COPY operation is running, but appears to leave the connection in an unclean state. Attempting to run another query on the connection, or exit the context manager for the connection, results in "sending query failed: another command is already in progress".

$ cat repro.py
from psycopg3.oids import builtins
import psycopg3

with psycopg3.connect() as conn:
    with conn.cursor() as cur:
        with cur.copy("COPY (VALUES (1), (2), (3)) TO STDOUT (FORMAT BINARY)") as copy:
            copy.set_types([builtins["int4"].oid])
            for row in copy.rows():
                print(row)

$ python repro.py 
(1,)
(2,)
(3,)
Traceback (most recent call last):
  File "repro.py", line 9, in <module>
    print(row)
  File "/home/benesch/Sites/psycopg3-play/venv/lib/python3.8/site-packages/psycopg3/connection.py", line 435, in __exit__
    self.commit()
  File "/home/benesch/Sites/psycopg3-play/venv/lib/python3.8/site-packages/psycopg3/connection.py", line 465, in commit
    self.wait(self._commit_gen())
  File "/home/benesch/Sites/psycopg3-play/venv/lib/python3.8/site-packages/psycopg3/connection.py", line 510, in wait
    return waiting.wait(gen, self.pgconn.socket, timeout=timeout)
  File "/home/benesch/Sites/psycopg3-play/venv/lib/python3.8/site-packages/psycopg3/waiting.py", line 223, in wait_epoll
    s = next(gen)
  File "/home/benesch/Sites/psycopg3-play/venv/lib/python3.8/site-packages/psycopg3/connection.py", line 380, in _commit_gen
    yield from self._exec_command(b"commit")
  File "/home/benesch/Sites/psycopg3-play/venv/lib/python3.8/site-packages/psycopg3/connection.py", line 346, in _exec_command
    self.pgconn.send_query(command)
  File "/home/benesch/Sites/psycopg3-play/venv/lib/python3.8/site-packages/psycopg3/pq/pq_ctypes.py", line 242, in send_query
    raise PQerror(f"sending query failed: {error_message(self)}")
psycopg3.pq.PQerror: sending query failed: another command is already in progress

The docs says Psycopg3 can run from Python 3.6 to 3.10 :

https://www.psycopg.org/psycopg3/docs/basic/install.html#supported-systems

Python: from version 3.6 to 3.10

But installing on Python 3.8 fails :

tests.txt :

pytest
psycopg[binary]

$ pip3 install --user -r requirements/tests.txt
Requirement already satisfied: pytest in /usr/local/lib/python3.8/dist-packages (from -r requirements/tests.txt (line 1)) (6.2.5)
Collecting psycopg[binary]
  Downloading psycopg-3.0b1-py3-none-any.whl (131 kB)
Requirement already satisfied: toml in /usr/local/lib/python3.8/dist-packages (from pytest->-r requirements/tests.txt (line 1)) (0.10.2)
Requirement already satisfied: iniconfig in /usr/local/lib/python3.8/dist-packages (from pytest->-r requirements/tests.txt (line 1)) (1.1.1)
Requirement already satisfied: packaging in /usr/local/lib/python3.8/dist-packages (from pytest->-r requirements/tests.txt (line 1)) (21.0)
Requirement already satisfied: py>=1.8.2 in /usr/local/lib/python3.8/dist-packages (from pytest->-r requirements/tests.txt (line 1)) (1.10.0)
Requirement already satisfied: attrs>=19.2.0 in /usr/local/lib/python3.8/dist-packages (from pytest->-r requirements/tests.txt (line 1)) (21.2.0)
Requirement already satisfied: pluggy<2.0,>=0.12 in /usr/local/lib/python3.8/dist-packages (from pytest->-r requirements/tests.txt (line 1)) (1.0.0)
Collecting backports.zoneinfo; python_version < "3.9"
  Downloading backports.zoneinfo-0.2.1-cp38-cp38-manylinux1_x86_64.whl (74 kB)
ERROR: Could not find a version that satisfies the requirement psycopg-binary==3.0.beta1; extra == "binary" (from psycopg[binary]->-r requirements/tests.txt (line 2)) (from versions: none)
ERROR: No matching distribution found for psycopg-binary==3.0.beta1; extra == "binary" (from psycopg[binary]->-r requirements/tests.txt (line 2))
Cleaning up file based variables 00:03
ERROR: Job failed: exit code 1

While on Python 3.9, it works well:

$ pip3 install --user -r requirements/tests.txt
Collecting pytest
  Downloading pytest-6.2.5-py3-none-any.whl (280 kB)
Collecting psycopg[binary]
  Downloading psycopg-3.0b1-py3-none-any.whl (131 kB)
...
Collecting psycopg-binary==3.0.beta1
  Downloading psycopg_binary-3.0b1-cp39-cp39-manylinux_2_24_x86_64.whl (5.9 MB)
Installing collected packages: pyparsing, toml, py, psycopg-binary, psycopg, pluggy, packaging, iniconfig, attrs, pytest

Am I missing something or is the doc up to date about the support of Python < 3.9 ?

I know sponsoring would be great for you, but I don’t have much money. Are there any major OR minor things people can help with? Features, potential bugs, TODOs/FIX
MEs? Or would it break your workflow or mindset of how psycopg3 should be designed?

I think many people would agree that major improvements and milestones should be developed by you based on your blog posts and that aforementioned mindset thing, but others would definitely love to help in some way or another! (e.g. I’m mostly interested in making Django’s ORM async capable, so the async portion would be of interest to me).

I just ran into this problem.

For consistency with execute(), shouldn't executemany() provide the same control over preparation?

On Windows there are the following issues:

• Async tests don't pass. Should be able to run the test suite without these skips.
• The C modules compile ok but fail import. It should be possible to run the tests on GH Action.
• The binary packages fail to build. It is necessary to install the libpq in the cibuildwheel windows image. This workflow must be enabled.

openGauss is an open source relational database management system that is released with the Mulan PSL v2. with the kernel derived from PostgreSQL.

If you can support opengauss databse, it will be great for python driver. How do you think about?

See conversation with @asqui at 5739957

check if the Copy object needs the same treatment.

I think the new stream stuff might not work correctly with cancel. Here's some example code using a version of materialize that correctly supports cancellation. I wasn't able to get a repro under postgres using SELECT, but I'm medium confident this is a psycopg3 stream related bug (my test using COPY with cancel and psycopg3 works fine).

Or I'm doing something totally wrong. Not sure.

import psycopg3

dsn = "postgresql://materialize@localhost:6875/materialize?sslmode=disable"

conn = psycopg3.connect(dsn)
with conn.cursor() as cur:
    conn.autocommit = True
    cur.execute("DROP TABLE IF EXISTS t")
    cur.execute("CREATE TABLE t (a int)")
    cur.execute("INSERT INTO t VALUES (1), (2)")
    conn.autocommit = True
    stream = cur.stream("TAIL t")
    print(next(stream))
    print(next(stream))
    conn.cancel()
    try:
        print(next(stream))
    except Exception as e:
        print("exception:", e)
    print("DONE")
    print(cur.execute("select 1"))

requirements.txt:

git+https://github.com/psycopg/psycopg3.git@4f053f14901454a7c7895b527ddeb9d8ab57d7fe#subdirectory=psycopg3

output is:

(Decimal('1612558651418'), 1, 1)
(Decimal('1612558651418'), 1, 2)
exception: canceling statement due to user request
DONE
Traceback (most recent call last):
  File "m.py", line 26, in <module>
    print(cur.execute("select 1"))
  File "/usr/local/lib/python3.7/site-packages/psycopg3/cursor.py", line 469, in execute
    self._conn.wait(self._execute_gen(query, params, prepare=prepare))
  File "/usr/local/lib/python3.7/site-packages/psycopg3/connection.py", line 519, in wait
    return waiting.wait(gen, self.pgconn.socket, timeout=timeout)
  File "/usr/local/lib/python3.7/site-packages/psycopg3/waiting.py", line 223, in wait_epoll
    s = next(gen)
  File "/usr/local/lib/python3.7/site-packages/psycopg3/cursor.py", line 192, in _execute_gen
    yield from self._maybe_prepare_gen(pgq, prepare)
  File "/usr/local/lib/python3.7/site-packages/psycopg3/cursor.py", line 224, in _maybe_prepare_gen
    self._execute_send(pgq)
  File "/usr/local/lib/python3.7/site-packages/psycopg3/cursor.py", line 332, in _execute_send
    self._conn.pgconn.send_query(query.query)
  File "/usr/local/lib/python3.7/site-packages/psycopg3/pq/pq_ctypes.py", line 242, in send_query
    raise PQerror(f"sending query failed: {error_message(self)}")
psycopg3.pq.PQerror: sending query failed: another command is already in progress

When using the binary version, ie psycopg[binary] calls into libpq,
the following initial setup (with *** being user / password )

await AsyncConnectionPool("postgresql://*****:******@localhost:26257/passport", max_size=50).wait()

causes a

WARNING:psycopg.pool:error connecting in 'pool-1': connection failed: Connection refused
could not send SSL negotiation packet: Connection refused

I use this url in other libs (java / rust sqlx). Is there a different url it's expecting? Typically, SSL takes care of itself.

In psycopg2 connections are accumulated during the normal querying process into a list on the connection (the notifies attribute), which has to be emptied periodically. If someone wants to read notifications explicitly (which is the most common use case, as you have to LISTEN explicitly in order to receive them) then you can use conn.poll() in a loop to receive notifications without running additional queries, and you can block and wait for readiness on the connection fd to avoid the loop to burn cpu uselessly and still receive notifications in a timely fashion. An example of usage is in the docs.

The shortcomings are the use of a list as a poor man's queue (although from psycopg2 2.7 the attribute is writable and it can be replaced by a better object) and that implementing a push loop requires dabbling with low level I/O multiplexing primitives such as select.

As of dc378b8 there are two ways to manage async notifications in psycopg3:

1: read them explicitly from a dedicated connection using the connection.notifies() generator, which will do the wait internally "the best possible way" (i.e. consistently with how the connection waits in the normal query process): notifications can be read just by a for n in conn.notifies() which can be left in a thread/greenlet/process of its own or block the main loop if used in a microservice that does only this - e.g. reading notifies from postgres and push them into rabbitmq... The generator can be stopped by send()ing it a True value and is an async generator for asyncio-based connections (i.e. it can be consumed using async for n in conn.notifies())

2: as a callback to process notifies generated during the normal querying process: you can conn.add_notify_handler(queue.put_nowait) and receive notifies there as the cursor.execute() returns a value.

You can see samples of both the uses in the tests committed in the changeset above.

I am not entirely convinced about this design. I don't think 2. is badly needed but I wouldn't know how to process notifies in-sync with the queries otherwise. I don't see the latter a very common use case anyway but I don't want to preclude it altogether.

What do you think about the matter? Comments are welcome. Cheers!

In this mode, several cursor may send commands to the same connection and then sit in a queue to receive the result.

https://www.postgresql.org/docs/14/libpq-pipeline-mode.html

Installed successfully:

pypy3 -mpip install git+https://github.com/psycopg/psycopg3.git#subdirectory=psycopg3
Defaulting to user installation because normal site-packages is not writeable
Collecting git+https://github.com/psycopg/psycopg3.git#subdirectory=psycopg3
  Cloning https://github.com/psycopg/psycopg3.git to /tmp/pip-req-build-38w058qk
  Installing build dependencies ... done
  Getting requirements to build wheel ... done
    Preparing wheel metadata ... done
Requirement already satisfied: typing-extensions in /root/.local/lib/pypy3.6/site-packages (from psycopg3==2.99.0) (3.7.4.3)

But the import fails:

Traceback (most recent call last):
  File "app.py", line 6, in <module>
    import psycopg3
  File "/root/.local/lib/pypy3.6/site-packages/psycopg3/__init__.py", line 7, in <module>
    from . import pq
  File "/root/.local/lib/pypy3.6/site-packages/psycopg3/pq/__init__.py", line 102, in <module>
    import_from_libpq()
  File "/root/.local/lib/pypy3.6/site-packages/psycopg3/pq/__init__.py", line 99, in import_from_libpq
    raise ImportError("no pq wrapper available")
ImportError: no pq wrapper available

Any ideas?

See here for reference.

Client-side statement can be merged using sql.SQL(). However we could create a cursor doing that automatically.

It would be handy in porting psycopg2 code more easily maybe, even if less efficient.

It could be made a mixin to add to ClientCursor and AsyncClientCursor, replacing the PostgresQuery with a PostgresClientQuery doing a different transformation.

Hello,
I wanted to try psycopg3 and found it seems not to work in macOS.

This is the error when importing psycopg3:

Traceback (most recent call last):
  File "/Users/jacopo/projects/pgtest/src/ingest_into_postgres.py", line 1, in <module>
    import psycopg3
  File "/Users/jacopo/projects/pgtest/.venv/lib/python3.9/site-packages/psycopg3/__init__.py", line 9, in <module>
    from .copy import Copy, AsyncCopy
  File "/Users/jacopo/projects/pgtest.venv/lib/python3.9/site-packages/psycopg3/copy.py", line 22, in <module>
    from .generators import copy_from, copy_to, copy_end
  File "/Users/jacopo/projects/pgtest.venv/lib/python3.9/site-packages/psycopg3/generators.py", line 25, in <module>
    from .waiting import Wait, Ready
  File "/Users/jacopo/projects/pgtest/.venv/lib/python3.9/site-packages/psycopg3/waiting.py", line 200, in <module>
    Wait.R: select.EPOLLONESHOT | select.EPOLLIN,
AttributeError: module 'select' has no attribute 'EPOLLONESHOT'

I'm using Python 3.9 under macOS 11.2, and installing it from the repo

From https://www.psycopg.org/psycopg3/docs/async.html it looks like the async support is tightly coupled to asyncio. Would you be willing to use anyio so that Trio users could also use psycopg3 async?

Thanks for your consideration and for your work on psycopg3!

I'm trying to understand how I can start logical replication from the beginning. I have this all setup...but I'm only capturing new changes to the tables but not the current data thats in there. I'm assuming I need to specify the LSN to start at (doc: https://www.psycopg.org/docs/extras.html?highlight=loggingconnection#psycopg2.extras.ReplicationMessage.data_start)...but how does one get the start of LSN?

by basic python script looks like this:

import json
import random
import calendar
import time
from datetime import datetime
import psycopg2
from psycopg2.extras import LogicalReplicationConnection

my_connection  = psycopg2.connect(
                   "dbname='dbname' host='host.com' user='user' password='pass'" ,
                   connection_factory = LogicalReplicationConnection)
cur = my_connection.cursor()
cur.drop_replication_slot('wal2json_test_slot')
cur.create_replication_slot('wal2json_test_slot', output_plugin = 'wal2json')
cur.start_replication(slot_name = 'wal2json_test_slot', options = {'pretty-print' : 1}, decode= True)

def consume(msg):
    print (msg.payload)

cur.consume_stream(consume)

Add two-phase commit interface as per DBAPI/psycopg2

Hey, @dvarrazzo! First of all, thanks for your amazing work for the open source :)

I've recently discovered this brand new psycopg3 repo and decided to participate a little bit with django backend driver #18.

The first failed test I've tried to fix was model_fields.test_uuid.TestQuerying.test_filter_with_expr. It currently fails with following message:

django.db.utils.ProgrammingError: could not determine data type of parameter $1

The corresponding sql query from debug log is:

SELECT 
   "model_fields_nullableuuidmodel"."id",
   "model_fields_nullableuuidmodel"."field", 
   CONCAT('8400', 'e29b') AS "value" 
FROM "model_fields_nullableuuidmodel" 
WHERE "model_fields_nullableuuidmodel"."field"::text LIKE '%%' 
   || REPLACE(REPLACE(REPLACE(CONCAT('8400', 'e29b'), E'\\', E'\\\\'), E'%%', E'\\%%'), E'_', E'\\_')
   || '%%'; args=('8400', 'e29b', '8400', 'e29b')

But where is $1? Well, psycopg3 is internally converting all python like placeholders %s to postgres like $1,$2. Code can be found in psycopg3.psycopg3._queries._query2pg function. So if we manually print converted query from there it'll be:

SELECT 
  "model_fields_nullableuuidmodel"."id", 
  "model_fields_nullableuuidmodel"."field", 
  CONCAT($1, $2) AS "value" 
FROM "model_fields_nullableuuidmodel" 
WHERE "model_fields_nullableuuidmodel"."field"::text LIKE '%' 
   || REPLACE(REPLACE(REPLACE(CONCAT($3, $4), E'\\', E'\\\\'), E'%', E'\\%'), E'_', E'\\_') 
   || '%'

And that's the place where the problem lies! psycopg3 uses extended-query protocol and thus requires more comprehensive type usage. Small example to understand difference between query and extended-query modes:

import psycopg2
import psycopg3

QUERIES = (
    "SELECT REPLACE(CONCAT(%s, %s), 'foo', 'bar')",
    "SELECT REPLACE(CONCAT(%s::text, %s::text), 'foo', 'bar')"
)
PARAMS = ('foo', 'bar')

if __name__ == '__main__':
    for query in QUERIES:
        for psycopg in (psycopg2, psycopg3):
            conn = psycopg.connect("dbname=postgres user=postgres")
            cur = conn.cursor()
            try:
                cur.execute(query, PARAMS)
            except Exception as e:
                print(f'{psycopg.__name__} query {query!r} failed with exception: {e!r}')
            else:
                res = cur.fetchall()
                print(f'{psycopg.__name__} query {query!r} result is {res!r}')
            cur.close()

psycopg2 query "SELECT REPLACE(CONCAT(%s, %s), 'foo', 'bar')" result is [('barbar',)]
psycopg3 query "SELECT REPLACE(CONCAT(%s, %s), 'foo', 'bar')" failed with exception: IndeterminateDatatype('could not determine data type of parameter $1')
psycopg2 query "SELECT REPLACE(CONCAT(%s::text, %s::text), 'foo', 'bar')" result is [('barbar',)]
psycopg3 query "SELECT REPLACE(CONCAT(%s::text, %s::text), 'foo', 'bar')" result is [('barbar',)]

Several questions here:

1. Shall we log transformed queries somehow? It would have saved me a lot of time in this case.
2. What is desirable strategy for adding ::text like type casts here? Is it user's responsibility, or _query2pg has to handle it?

  File "/Users/anhtran/PyCharmProjects/project/venv/lib/python3.8/site-packages/django/contrib/postgres/apps.py", line 1, in <module>
    from psycopg2.extras import (
  File "/Users/anhtran/PyCharmProjects/project/venv/lib/python3.8/site-packages/psycopg2/__init__.py", line 51, in <module>
    from psycopg2._psycopg import (                     # noqa
ImportError: dlopen(/Users/anhtran/PyCharmProjects/project/venv/lib/python3.8/site-packages/psycopg2/_psycopg.cpython-38-darwin.so, 2): Symbol not found: ____chkstk_darwin
  Referenced from: /Users/anhtran/PyCharmProjects/project/venv/lib/python3.8/site-packages/psycopg2/.dylibs/libcrypto.1.1.dylib (which was built for Mac OS X 10.15)
  Expected in: /usr/lib/libSystem.B.dylib
 in /Users/anhtran/PyCharmProjects/project/venv/lib/python3.8/site-packages/psycopg2/.dylibs/libcrypto.1.1.dylib

I have to go back to 2.8.x to run the project in my old macbook.

Multirange objects introduced in PostgreSQL 14.

Many large companies view GPL and LGPL licenses as toxic and do not allow their usage. Would it be possible to consider releasing psycopg3 under a more corporate-friendly license like MIT, BSD, or Apache?

Setup is AsyncPoolConnection with CockroachDB Postgresql Database,
running

 async with config.pg_pool.connection() as conn:
   async with conn.cursor() as tx:
     async with tx.copy("COPY ACCOUNT (id, handle, avatar_id, metadata, status, privacy, indexer, timestamp, phone, name) FROM STDIN") as stream:
       for i in batch:
         await stream.write_row((
           i.id,
           i.handle,
           i.avatar_id,
           i.metadata,
           i.status,
           i.privacy,
           i.indexer,
           i.timestamp,
           i.phone,
           i.name
         ))

causes the errror

async with tx.copy("COPY ACCOUNT (id, handle, avatar_id, metadata, status, privacy, indexer, timestamp, phone, name) FROM STDIN") as stream:
  File "/usr/local/Cellar/[email protected]/3.9.6/Frameworks/Python.framework/Versions/3.9/lib/python3.9/contextlib.py", line 175, in __aenter__
    return await self.gen.__anext__()
  File "/Users/tsalemy/passport/florence/skin/passport/lib/python3.9/site-packages/psycopg/cursor_async.py", line 158, in copy
    await self._conn.wait(self._start_copy_gen(statement))
  File "/Users/tsalemy/passport/florence/skin/passport/lib/python3.9/site-packages/psycopg/connection_async.py", line 273, in wait
    return await waiting.wait_async(gen, self.pgconn.socket)
  File "/Users/tsalemy/passport/florence/skin/passport/lib/python3.9/site-packages/psycopg/waiting.py", line 144, in wait_async
    s = gen.send(ready)
  File "/Users/tsalemy/passport/florence/skin/passport/lib/python3.9/site-packages/psycopg/cursor.py", line 326, in _start_copy_gen
    self._check_copy_result(result)
  File "/Users/tsalemy/passport/florence/skin/passport/lib/python3.9/site-packages/psycopg/cursor.py", line 464, in _check_copy_result
    raise e.error_from_result(
psycopg.InternalError: CopyFrom not supported in extended protocol mode

I know that psycopg2 doesn't allow async COPY command - didn't know if this was similar issue or not

Hello!
In psycopg2 to insert LOB we should used function "lo_from_bytea(0, %s)" (wich gets only oid of LOB) and to read LOB we used "lo_get(oid)". Can we simpy send opened file stream to insert and select? It will be more frendly feauter in psycopg3 version. Thanks!

Windows asyncio tests don't pass and they are currently skipped.

Does async work on Windows at all?

Currently, I'm not really satisfied with how any Python SQL libraries handle parameterizing queries. I think it would be interesting to look into alternative syntaxes while psycopg3 is still in beta. I'm thinking something like this:

cur.exec('INSERT INTO students VALUES ({name}, {date})')

Everything in a bracket is pulled from the local scope and escaped. If you want to override a variable (for instance, you are using that variable name elsewhere, but would like to use a short name), you can pass it as a named argument to override anything in the local scope:

cur.exec('INSERT INTO students VALUES ({name}, {date})', name=student_name)

Depending on how comfortable people are magic, we could support f-string like functionality, sadly PEP501 is still deferred.

While on first glance I'm sure this looks like a SQL injection waiting to happen, I think this will actually limit SQL injections since it makes the correct approach the easiest.

The other idea that I would like to try is using typing.get_type_hints to try to find the correct types for the function parameters. Building off this, I think there could be a strict=True argument that raises an exception if the type cannot be identified, this could also potentially integrate with Mypy.

A few ideas for type overrides are as follows. The simplest option is just having the user pass a typing.TypedDict to set the types:

cur.exec('INSERT INTO students VALUES ({name}, {date})', name=student_name, types={'name': str})

I don't like this option since the type annotation isn't "local" which makes it easy to make a mistake. Another option is using typing.Annotated:

cur.exec('INSERT INTO students VALUES ({name}, {date})', name=Annotated[str, student_name])

This also isn't a great option, since it requires an import, but really there isn't any reason we couldn't support both. The third option would be to support a type annotation in the actual SQL expression:

cur.exec('INSERT INTO students VALUES ({name: str}, {date})')

I like this, but sadly IDEs might find it hard to autocomplete.

Of course, I am more than happy to make a PR for this, I just wanted to get some other opinions before deciding this is useful for other people as well and if it is a good fit for psycopg3.