feat: add support for psycopg

README.md

@@ -42,6 +42,7 @@ The Cloud SQL Python Connector is a package to be used alongside a database driv
 Currently supported drivers are:
  - [`pymysql`](https://github.com/PyMySQL/PyMySQL) (MySQL)
  - [`pg8000`](https://github.com/tlocke/pg8000) (PostgreSQL)
+ - [`psycopg`](https://github.com/psycopg/psycopg) (PostgreSQL)
  - [`asyncpg`](https://github.com/MagicStack/asyncpg) (PostgreSQL)
  - [`pytds`](https://github.com/denisenkom/pytds) (SQL Server)
 
@@ -56,12 +57,16 @@ based on your database dialect.
 pip install "cloud-sql-python-connector[pymysql]"
 ```
 ### Postgres
-There are two different database drivers that are supported for the Postgres dialect:
+There are three different database drivers that are supported for the Postgres dialect:
 
 #### pg8000
 ```
 pip install "cloud-sql-python-connector[pg8000]"
 ```
+#### psycopg
+```
+pip install "cloud-sql-python-connector[psycopg]"
+```
 #### asyncpg
 ```
 pip install "cloud-sql-python-connector[asyncpg]"
@@ -137,6 +142,18 @@ pool = sqlalchemy.create_engine(
         db="my-db-name"
     ),
 )
+
+# Or with Postgres (psycopg):
+pool = sqlalchemy.create_engine(
+    "postgresql+psycopg://",
+    creator=lambda: connector.connect(
+        "project:region:instance",
+        "psycopg",
+        user="my-user",
+        password="my-password",
+        db="my-db-name"
+    ),
+)
 ```
 
 The returned connection pool engine can then be used to query and modify the database.

google/cloud/sql/connector/connector.py

@@ -31,6 +31,7 @@
 
 from google.cloud.sql.connector import asyncpg
 from google.cloud.sql.connector import pg8000
+from google.cloud.sql.connector import psycopg
 from google.cloud.sql.connector import pymysql
 from google.cloud.sql.connector import pytds
 from google.cloud.sql.connector.client import CloudSQLClient
@@ -362,6 +363,7 @@ async def connect_async(
             "pg8000": pg8000.connect,
             "asyncpg": asyncpg.connect,
             "pytds": pytds.connect,
+            "psycopg": psycopg.connect,
         }
 
         # only accept supported database drivers

google/cloud/sql/connector/enums.py

@@ -62,6 +62,7 @@ class DriverMapping(Enum):
 
     ASYNCPG = "POSTGRES"
     PG8000 = "POSTGRES"  # noqa: PIE796
+    PSYCOPG = "POSTGRES"  # noqa: PIE796
     PYMYSQL = "MYSQL"
     PYTDS = "SQLSERVER"
 

google/cloud/sql/connector/psycopg.py

@@ -0,0 +1,233 @@
+# 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 selectors
+import socket
+import ssl
+import tempfile
+import threading
+from typing import Any, TYPE_CHECKING
+
+if TYPE_CHECKING:
+    import psycopg
+
+logger = logging.getLogger(name=__name__)
+
+
+def _proxy(local: socket.socket, remote: "ssl.SSLSocket") -> None:
+    """Single-threaded selectors-based proxy to avoid SSLSocket thread-safety issues."""
+    sel = selectors.DefaultSelector()
+    sel.register(local, selectors.EVENT_READ, data="local")
+    sel.register(remote, selectors.EVENT_READ, data="remote")
+
+    def forward_pending() -> bool:
+        """Read any pending decrypted data from SSL buffer and forward it.
+        Returns True if EOF was reached or error occurred (should exit).
+        """
+        if not hasattr(remote, "pending"):
+            return False
+        pending_bytes = remote.pending()
+        if not isinstance(pending_bytes, int):
+            return False
+
+        while pending_bytes > 0:
+            try:
+                data = remote.recv(8192)
+            except OSError as e:
+                logger.debug("psycopg proxy: remote recv pending error: %s", e)
+                return True
+            if not data:
+                logger.debug("psycopg proxy: remote pending EOF")
+                return True
+            try:
+                local.sendall(data)
+            except OSError as e:
+                logger.debug("psycopg proxy: local send pending error: %s", e)
+                return True
+            try:
+                pending_bytes = remote.pending()
+            except OSError:
+                break
+            if not isinstance(pending_bytes, int):
+                break
+        return False
+
+    try:
+        while True:
+            # First check if there is any pending data in SSL buffer
+            if forward_pending():
+                break
+
+            events = sel.select(timeout=30)
+            if not events:
+                logger.debug("psycopg proxy: inactivity timeout (30s)")
+                break
+
+            for key, mask in events:
+                if key.data == "local":
+                    try:
+                        data = local.recv(8192)
+                    except OSError as e:
+                        logger.debug("psycopg proxy: local recv error: %s", e)
+                        return
+                    if not data:
+                        logger.debug("psycopg proxy: local EOF")
+                        return
+                    try:
+                        remote.sendall(data)
+                    except OSError as e:
+                        logger.debug("psycopg proxy: remote send error: %s", e)
+                        return
+                elif key.data == "remote":
+                    try:
+                        data = remote.recv(8192)
+                    except OSError as e:
+                        logger.debug("psycopg proxy: remote recv error: %s", e)
+                        return
+                    if not data:
+                        logger.debug("psycopg proxy: remote EOF")
+                        return
+                    try:
+                        local.sendall(data)
+                    except OSError as e:
+                        logger.debug("psycopg proxy: local send error: %s", e)
+                        return
+    except OSError as e:
+        logger.debug("psycopg proxy: OSError in loop: %s", e)
+    finally:
+        sel.close()
+        for s in (local, remote):
+            try:
+                s.shutdown(socket.SHUT_RDWR)
+            except OSError:
+                pass
+            try:
+                s.close()
+            except OSError:
+                pass
+
+
+def connect(
+    ip_address: str, 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 Cloud SQL TLS connection.
+
+    Args:
+        ip_address (str): IP address of the Cloud SQL instance.
+        remote_sock (ssl.SSLSocket): SSL/TLS secure socket stream connected to the
+            Cloud SQL proxy server.
+
+    Returns:
+        psycopg.Connection: A psycopg Connection object for the Cloud SQL 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."""
+        unix_conn = None
+        try:
+            unix_conn, _ = local_sock.accept()
+            local_sock.close()
+            logger.debug("psycopg proxy: accepted connection, starting proxy")
+            _proxy(unix_conn, remote_sock)
+        except Exception as e:  # noqa: BLE001
+            logger.debug("psycopg proxy: error in accept/proxy thread: %s", e)
+            # Ensure cleanup on any exception
+            if unix_conn:
+                try:
+                    unix_conn.shutdown(socket.SHUT_RDWR)
+                except OSError:
+                    pass
+                try:
+                    unix_conn.close()
+                except OSError:
+                    pass
+            try:
+                remote_sock.shutdown(socket.SHUT_RDWR)
+            except OSError:
+                pass
+            try:
+                remote_sock.close()
+            except OSError:
+                pass
+
+    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)
+    timeout = kwargs.pop("timeout", None)
+    if timeout is not None:
+        kwargs["connect_timeout"] = int(timeout)
+
+    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

@@ -63,6 +63,7 @@ pymysql = ["PyMySQL>=1.1.0"]
 pg8000 = ["pg8000>=1.31.1"]
 pytds = ["python-tds>=1.15.0"]
 asyncpg = ["asyncpg>=0.30.0"]
+psycopg = ["psycopg>=3.1.0"]
 
 [tool.setuptools.dynamic]
 version = { attr = "google.cloud.sql.connector.version.__version__" }

requirements-test.txt

@@ -11,4 +11,6 @@ asyncpg==0.31.0
 python-tds==1.17.1
 aioresponses==0.7.9
 pytest-aiohttp==1.1.1
+psycopg==3.3.4
+psycopg-binary==3.3.4