Changelog:

- Add proxy for connections that can only be made through an unix socket, to support the TLS connection
- Add support for psycopg, using the proxy server
- Add unit and integration tests
- Update docs

Author: Uziel Silva

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)
 
@@ -587,7 +588,7 @@ async def main():
     # acquire connection and query Cloud SQL database
     async with pool.acquire() as conn:
         res = await conn.fetch("SELECT NOW()")
-    
+
     # close Connector
     await connector.close_async()
 ```

google/cloud/sql/connector/connector.py

@@ -37,6 +37,7 @@
 from google.cloud.sql.connector.lazy import LazyRefreshCache
 from google.cloud.sql.connector.monitored_cache import MonitoredCache
 import google.cloud.sql.connector.pg8000 as pg8000
+import google.cloud.sql.connector.psycopg as psycopg
 import google.cloud.sql.connector.pymysql as pymysql
 import google.cloud.sql.connector.pytds as pytds
 from google.cloud.sql.connector.resolver import DefaultResolver
@@ -230,7 +231,7 @@ def connect(
                 Example: "my-project:us-central1:my-instance"
 
             driver (str): A string representing the database driver to connect
-                with. Supported drivers are pymysql, pg8000, and pytds.
+                with. Supported drivers are pymysql, pg8000, psycopg, and pytds.
 
             **kwargs: Any driver-specific arguments to pass to the underlying
                 driver .connect call.
@@ -266,7 +267,8 @@ async def connect_async(
                 Example: "my-project:us-central1:my-instance"
 
             driver (str): A string representing the database driver to connect
-                with. Supported drivers are pymysql, asyncpg, pg8000, and pytds.
+                with. Supported drivers are pymysql, asyncpg, pg8000, psycopg, and
+                pytds.
 
             **kwargs: Any driver-specific arguments to pass to the underlying
                 driver .connect call.
@@ -278,7 +280,7 @@ async def connect_async(
             ValueError: Connection attempt with built-in database authentication
                 and then subsequent attempt with IAM database authentication.
             KeyError: Unsupported database driver Must be one of pymysql, asyncpg,
-                pg8000, and pytds.
+                pg8000, psycopg, and pytds.
         """
         if self._keys is None:
             self._keys = asyncio.create_task(generate_keys())
@@ -332,6 +334,7 @@ async def connect_async(
         connect_func = {
             "pymysql": pymysql.connect,
             "pg8000": pg8000.connect,
+            "psycopg": psycopg.connect,
             "asyncpg": asyncpg.connect,
             "pytds": pytds.connect,
         }

google/cloud/sql/connector/enums.py

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

google/cloud/sql/connector/proxy.py

@@ -0,0 +1,56 @@
+"""
+Copyright 2025 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
+
+  https://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 socket
+import os
+import threading
+
+SERVER_PROXY_PORT = 3307
+
+def start_local_proxy(
+  ssl_sock,
+  socket_path,
+):
+  if os.path.exists(socket_path):
+      os.remove(socket_path)
+  conn_unix = None
+  unix_socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+
+  unix_socket.bind(socket_path)
+  unix_socket.listen(1)
+
+  threading.Thread(target=local_communication, args=(unix_socket, ssl_sock, socket_path)).start()
+
+
+def local_communication(
+  unix_socket, ssl_sock, socket_path
+):
+  try:
+    conn_unix, addr_unix = unix_socket.accept()
+
+    while True:
+      data = conn_unix.recv(10485760)
+      if not data:
+        break
+      ssl_sock.sendall(data)
+      response = ssl_sock.recv(10485760)
+      conn_unix.sendall(response)
+
+  finally:
+    if conn_unix is not None:
+      conn_unix.close()
+    unix_socket.close()
+    os.remove(socket_path) # Clean up the socket file

google/cloud/sql/connector/psycopg.py

@@ -0,0 +1,72 @@
+"""
+Copyright 2025 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
+
+  https://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 ssl
+from typing import Any, TYPE_CHECKING
+import threading
+
+SERVER_PROXY_PORT = 3307
+
+if TYPE_CHECKING:
+    import psycopg
+
+
+def connect(
+    ip_address: str, sock: ssl.SSLSocket, **kwargs: Any
+) -> "psycopg.Connection":
+    """Helper function to create a psycopg DB-API connection object.
+
+    Args:
+        ip_address (str): A string containing an IP address for the Cloud SQL
+            instance.
+        sock (ssl.SSLSocket): An SSLSocket object created from the Cloud SQL
+            server CA cert and ephemeral cert.
+        kwargs: Additional arguments to pass to the psycopg connect method.
+
+    Returns:
+        psycopg.Connection: A psycopg connection to the Cloud SQL
+            instance.
+
+    Raises:
+        ImportError: The psycopg module cannot be imported.
+    """
+    try:
+        from psycopg.rows import dict_row
+        from psycopg import Connection
+        import threading
+        from google.cloud.sql.connector.proxy import start_local_proxy
+    except ImportError:
+        raise ImportError(
+            'Unable to import module "psycopg." Please install and try again.'
+        )
+
+    user = kwargs.pop("user")
+    db = kwargs.pop("db")
+    passwd = kwargs.pop("password", None)
+
+    kwargs.pop("timeout", None)
+
+    start_local_proxy(sock, f"/tmp/connector-socket/.s.PGSQL.3307")
+
+    conn = Connection.connect(
+        f"host=/tmp/connector-socket port={SERVER_PROXY_PORT} dbname={db} user={user} password={passwd} sslmode=require",
+        autocommit=True,
+        row_factory=dict_row,
+        **kwargs
+    )
+
+    conn.autocommit = True
+    return conn

pyproject.toml

@@ -59,6 +59,7 @@ Changelog = "https://github.com/GoogleCloudPlatform/cloud-sql-python-connector/b
 [project.optional-dependencies]
 pymysql = ["PyMySQL>=1.1.0"]
 pg8000 = ["pg8000>=1.31.1"]
+psycopg = ["psycopg>=3.2.9"]
 pytds = ["python-tds>=1.15.0"]
 asyncpg = ["asyncpg>=0.30.0"]
 

requirements-test.txt

@@ -7,6 +7,7 @@ sqlalchemy-pytds==1.0.2
 sqlalchemy-stubs==0.4
 PyMySQL==1.1.1
 pg8000==1.31.2
+psycopg[binary]==3.2.9
 asyncpg==0.30.0
 python-tds==1.16.1
 aioresponses==0.7.8

tests/system/test_psycopg_connection.py

@@ -0,0 +1,60 @@
+"""
+Copyright 2025 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
+
+  https://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.
+"""
+
+from datetime import datetime
+import os
+
+# [START cloud_sql_connector_postgres_psycopg]
+
+from google.cloud.sql.connector import Connector
+from google.cloud.sql.connector import DefaultResolver
+
+from sqlalchemy.dialects.postgresql.base import PGDialect
+PGDialect._get_server_version_info = lambda *args: (9, 2)
+
+# [END cloud_sql_connector_postgres_psycopg]
+
+
+def test_psycopg_connection() -> None:
+    """Basic test to get time from database."""
+    inst_conn_name = os.environ["POSTGRES_CONNECTION_NAME"]
+    user = os.environ["POSTGRES_USER"]
+    password = os.environ["POSTGRES_PASS"]
+    db = os.environ["POSTGRES_DB"]
+    ip_type = os.environ.get("IP_TYPE", "public")
+
+    connector = Connector(refresh_strategy="background", resolver=DefaultResolver)
+
+    pool = connector.connect(
+        inst_conn_name,
+        "psycopg",
+        user=user,
+        password=password,
+        db=db,
+        ip_type=ip_type,  # can be "public", "private" or "psc"
+    )
+
+    with pool as conn:
+
+        # Open a cursor to perform database operations
+        with conn.cursor() as cur:
+
+            # Query the database and obtain data as Python objects.
+            cur.execute("SELECT NOW()")
+            curr_time = cur.fetchone()["now"]
+            assert type(curr_time) is datetime
+
+

tests/unit/test_psycopg.py

@@ -0,0 +1,40 @@
+"""
+Copyright 2025 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
+
+  https://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 socket
+import ssl
+from typing import Any
+
+from mock import patch, PropertyMock
+import pytest
+
+from google.cloud.sql.connector.psycopg import connect
+
+
+@pytest.mark.usefixtures("proxy_server")
+async def test_psycopg(context: ssl.SSLContext, kwargs: Any) -> None:
+    """Test to verify that psycopg gets to proper connection call."""
+    ip_addr = "127.0.0.1"
+    sock = context.wrap_socket(
+        socket.create_connection((ip_addr, 3307)),
+        server_hostname=ip_addr,
+    )
+    with patch("psycopg.connect") as mock_connect:
+        type(mock_connect.return_value).autocommit = PropertyMock(return_value=True)
+        connection = connect(ip_addr, sock, **kwargs)
+        assert connection.autocommit is True
+        # verify that driver connection call would be made
+        assert mock_connect.assert_called_once