fix(main): Prevent asyncio destroyed task warning

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

Author: Uziel Silva

google/cloud/sql/connector/connector.py

@@ -401,7 +401,7 @@ async def connect_async(
             if driver in LOCAL_PROXY_DRIVERS:
                 local_socket_path = kwargs.pop("local_socket_path", "/tmp/connector-socket")
                 host = local_socket_path
-                start_local_proxy(
+                self._proxy = start_local_proxy(
                     sock,
                     socket_path=f"{local_socket_path}/.s.PGSQL.{SERVER_PROXY_PORT}",
                     loop=self._loop
@@ -486,6 +486,13 @@ async def close_async(self) -> None:
         await asyncio.gather(*[cache.close() for cache in self._cache.values()])
         if self._client:
             await self._client.close()
+        if self._proxy:
+            proxy_task = asyncio.gather(self._proxy)
+            try:
+                await asyncio.wait_for(proxy_task, timeout=0.1)
+            except TimeoutError:
+                pass # This task runs forever so it is expected to throw this exception
+
 
 
 async def create_async_connector(

google/cloud/sql/connector/proxy.py

@@ -30,7 +30,7 @@ def start_local_proxy(
     ssl_sock: ssl.SSLSocket,
     socket_path: Optional[str] = "/tmp/connector-socket",
     loop: Optional[asyncio.AbstractEventLoop] = None,
-):
+) -> asyncio.Task:
     """Helper function to start a UNIX based local proxy for
     transport messages through the SSL Socket.
 
@@ -40,6 +40,9 @@ def start_local_proxy(
         socket_path: A system path that is going to be used to store the socket.
         loop (asyncio.AbstractEventLoop): Event loop to run asyncio tasks.
 
+    Returns:
+        asyncio.Task: The asyncio task containing the proxy server process.
+
     Raises:
         LocalProxyStartupError: Local UNIX socket based proxy was not able to
         get started.
@@ -66,7 +69,7 @@ def start_local_proxy(
             'Local UNIX socket based proxy was not able to get started.'
         )
 
-    loop.create_task(local_communication(unix_socket, ssl_sock, socket_path, loop))
+    return loop.create_task(local_communication(unix_socket, ssl_sock, socket_path, loop))
 
 
 async def local_communication(