chore: update type hints to standard collections (GoogleCloudPlatform#1183)

Author: jackwotherspoon

google/cloud/sql/connector/asyncpg.py

@@ -13,6 +13,7 @@
 See the License for the specific language governing permissions and
 limitations under the License.
 """
+
 import ssl
 from typing import Any, TYPE_CHECKING
 

google/cloud/sql/connector/client.py

@@ -17,7 +17,7 @@
 import asyncio
 import datetime
 import logging
-from typing import Any, Dict, Optional, Tuple, TYPE_CHECKING
+from typing import Any, Optional, TYPE_CHECKING
 
 import aiohttp
 from cryptography.hazmat.backends import default_backend
@@ -98,7 +98,7 @@ async def _get_metadata(
         project: str,
         region: str,
         instance: str,
-    ) -> Dict[str, Any]:
+    ) -> dict[str, Any]:
         """Requests metadata from the Cloud SQL Instance
         and returns a dictionary containing the IP addresses and certificate
         authority of the Cloud SQL Instance.
@@ -113,7 +113,7 @@ async def _get_metadata(
         :type instance: str
         :param instance: A string representing the name of the instance.
 
-        :rtype: Dict[str: Union[Dict, str]]
+        :rtype: dict[str: Union[dict, str]]
         :returns: Returns a dictionary containing a dictionary of all IP
             addresses and their type and a string representing the
             certificate authority.
@@ -161,7 +161,7 @@ async def _get_ephemeral(
         instance: str,
         pub_key: str,
         enable_iam_auth: bool = False,
-    ) -> Tuple[str, datetime.datetime]:
+    ) -> tuple[str, datetime.datetime]:
         """Asynchronously requests an ephemeral certificate from the Cloud SQL Instance.
 
         :type project: str

google/cloud/sql/connector/connection_info.py

@@ -17,7 +17,7 @@
 from dataclasses import dataclass
 import logging
 import ssl
-from typing import Any, Dict, Optional, TYPE_CHECKING
+from typing import Any, Optional, TYPE_CHECKING
 
 from aiofiles.tempfile import TemporaryDirectory
 
@@ -41,7 +41,7 @@ class ConnectionInfo:
     client_cert: str
     server_ca_cert: str
     private_key: bytes
-    ip_addrs: Dict[str, Any]
+    ip_addrs: dict[str, Any]
     database_version: str
     expiration: datetime.datetime
     context: Optional[ssl.SSLContext] = None

google/cloud/sql/connector/connector.py

@@ -21,7 +21,7 @@
 import logging
 from threading import Thread
 from types import TracebackType
-from typing import Any, Dict, Optional, Tuple, Type, Union
+from typing import Any, Optional, Type, Union
 
 import google.auth
 from google.auth.credentials import Credentials
@@ -133,8 +133,8 @@ def __init__(
                 )
         # initialize dict to store caches, key is a tuple consisting of instance
         # connection name string and enable_iam_auth boolean flag
-        self._cache: Dict[
-            Tuple[str, bool], Union[RefreshAheadCache, LazyRefreshCache]
+        self._cache: dict[
+            tuple[str, bool], Union[RefreshAheadCache, LazyRefreshCache]
         ] = {}
         self._client: Optional[CloudSQLClient] = None
 

google/cloud/sql/connector/instance.py

@@ -22,7 +22,6 @@
 from datetime import timezone
 import logging
 import re
-from typing import Tuple
 
 import aiohttp
 
@@ -43,7 +42,7 @@
 CONN_NAME_REGEX = re.compile(("([^:]+(:[^:]+)?):([^:]+):([^:]+)"))
 
 
-def _parse_instance_connection_name(connection_name: str) -> Tuple[str, str, str]:
+def _parse_instance_connection_name(connection_name: str) -> tuple[str, str, str]:
     if CONN_NAME_REGEX.fullmatch(connection_name) is None:
         raise ValueError(
             "Arg `instance_connection_string` must have "

google/cloud/sql/connector/refresh_utils.py

@@ -21,7 +21,7 @@
 import datetime
 import logging
 import random
-from typing import Any, Callable, List
+from typing import Any, Callable
 
 import aiohttp
 from google.auth.credentials import Credentials
@@ -77,15 +77,15 @@ async def _is_valid(task: asyncio.Task) -> bool:
 
 def _downscope_credentials(
     credentials: Credentials,
-    scopes: List[str] = ["https://www.googleapis.com/auth/sqlservice.login"],
+    scopes: list[str] = ["https://www.googleapis.com/auth/sqlservice.login"],
 ) -> Credentials:
     """Generate a down-scoped credential.
 
     :type credentials: google.auth.credentials.Credentials
     :param credentials
         Credentials object used to generate down-scoped credentials.
 
-    :type scopes: List[str]
+    :type scopes: list[str]
     :param scopes
         List of Google scopes to include in down-scoped credentials object.
 

google/cloud/sql/connector/utils.py

@@ -14,15 +14,13 @@
 limitations under the License.
 """
 
-from typing import Tuple
-
 import aiofiles
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives import serialization
 from cryptography.hazmat.primitives.asymmetric import rsa
 
 
-async def generate_keys() -> Tuple[bytes, str]:
+async def generate_keys() -> tuple[bytes, str]:
     """A helper function to generate the private and public keys.
 
     backend - The value specified is default_backend(). This is because the
@@ -61,7 +59,7 @@ async def generate_keys() -> Tuple[bytes, str]:
 
 async def write_to_file(
     dir_path: str, serverCaCert: str, ephemeralCert: str, priv_key: bytes
-) -> Tuple[str, str, str]:
+) -> tuple[str, str, str]:
     """
     Helper function to write the serverCaCert, ephemeral certificate and
     private key to .pem files in a given directory

tests/system/test_asyncpg_connection.py

@@ -16,7 +16,7 @@
 
 import asyncio
 import os
-from typing import Any, Tuple
+from typing import Any
 
 import asyncpg
 import sqlalchemy
@@ -31,7 +31,7 @@ async def create_sqlalchemy_engine(
     password: str,
     db: str,
     refresh_strategy: str = "background",
-) -> Tuple[sqlalchemy.ext.asyncio.engine.AsyncEngine, Connector]:
+) -> tuple[sqlalchemy.ext.asyncio.engine.AsyncEngine, Connector]:
     """Creates a connection pool for a Cloud SQL instance and returns the pool
     and the connector. Callers are responsible for closing the pool and the
     connector.
@@ -94,7 +94,7 @@ async def create_asyncpg_pool(
     password: str,
     db: str,
     refresh_strategy: str = "background",
-) -> Tuple[asyncpg.Pool, Connector]:
+) -> tuple[asyncpg.Pool, Connector]:
     """Creates a native asyncpg connection pool for a Cloud SQL instance and
     returns the pool and the connector. Callers are responsible for closing the
     pool and the connector.

tests/system/test_asyncpg_iam_auth.py

@@ -16,7 +16,6 @@
 
 import asyncio
 import os
-from typing import Tuple
 
 import asyncpg
 import sqlalchemy
@@ -30,7 +29,7 @@ async def create_sqlalchemy_engine(
     user: str,
     db: str,
     refresh_strategy: str = "background",
-) -> Tuple[sqlalchemy.ext.asyncio.engine.AsyncEngine, Connector]:
+) -> tuple[sqlalchemy.ext.asyncio.engine.AsyncEngine, Connector]:
     """Creates a connection pool for a Cloud SQL instance and returns the pool
     and the connector. Callers are responsible for closing the pool and the
     connector.

tests/system/test_ip_types.py

@@ -13,6 +13,7 @@
 See the License for the specific language governing permissions and
 limitations under the License.
 """
+
 import os
 import uuid