feat: Support Developer Edition connections

This PR adds support for Developer Edition connections via the SqlDataService. 
It includes a fallback mechanism to standard IP connections if the SqlDataService
is not supported by the instance edition.

See GoogleCloudPlatform/cloud-sql-go-connector#1108

Author: hessjcg

.gitignore

@@ -10,6 +10,7 @@ dist/
 sponge_log.xml
 .envrc
 *.iml
+build/
 .mypy_cache/
 .nox/
 .pytest_cache/

google/cloud/sql/connector/asyncpg.py

@@ -14,6 +14,8 @@
 limitations under the License.
 """
 
+from __future__ import annotations
+
 import ssl
 from typing import Any, TYPE_CHECKING
 
@@ -24,16 +26,15 @@
 
 
 async def connect(
-    ip_address: str, ctx: ssl.SSLContext, **kwargs: Any
-) -> "asyncpg.Connection":
+    ip_address: str, ctx: ssl.SSLContext | None, **kwargs: Any
+) -> asyncpg.Connection:
     """Helper function to create an asyncpg DB-API connection object.
 
     Args:
         ip_address (str): A string containing an IP address for the Cloud SQL
             instance.
         ctx (ssl.SSLContext): An SSLContext object created from the Cloud SQL
-            server CA cert and ephemeral cert.
-            server CA cert and ephemeral cert.
+            server CA cert and ephemeral cert. Pass None to disable SSL.
         kwargs: Keyword arguments for establishing asyncpg connection
             object to Cloud SQL instance.
 
@@ -55,14 +56,18 @@ async def connect(
     if db is None:
         raise KeyError("database")
     passwd = kwargs.pop("password", None)
+    port = kwargs.pop("port", SERVER_PROXY_PORT)
 
-    return await asyncpg.connect(
-        user=user,
-        database=db,
-        password=passwd,
-        host=ip_address,
-        port=SERVER_PROXY_PORT,
-        ssl=ctx,
-        direct_tls=True,
+    connect_args = {
+        "user": user,
+        "database": db,
+        "password": passwd,
+        "host": ip_address,
+        "port": port,
         **kwargs,
-    )
+    }
+    if ctx is not None:
+        connect_args["ssl"] = ctx
+        connect_args["direct_tls"] = True
+
+    return await asyncpg.connect(**connect_args)

google/cloud/sql/connector/client.py

@@ -173,9 +173,13 @@ async def _get_metadata(
             if psc_dns_names:
                 ip_addresses["PSC"] = psc_dns_names
 
+        server_ca_cert = None
+        if "serverCaCert" in ret_dict and "cert" in ret_dict["serverCaCert"]:
+            server_ca_cert = ret_dict["serverCaCert"]["cert"]
+
         return {
             "ip_addresses": ip_addresses,
-            "server_ca_cert": ret_dict["serverCaCert"]["cert"],
+            "server_ca_cert": server_ca_cert,
             "database_version": ret_dict["databaseVersion"],
         }
 
@@ -271,7 +275,11 @@ async def _get_ephemeral(
         finally:
             resp.raise_for_status()
 
-        ephemeral_cert: str = ret_dict["ephemeralCert"]["cert"]
+        try:
+            ephemeral_cert: str = ret_dict["ephemeralCert"]["cert"]
+        except KeyError as e:
+            logger.error(f"KeyError in _get_ephemeral parsing generateEphemeralCert: {e}. Response dict: {ret_dict}")
+            raise
 
         # decode cert to read expiration
         x509 = load_pem_x509_certificate(

google/cloud/sql/connector/connection_info.py

@@ -62,7 +62,7 @@ class ConnectionInfo:
 
     conn_name: ConnectionName
     client_cert: str
-    server_ca_cert: str
+    server_ca_cert: str | None
     private_key: bytes
     ip_addrs: dict[str, Any]
     database_version: str
@@ -78,6 +78,10 @@ async def create_ssl_context(self, enable_iam_auth: bool = False) -> ssl.SSLCont
         # if SSL context is cached, use it
         if self.context is not None:
             return self.context
+        
+        if self.server_ca_cert is None:
+            raise ValueError("Cannot create SSL context: server CA certificate is missing.")
+
         context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
 
         # update ssl.PROTOCOL_TLS_CLIENT default