feat: add support for psycopg

README.md

@@ -32,7 +32,11 @@ Python applications. It provides:
 
 [iam-db-authn]: https://cloud.google.com/alloydb/docs/manage-iam-authn
 
-**Supported drivers:** [`pg8000`](https://codeberg.org/tlocke/pg8000) (sync) · [`asyncpg`](https://magicstack.github.io/asyncpg) (async)
+**Supported drivers:** [`pg8000`][pg8000] (sync) · [`psycopg`][psycopg] (sync) · [`asyncpg`][asyncpg] (async)
+
+[pg8000]: https://codeberg.org/tlocke/pg8000
+[psycopg]: https://www.psycopg.org/
+[asyncpg]: https://magicstack.github.io/asyncpg
 
 ## Quickstart
 
@@ -67,6 +71,37 @@ with Connector() as connector:
         print(result)
 ```
 
+### Sync (psycopg + SQLAlchemy)
+
+**Install:**
+```sh
+pip install "google-cloud-alloydb-connector[psycopg]" sqlalchemy
+```
+
+**Connect:**
+```python
+import sqlalchemy
+from google.cloud.alloydbconnector import Connector
+
+INSTANCE_URI = "projects/MY_PROJECT/locations/MY_REGION/clusters/MY_CLUSTER/instances/MY_INSTANCE"
+
+with Connector() as connector:
+    pool = sqlalchemy.create_engine(
+        "postgresql+psycopg://",
+        creator=lambda: connector.connect(
+            INSTANCE_URI,
+            "psycopg",
+            user="my-user",
+            password="my-password",
+            db="my-db",
+        ),
+    )
+
+    with pool.connect() as conn:
+        result = conn.execute(sqlalchemy.text("SELECT NOW()")).fetchone()
+        print(result)
+```
+
 ### Async (asyncpg + SQLAlchemy)
 
 **Install:**
@@ -188,7 +223,7 @@ database user][add-iam-user].
 ```python
 connector.connect(
     INSTANCE_URI,
-    "pg8000",  # or "asyncpg"
+    "pg8000",  # or "psycopg" (sync) / "asyncpg" (async)
     user="service-account@my-project.iam",  # omit .gserviceaccount.com suffix
     db="my-db",
     enable_iam_auth=True,

google/cloud/alloydbconnector/client.py

@@ -97,7 +97,7 @@ def __init__(
         self._is_sync = False
         if client:
             self._client = client
-        elif driver == "pg8000":
+        elif driver == "pg8000" or driver == "psycopg":
             self._client = v1beta.AlloyDBAdminClient(
                 credentials=credentials,
                 transport="grpc",
@@ -126,7 +126,11 @@ def __init__(
         self._credentials = credentials
         # asyncpg does not currently support using metadata exchange
         # only use metadata exchange for pg8000 driver
-        self._use_metadata = True if driver == "pg8000" else False
+        use_metadata = True
+        if driver in ("asyncpg", None):
+            use_metadata = False
+
+        self._use_metadata = use_metadata
         self._user_agent = user_agent
 
     async def _get_metadata(

google/cloud/alloydbconnector/connector.py

@@ -24,7 +24,7 @@
 import struct
 from threading import Thread
 from types import TracebackType
-from typing import Any, Optional, TYPE_CHECKING
+from typing import Any, Callable, Optional, TYPE_CHECKING
 
 from google.auth import default
 from google.auth.credentials import TokenState
@@ -39,6 +39,7 @@
 from google.cloud.alloydbconnector.instance import RefreshAheadCache
 from google.cloud.alloydbconnector.lazy import LazyRefreshCache
 import google.cloud.alloydbconnector.pg8000 as pg8000
+import google.cloud.alloydbconnector.psycopg as psycopg
 from google.cloud.alloydbconnector.static import StaticConnectionInfoCache
 from google.cloud.alloydbconnector.types import CacheTypes
 from google.cloud.alloydbconnector.utils import generate_keys
@@ -227,8 +228,9 @@ async def connect_async(self, instance_uri: str, driver: str, **kwargs: Any) ->
             self._cache[instance_uri] = cache
             logger.debug(f"['{instance_uri}']: Connection info added to cache")
 
-        connect_func = {
+        connect_func: dict[str, Callable[..., Any]] = {
             "pg8000": pg8000.connect,
+            "psycopg": psycopg.connect,
         }
         # only accept supported database drivers
         try:

google/cloud/alloydbconnector/psycopg.py

@@ -0,0 +1,167 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import logging
+import os
+import socket
+import ssl
+import tempfile
+import threading
+from typing import Any, TYPE_CHECKING
+
+if TYPE_CHECKING:
+    import psycopg
+
+logger = logging.getLogger(name=__name__)
+
+_CHUNK_SIZE = 8 * 1024  # bytes per recv() call inside the proxy forwarding loop
+
+
+def _proxy(local: socket.socket, remote: "ssl.SSLSocket") -> None:
+    """Bidirectionally proxy bytes between a local Unix socket and a remote
+    SSL socket.
+
+    Spawns one daemon thread for the remote→local direction and runs the
+    local→remote direction in the calling thread. Blocks until the calling
+    thread's direction reaches EOF or a socket error, at which point both
+    sockets are closed so the other thread also unblocks and exits.
+
+    Args:
+        local: The Unix domain socket connected to the database driver.
+        remote: The SSL socket connected to the AlloyDB proxy server.
+    """
+
+    def forward(src: Any, dst: Any) -> None:
+        buf = bytearray(_CHUNK_SIZE)
+        view = memoryview(buf)
+        try:
+            while True:
+                n = src.recv_into(view)
+                if n == 0:
+                    logger.debug("psycopg proxy: EOF on %s, closing both sockets", src)
+                    break
+                dst.sendall(view[:n])
+        except (OSError, ssl.SSLError) as e:
+            logger.debug("psycopg proxy: socket error on %s: %s", src, e)
+        finally:
+            # Close both ends so the sibling thread also unblocks.
+            for s in (local, remote):
+                try:
+                    # shutdown is required on POSIX systems to forcefully
+                    # interrupt the sibling thread's blocking recv_into() call.
+                    # Simply closing the socket does not interrupt the system
+                    # call and leads to leaked threads.
+                    s.shutdown(socket.SHUT_RDWR)
+                except OSError:
+                    pass
+                try:
+                    s.close()
+                except OSError:
+                    pass
+
+    threading.Thread(target=forward, args=(remote, local), daemon=True).start()
+    forward(local, remote)  # run in calling thread rather than spawning a third
+
+
+def connect(remote_sock: "ssl.SSLSocket", **kwargs: Any) -> "psycopg.Connection":
+    """Create a psycopg DBAPI connection object.
+
+    Because psycopg does not accept a pre-connected socket, this function
+    creates a temporary Unix domain socket, tells psycopg to connect there,
+    and runs a background proxy that forwards bytes between that socket and
+    the already-established AlloyDB TLS connection.
+
+    Args:
+        remote_sock (ssl.SSLSocket): SSL/TLS secure socket stream connected to the
+            AlloyDB proxy server.
+
+    Returns:
+        psycopg.Connection: A psycopg Connection object for the AlloyDB instance.
+    """
+    try:
+        import psycopg
+    except ImportError:
+        raise ImportError(
+            'Unable to import module "psycopg." Please install and try again.'
+        )
+
+    tmpdir = tempfile.mkdtemp()
+    socket_path = os.path.join(tmpdir, ".s.PGSQL.5432")
+    logger.debug("psycopg: created Unix socket at %s", socket_path)
+
+    local_sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+    local_sock.bind(socket_path)
+    local_sock.listen(1)
+
+    def _accept_and_proxy() -> None:
+        """Accept one connection then proxy bytes until the connection closes."""
+        try:
+            unix_conn, _ = local_sock.accept()
+            local_sock.close()
+            logger.debug("psycopg proxy: accepted connection, starting proxy")
+        except OSError as e:
+            logger.debug("psycopg proxy: accept failed: %s", e)
+            try:
+                remote_sock.close()
+            except OSError:
+                pass
+            return
+        _proxy(unix_conn, remote_sock)
+
+    threading.Thread(target=_accept_and_proxy, daemon=True).start()
+
+    user = kwargs.pop("user")
+    db = kwargs.pop("db")
+    passwd = kwargs.pop("password", None)
+    # SSL is already handled by the underlying SSLSocket; disable it on the
+    # Unix socket so psycopg does not attempt a second TLS handshake.
+    kwargs.pop("sslmode", None)
+
+    logger.debug("psycopg: connecting as user=%s dbname=%s", user, db)
+    try:
+        conn = psycopg.connect(
+            user=user,
+            dbname=db,
+            password=passwd,
+            host=tmpdir,
+            port=5432,
+            sslmode="disable",
+            **kwargs,
+        )
+        logger.debug("psycopg: connection established")
+        return conn
+    except Exception as e:
+        logger.debug("psycopg: connection failed: %s", e)
+        # psycopg never connected (or failed mid-handshake); close the server
+        # socket so the proxy thread unblocks and exits cleanly.
+        try:
+            local_sock.close()
+        except OSError:
+            pass
+        try:
+            remote_sock.close()
+        except OSError:
+            pass
+        raise
+    finally:
+        # The socket file and its parent directory are only needed during the
+        # initial connect() call; remove them now regardless of outcome.
+        try:
+            os.remove(socket_path)
+        except OSError:
+            pass
+        try:
+            os.rmdir(tmpdir)
+        except OSError:
+            pass

pyproject.toml

@@ -60,6 +60,7 @@ Changelog = "https://github.com/GoogleCloudPlatform/alloydb-python-connector/blo
 [project.optional-dependencies]
 pg8000 = ["pg8000>=1.31.1"]
 asyncpg = ["asyncpg>=0.31.0"]
+psycopg = ["psycopg>=3.1.0"]
 
 [tool.setuptools.dynamic]
 version = { attr = "google.cloud.alloydbconnector.version.__version__" }

requirements-test.txt

@@ -2,6 +2,8 @@ asyncpg==0.31.0
 mock==5.2.0
 pg8000==1.31.5
 psycopg2-binary==2.9.11
+psycopg==3.3.3
+psycopg-binary==3.3.3
 pytest==9.0.2
 pytest-asyncio==1.3.0
 pytest-cov==7.0.0