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

UzielSilva requested a review froma team as acode owner

July 16, 2025 23:42

Copy link

google-clabot commentedJul 16, 2025

Thanks for your pull request! It looks like this may be your first contribution to a Google open source project. Before we can look at your pull request, you'll need to sign a Contributor License Agreement (CLA).

View thisfailed invocation of the CLA check for more information.

For the most up to date status, view the checks section at the bottom of the pull request.

blunderbuss-gcfbot assignedhessjcg

Jul 16, 2025

UzielSilva force-pushed theissues/214-psycopg-support branch from2c802ae to4753351Compare

July 16, 2025 23:48

UzielSilva mentioned this pull request

Jul 16, 2025

Add support for psycopg2 driver.#214

Open

chore(main): release 1.18.3 (GoogleCloudPlatform#1309)

280b72c

Co-authored-by: release-please[bot] <55107282+release-please[bot]@users.noreply.github.com>

UzielSilva force-pushed theissues/214-psycopg-support branch 4 times, most recently from9da84d8 to0ba0fb7Compare

July 17, 2025 00:11

UzielSilva changed the title~~Issue #214: Add support for psycopg~~feat(main): Add support for psycopg (Issue #214)

Jul 17, 2025

Copy link

Collaborator

hessjcg commentedJul 17, 2025

Thank you for the contribution! We are excited to add this to the code. I will give this careful review and get back in touch soon with some additional comments.

hessjcg reviewed

Jul 17, 2025

View reviewed changes

Copy link

Collaborator

hessjcg left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others.Learn more.

Nice work so far. This is going to be a very popular enhancement.

You may want to look at a similar local unix proxy implementation in thenode connector.

There are a few things I'd like you to add to this PR. This will get it closer to production-worthy code:

The socket path should be a configuration parameterlocal_socket_path onconnector.connect() instead of being hard-coded. This will allow an application to open many connectors to different database instances.
When the listener opens the unix socket, it should set the file permissions to be readable only by the user.
Handle exceptions from creating and binding the unix socket, log an appropriate error, and raise a LocalProxyStartupError. Define this new exception in google/cloud/sql/connector/exceptions.py
Consider using asyncio instead of a thread to copy data to and from the proxy.
When the connector is closed, thelocal_communication loop must be stopped and the socket file removed.

Again, I'm grateful that you put in the time and wrote this feature. Thank you.

Copy link

Member

enocom commentedJul 17, 2025

cc@rhatgadkar-goog who owns the AlloyDB Python Connector and will be interested to see this.

Big +1 on what@hessjcg. This will be a huge win for people wanting to use psycopg and Cloud SQL (or AlloyDB).

Copy link

Author

UzielSilva commentedJul 18, 2025

All your comments were addressed on my last commit, I even improve the integration test to use SQLAlchemy like other tests.
I only have this issue when I ran the integration test:

Task was destroyed but it is pending!task: <Task pending name='Task-12' coro=<local_communication() done, defined at /home/uziel/Enterprises/TSA/cloud-sql-python-connector/google/cloud/sql/connector/proxy.py:72> wait_for=<Future pending cb=[BaseSelectorEventLoop._sock_read_done(20, handle=<Handle BaseS...0>, 10485760)>)(), Task.task_wakeup()]>>

Not sure how to handle or if it is expected, this is caused because thelocal_communication is still expecting your input when you close the connection. Not sure if there is a more friendly way to stop it, but oh well.
Let me know if this is good enough for production.

Uziel Silva added2 commits

July 18, 2025 13:06

feat(main): add support for psycopg

89719a4

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

fix(main): fix feedback PR

28c1c40

Changelog:- Make local_socket_path configurable- Set right file permissions- Handle exceptions properly- Use asyncio and the main loop to stop the local proxy and clear the file when the connector is stopped

UzielSilva force-pushed theissues/214-psycopg-support branch frombb2e9d8 to28c1c40Compare

July 18, 2025 19:07

enocom mentioned this pull request

Jul 21, 2025

Consider support for psycopg3 driver.#219

Open

Copy link

Author

UzielSilva commentedJul 22, 2025

FYI: The last issue I mentioned is only a warning, so, actually, it is not affecting the feature.

Copy link

rhatgadkar-goog commentedJul 23, 2025•
edited
Loading

All your comments were addressed on my last commit, I even improve the integration test to use SQLAlchemy like other tests. I only have this issue when I ran the integration test:
Task was destroyed but it is pending!task: <Task pending name='Task-12' coro=<local_communication() done, defined at /home/uziel/Enterprises/TSA/cloud-sql-python-connector/google/cloud/sql/connector/proxy.py:72> wait_for=<Future pending cb=[BaseSelectorEventLoop._sock_read_done(20, handle=<Handle BaseS...0>, 10485760)>)(), Task.task_wakeup()]>>
Not sure how to handle or if it is expected, this is caused because thelocal_communication is still expecting your input when you close the connection. Not sure if there is a more friendly way to stop it, but oh well. Let me know if this is good enough for production.

Thank you for this PR of adding support for psycopg! We really appreciate this change.

I took a look at this StackOverflow post to help understand what might be happening here:https://stackoverflow.com/questions/71542947/how-can-i-fix-task-was-destroyed-but-it-is-pending.

I think you're getting the error, because when the connector closes, you're not cancelling the local proxy server task and waiting for it to shutdown. So maybe you can try making the following changes:

Modifystart_local_proxy() function to return the created asyncio task and save this as a member variable in theConnector object.
And then in theConnector class'sclose_async() function, you can cancel the proxy server's asyncio task and gracefully wait for it to cancel.RefreshAheadCache's close() function might be a good example for how to do this.

fix(main): Prevent asyncio destroyed task warning

f5be3ac

Changelog:- Return the asyncio task from `start_local_proxy`- Handle it in `close_async` to cancel it gracefully

Copy link

Author

UzielSilva commentedJul 23, 2025

@rhatgadkar-goog It worked!, now the warning is gone and the server is being stopped gracefully. Thanks!

So, I think the PR is ready for a review, let me know if you want me to do some other modifications. :)

Copy link

Collaborator

kgala2 commentedJul 24, 2025

/gcbrun

kgala2 requested changes

Jul 24, 2025

View reviewed changes

google/cloud/sql/connector/connector.pyShow resolvedHide resolved

google/cloud/sql/connector/connector.py OutdatedShow resolvedHide resolved

google/cloud/sql/connector/psycopg.py OutdatedShow resolvedHide resolved

Uziel Silva added2 commits

July 24, 2025 20:35

fix(main) Fix linting and undefined cases

2702590

Changelog:- Fix linting issues- Define `self.proxy` on the constructor- Prevent issues with undefined variables

fix(main): Fix psycopg unit test

42dab35

Copy link

Author

UzielSilva commentedJul 25, 2025

Also fixed a unit test, test coverage is restored to 100%

fix(main): Fix linting

8c3ce21

Copy link

Collaborator

kgala2 commentedJul 28, 2025

/gcbrun

Copy link

Author

UzielSilva commentedJul 29, 2025

Tests were added, code coverage is now at 94%

Copy link

Collaborator

kgala2 commentedJul 29, 2025

/gcbrun

Copy link

Author

UzielSilva commentedJul 29, 2025

Huh, do I need to also keep support for python 3.9? let me check then the fixes I need to do for it. :)

fix(main): Add support for Python 3.9

c8985b6

Copy link

Author

UzielSilva commentedJul 29, 2025

Added support for Python 3.9, only needed to add some exceptions to handle when I close the server

Copy link

Collaborator

kgala2 commentedJul 29, 2025

/gcbrun

Copy link

Author

UzielSilva commentedJul 29, 2025

Still not able to merge it, it says changes requested

Copy link

Member

enocom commentedJul 29, 2025

Only maintainers can merge.

@kgala2 will have to merge.

Great work on this@UzielSilva 👏.

enocom mentioned this pull request

Jul 29, 2025

Add support for psycopgGoogleCloudPlatform/alloydb-python-connector#377

Open

Copy link

Collaborator

kgala2 commentedJul 30, 2025

Hi@UzielSilva,

The overall PR looks good, I am checking for corner cases that might not be covered, I'll comment on this PR if I find anything and have psyopg support available as soon as possible.
cc:@hessjcg

Thanks

Copy link

sahibpandori commentedAug 11, 2025

Thanks for working on this@UzielSilva, very excited for this change! Hope this can be merged soon@kgala2 @hessjcg

Copy link

Author

UzielSilva commentedAug 11, 2025

Wondering if I should specify the binary flavor for the psycopg optional dependency in the pyproject.toml. I'm using that one for the tests that are working well... not sure how it is going to behave with the purely python based installation.

Ref:https://www.psycopg.org/psycopg3/docs/basic/install.html

CC:@kgala2

UzielSilva commented

Aug 11, 2025

View reviewed changes

pyproject.toml OutdatedShow resolvedHide resolved

hessjcg requested changes

Aug 15, 2025

View reviewed changes

Copy link

Collaborator

hessjcg left a comment•
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others.Learn more.

Excellent work so far@UzielSilva. This will fix both#219 and#214

In reviewing the implementation, this is a good prototype, but it needs to handle sockets in a different way in order to be production-worthy. As implemented today, it can only handle a single connection.Socket Programming in Python, particularly the section on "handling multiple connections" explains how the code should handle the unix server socket.

Also, I think we should rename this feature to "Local Unix Socket" instead of "Psychopg support". We are implementing this in a way that is not database driver specific, and will work on any system that supports unix sockets.

First, lets make a new class inproxy.py calledProxy. This will encapsulate the details for the proxy server.

Create a new method on the Connector:Connector.start_unix_socket_proxy_async, which will create a newProxy object. The Connector should allowConnector.start_unix_socket_proxy_async() to be called multiple times, opening multiple unix sockets as long as thelocal_socket_path parameter does not conflict with an existing socket.

When theProxy object is created, it will start an "accept loop" async task which will open the unix server socket and listen for new connections. TheProxy should keep track of these async tasks so that they can be canceled whenConnector.close() is called.

When the "accept loop" task accepts a new connection, a new client socket is opened. Then the accept loop task will create a new async task to handle the client socket. This "client socket" task will do two things:

call connector.connect() to establish a new socket to the server proxy
create async tasks to copy data between the unix socket and the database instance socket

The client socket task will also need to handle the following error cases:

When theconnector.connect() fails, it should log an error and close the unix socket
When the database socket is suddenly closed, we should close the unix socket
When the unix client socket is suddenly closed, we should close the database socket

Cleanup: When connector.close() is called, the connector

cancel the "accept loop" task
close the unix socket server and delete the socket from the filesystem.
briefly wait for all in-progress async tasks handling connections to close
close all open sockets.

I will added additional comments in-line with the code.

Example of how to use Connector with psychopg and SqlAlchemy

fromgoogle.cloud.sql.connectorimportConnectorasyncdefcreate_sqlalchemy_engine(instance_connection_name:str,user:str,password:str,db:str,ip_type:str="public",refresh_strategy:str="background",resolver:Union[type[DefaultResolver],type[DnsResolver]]=DefaultResolver,)->tuple[sqlalchemy.engine.Engine,Connector]:unix_socket_path="/tmp/conn"connector=Connector(refresh_strategy=refresh_strategy,resolver=resolver)awaitconnector.start_unix_socket_proxy_async(instance_connection_name,user=user,password=password,db=db,local_socket_path=unix_socket_path,ip_type=ip_type,# can be "public", "private" or "psc"autocommit=True,    )# Construct the connection string# Note: The 'host' parameter points to the directory, not the socket file.connection_string=f"postgresql+psycopg2://{user}:{password}@{unix_socket_path}/{db}"# Create the SQLAlchemy engineengine=create_engine(connection_string)returnengine,connector

Thank you again for your hard work on this. I look forward to seeing your next revision.
-Jonathan

google/cloud/sql/connector/proxy.py OutdatedShow resolvedHide resolved

google/cloud/sql/connector/proxy.py Outdated

		whileTrue:
		data=awaitloop.sock_recv(client,LOCAL_PROXY_MAX_MESSAGE_SIZE)
		ifnotdata:
		client.close()

Copy link

Collaborator

hessjcgAug 15, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others.Learn more.

Update the code for sending/receiving data based on best practices described inSocket Programming in Python

Copy link

Author

UzielSilvaAug 19, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others.Learn more.

The reference suggests to create our own event loop. It is probably better to just use asyncio in a way to allow the proxy to receive multiple collections (working on it), but I won't be following 100% what is said in the ref since we are already using asyncio in other parts of our codebase.

google/cloud/sql/connector/proxy.py OutdatedShow resolvedHide resolved

google/cloud/sql/connector/connector.py OutdatedShow resolvedHide resolved

google/cloud/sql/connector/connector.pyShow resolvedHide resolved

google/cloud/sql/connector/connector.py OutdatedShow resolvedHide resolved

google/cloud/sql/connector/enums.pyShow resolvedHide resolved

fix(main): Make local proxy to accept multiple connections (WIP)

4f6f388

Copy link

Collaborator

hessjcg commentedSep 5, 2025

Hi@UzielSilva, I was able to spend some time on the tricky IO code to set up a local proxy server. This is still a work in progress, but I wanted to share it anyway. Please check out this branch:

https://github.com/GoogleCloudPlatform/cloud-sql-python-connector/tree/feat-local-proxy

Currently there are still some failing tests and edge cases that are not properly handled. Please feel free to continue my work.

Copy link

Author

UzielSilva commentedSep 10, 2025

Hi there, sorry for the lack of updates, took some mini vacations.
@hessjcg great stuff!!, I'm moving your proxy server into my branch. This is very helpful. :)

Uziel Silvaand others added4 commits

September 10, 2025 13:18

test: Fix compilation errors

7362233

feat: Add proxy server and fix all unit tests

1eb6af2

Merge branch 'main' into issues/214-psycopg-support

b071556

feat: replace ssl_object with sslcontext and build a socket manually

f8391aa

Copy link

Author

UzielSilva commentedOct 17, 2025

I'm having issues trying to fix one integration test, the psycopg one actually.

The issue is about a task that got Future attached to a different loop. The thing is that if I inject the running loop to the connector the issue appears to vanish but the connection now gives us a timeout error. All unit and integration tests are now passing, except for that one, and that is the only issue that is stopping me from finishing this PR. Can I get some help?

feat: Pass loop to aiohttp client

3791368

Copy link

Author

UzielSilva commentedOct 17, 2025

NVM, I solved it.

@hessjcg I think I'm ready for a new review.

Copy link

seeker25 commentedNov 13, 2025

This would be very useful usingpg8000 it's fairly slow especially when it comes to inserting or fetching many rows. I don't thinkasyncpg would be super easy to implement in an existing app that is synchronous. I believepsycopg would be much faster.