feat: add support for psycopg

[enocom]

Psycopg does not support providing a pre-connected socket in its
connect API as there is no matching API in the underlying libpq C
interface. To work around this limitation, this commit creates an
in-memory proxy that copies data between a Unix domain socket and the
secure SSL Socket created by the Connector.

This approach works, but suffers from poor performance and makes the
driver comparable to pg8000 (a pure Python implementation).

The best solution would be for psycopg (and libpq) to expose a way to
provide a socket creator function as we have in the other drivers. Until
then, this is better than nothing.

Fixes #377.

[enocom]

Marking this as ready for review.

[rhatgadkar-goog]

Awesome job on implementing this!

[enocom]

Just to double check the resource usage, I ran this script to compare with pg8000. I created one Connector and then repeatedly created a pool and read a 1000 records out of my target instance, looking at memory usage, file descriptors, and threads.

import argparse
import os
import threading
import sqlalchemy
import time

import psutil

from google.cloud.alloydb.connector import Connector

parser = argparse.ArgumentParser()
parser.add_argument("--driver", choices=["pg8000", "psycopg"], default="psycopg")
args = parser.parse_args()

if args.driver == "pg8000":
    import pg8000  # noqa: F401
else:
    import psycopg  # noqa: F401

_proc = psutil.Process(os.getpid())

DRIVER_URL = {
    "pg8000": "postgresql+pg8000://",
    "psycopg": "postgresql+psycopg://",
}


# helper function to return SQLAlchemy connection pool
def init_connection_pool(connector: Connector) -> sqlalchemy.engine.Engine:
    # function used to generate database connection
    def getconn():
        conn = connector.connect(
            "<INSTANCE_ID>",
            args.driver,
            user="postgres",
            db="postgres",
            ip_type="PUBLIC",
            password="<PASSWORD>",
        )
        return conn

    # create connection pool
    pool = sqlalchemy.create_engine(
        DRIVER_URL[args.driver],
        creator=getconn,
    )
    return pool


print(f"Driver: {args.driver}")
print(f"Start Active threads: {threading.active_count()}")
print(f"Start FDs: {_proc.num_fds()}")
print(f"Start RSS: {_proc.memory_info().rss / 1024 / 1024:.1f} MB")
start_time = time.time()
with Connector() as connector:
    for i in range(500):
        engine = init_connection_pool(connector)
        with engine.connect() as db_conn:
            db_conn.execute(sqlalchemy.text("SELECT * from my_random_data")).fetchall()
        engine.dispose()
        if (i + 1) % 10 == 0:
            print(f"i={i + 1} RSS: {_proc.memory_info().rss / 1024 / 1024:.1f} MB")

end_time = time.time()
elapsed_time = end_time - start_time
print(f"Elapsed time: {elapsed_time}")
print(f"Finish Active threads: {threading.active_count()}")
print(f"Finish FDs: {_proc.num_fds()}")
print(f"Finish RSS: {_proc.memory_info().rss / 1024 / 1024:.1f} MB")

Observations:

• Thread usage is flat (starts with 1, ends with 2 for both drivers)
• File descriptor usage is consistent across drivers (starts at 3, ends with 10)
• Memory usage is comparable across drivers and is flat after 100 iterations

So I think we're good to merge this.