Use python3.10 typing primitives

bin/cleanup/cleanup.py

@@ -31,7 +31,7 @@
 import re
 import subprocess
 import sys
-from typing import Any, Callable, List, Optional
+from typing import Any, Callable, List
 
 from absl import app
 from absl import flags
@@ -379,7 +379,7 @@ def cleanup_client(
     gcp_api_manager,
     gcp_service_account,
     *,
-    suffix: Optional[str] = "",
+    suffix: str | None = "",
 ):
     deployment_name = xds_flags.CLIENT_NAME.value
     if suffix:
@@ -422,7 +422,7 @@ def cleanup_server(
     gcp_api_manager,
     gcp_service_account,
     *,
-    suffix: Optional[str] = "",
+    suffix: str | None = "",
 ):
     deployment_name = xds_flags.SERVER_NAME.value
     if suffix:
@@ -563,7 +563,7 @@ def delete_k8s_resources(
 
 def _rule_match_k8s_namespace(
     namespace_name: str, k8s_resource_rules: List[K8sResourceRule]
-) -> Optional[K8sResourceRule]:
+) -> K8sResourceRule | None:
     for rule in k8s_resource_rules:
         result = re.search(rule.expression, namespace_name)
         if result is not None:

framework/bootstrap_generator_testcase.py

@@ -12,7 +12,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import logging
-from typing import Optional
 
 from typing_extensions import override
 
@@ -127,7 +126,7 @@ def initTrafficDirectorManager(cls) -> TrafficDirectorManager:
 
     @classmethod
     def initKubernetesServerRunner(
-        cls, *, td_bootstrap_image: Optional[str] = None
+        cls, *, td_bootstrap_image: str | None = None
     ) -> KubernetesServerRunner:
         if not td_bootstrap_image:
             td_bootstrap_image = cls.td_bootstrap_image
@@ -165,7 +164,7 @@ def startTestServer(
         return test_server
 
     def initKubernetesClientRunner(
-        self, td_bootstrap_image: Optional[str] = None
+        self, td_bootstrap_image: str | None = None
     ) -> KubernetesClientRunner:
         if not td_bootstrap_image:
             td_bootstrap_image = self.td_bootstrap_image

framework/helpers/datetime.py

@@ -14,7 +14,7 @@
 """This contains common helpers for working with dates and time."""
 import datetime
 import re
-from typing import Optional, Pattern
+from typing import Pattern
 
 import dateutil.parser
 
@@ -57,7 +57,7 @@ def datetime_suffix(*, seconds: bool = False) -> str:
     return utc_now().strftime("%Y%m%d-%H%M" + ("%S" if seconds else ""))
 
 
-def ago(date_from: datetime.datetime, now: Optional[datetime.datetime] = None):
+def ago(date_from: datetime.datetime, now: datetime.datetime | None = None):
     if not now:
         now = utc_now()
 

framework/helpers/grpc.py

@@ -15,7 +15,6 @@
 import collections
 import dataclasses
 import functools
-from typing import Optional
 
 import grpc
 from typing_extensions import TypeAlias
@@ -31,7 +30,7 @@
 
 
 @functools.cache
-def status_from_int(grpc_status_int: int) -> Optional[grpc.StatusCode]:
+def status_from_int(grpc_status_int: int) -> grpc.StatusCode | None:
     """Converts the integer gRPC status code to the grpc.StatusCode enum."""
     for grpc_status in grpc.StatusCode:
         if grpc_status.value[0] == grpc_status_int:
@@ -77,7 +76,7 @@ def from_response(
     ) -> "PrettyStatsPerMethod":
         stats: dict[str, int] = dict()
         for status_int, count in method_stats.result.items():
-            status: Optional[grpc.StatusCode] = status_from_int(status_int)
+            status: grpc.StatusCode | None = status_from_int(status_int)
             status_formatted = status_pretty(status) if status else "None"
             stats[status_formatted] = count
         return PrettyStatsPerMethod(

framework/helpers/highlighter.py

@@ -19,7 +19,6 @@
 TODO(sergiitk): This can be used to output protobuf responses formatted as JSON.
 """
 import logging
-from typing import Optional
 
 from absl import flags
 import pygments
@@ -63,14 +62,14 @@ class Highlighter:
     formatter: Formatter
     lexer: Lexer
     color: bool
-    color_style: Optional[str] = None
+    color_style: str | None = None
 
     def __init__(
         self,
         *,
         lexer: Lexer,
-        color: Optional[bool] = None,
-        color_style: Optional[str] = None,
+        color: bool | None = None,
+        color_style: str | None = None,
     ):
         self.lexer = lexer
         self.color = color if color is not None else COLOR.value
@@ -97,7 +96,7 @@ def highlight(self, code: str) -> str:
 
 class HighlighterYaml(Highlighter):
     def __init__(
-        self, *, color: Optional[bool] = None, color_style: Optional[str] = None
+        self, *, color: bool | None = None, color_style: str | None = None
     ):
         super().__init__(
             lexer=YamlLexer(encoding="utf-8"),

framework/helpers/retryers.py

@@ -22,7 +22,7 @@
 """
 import datetime
 import logging
-from typing import Any, Callable, List, Optional, Tuple, Type
+from typing import Any, Callable, List, Tuple, Type
 
 import tenacity
 from tenacity import _utils as tenacity_utils
@@ -41,8 +41,8 @@
 
 def _build_retry_conditions(
     *,
-    retry_on_exceptions: Optional[_ExceptionClasses] = None,
-    check_result: Optional[CheckResultFn] = None,
+    retry_on_exceptions: _ExceptionClasses | None = None,
+    check_result: CheckResultFn | None = None,
 ) -> List[retry_base]:
     # Retry on all exceptions by default
     if retry_on_exceptions is None:
@@ -63,10 +63,10 @@ def exponential_retryer_with_timeout(
     wait_min: timedelta,
     wait_max: timedelta,
     timeout: timedelta,
-    retry_on_exceptions: Optional[_ExceptionClasses] = None,
-    check_result: Optional[CheckResultFn] = None,
-    logger: Optional[logging.Logger] = None,
-    log_level: Optional[int] = logging.DEBUG,
+    retry_on_exceptions: _ExceptionClasses | None = None,
+    check_result: CheckResultFn | None = None,
+    logger: logging.Logger | None = None,
+    log_level: int | None = logging.DEBUG,
     error_note: str = "",
 ) -> Retrying:
     if logger is None:
@@ -95,11 +95,11 @@ def constant_retryer(
     *,
     wait_fixed: timedelta,
     attempts: int = 0,
-    timeout: Optional[timedelta] = None,
-    retry_on_exceptions: Optional[_ExceptionClasses] = None,
-    check_result: Optional[CheckResultFn] = None,
-    logger: Optional[logging.Logger] = None,
-    log_level: Optional[int] = logging.DEBUG,
+    timeout: timedelta | None = None,
+    retry_on_exceptions: _ExceptionClasses | None = None,
+    check_result: CheckResultFn | None = None,
+    logger: logging.Logger | None = None,
+    log_level: int | None = logging.DEBUG,
     error_note: str = "",
 ) -> Retrying:
     if logger is None:
@@ -134,9 +134,9 @@ def constant_retryer(
 
 def _on_error_callback(
     *,
-    timeout: Optional[timedelta] = None,
+    timeout: timedelta | None = None,
     attempts: int = 0,
-    check_result: Optional[CheckResultFn] = None,
+    check_result: CheckResultFn | None = None,
     error_note: str = "",
 ):
     """A helper to propagate the initial state to the RetryError, so that
@@ -234,9 +234,9 @@ def __init__(
         self,
         retry_state,
         *,
-        timeout: Optional[timedelta] = None,
+        timeout: timedelta | None = None,
         attempts: int = 0,
-        check_result: Optional[CheckResultFn] = None,
+        check_result: CheckResultFn | None = None,
         note: str = "",
     ):
         last_attempt: tenacity.Future = retry_state.outcome
@@ -292,7 +292,7 @@ def reason_str(self):
         return self.exception_str() if self.exception() else self.result_str()
 
     @classmethod
-    def _exception_str(cls, err: Optional[BaseException]) -> str:
+    def _exception_str(cls, err: BaseException | None) -> str:
         return f"{type(err).__name__}: {err}" if err else "???"
 
     # TODO(sergiitk): Remove in py3.11, this will be built-in. See PEP 678.

framework/helpers/skips.py

@@ -16,7 +16,6 @@
 import enum
 import logging
 import re
-from typing import Optional
 
 from packaging import version as pkg_version
 
@@ -51,7 +50,7 @@ class TestConfig:
 
     client_lang: Lang
     server_lang: Lang
-    version: Optional[str]
+    version: str | None
 
     def version_gte(self, another: str) -> bool:
         """Returns a bool for whether this VERSION is >= then ANOTHER version.

framework/infrastructure/gcp/api.py

@@ -16,7 +16,7 @@
 import functools
 import json
 import logging
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List
 
 from absl import flags
 from google.cloud import secretmanager_v1
@@ -88,7 +88,7 @@
 
 class GcpApiManager:
     # The previous value of googleapiclient.model.dump_request_response.
-    _dump_req_resp: Optional[bool] = None
+    _dump_req_resp: bool | None = None
 
     def __init__(
         self,
@@ -263,8 +263,8 @@ def _build_from_discovery_v2(
         api_name,
         version,
         *,
-        api_key: Optional[str] = None,
-        visibility_labels: Optional[List] = None,
+        api_key: str | None = None,
+        visibility_labels: List | None = None,
     ):
         params = {}
         if api_key:
@@ -305,8 +305,8 @@ class ResponseError(Error):
 
     reason: str
     uri: str
-    error_details: Optional[str]
-    status: Optional[int]
+    error_details: str | None
+    status: int | None
     cause: _HttpError
 
     def __init__(self, cause: _HttpError):
@@ -441,7 +441,7 @@ def _execute(
         self,
         request: HttpRequest,
         *,
-        num_retries: Optional[int] = _GCP_API_RETRIES,
+        num_retries: int | None = _GCP_API_RETRIES,
     ) -> Dict[str, Any]:
         """Execute the immediate request.
 
@@ -515,7 +515,7 @@ def wait_for_operation(
 class GcpStandardCloudApiResource(GcpProjectApiResource, metaclass=abc.ABCMeta):
     GLOBAL_LOCATION = "global"
 
-    def parent(self, location: Optional[str] = GLOBAL_LOCATION):
+    def parent(self, location: str | None = GLOBAL_LOCATION):
         if location is None:
             location = self.GLOBAL_LOCATION
         return f"projects/{self.project}/locations/{location}"

framework/infrastructure/gcp/compute.py

@@ -15,7 +15,7 @@
 import datetime
 import enum
 import logging
-from typing import Any, Dict, List, Optional, Set
+from typing import Any, Dict, List, Set
 
 from googleapiclient import discovery
 import googleapiclient.errors
@@ -38,7 +38,7 @@ class ComputeV1(
     _WAIT_FOR_BACKEND_SEC = 60 * 10
     _WAIT_FOR_BACKEND_SLEEP_SEC = 4
     _WAIT_FOR_OPERATION_SEC = 60 * 10
-    gfe_debug_header: Optional[str]
+    gfe_debug_header: str | None
 
     @dataclasses.dataclass(frozen=True)
     class GcpResource:
@@ -60,7 +60,7 @@ def __init__(
         self,
         api_manager: gcp.api.GcpApiManager,
         project: str,
-        gfe_debug_header: Optional[str] = None,
+        gfe_debug_header: str | None = None,
         version: str = "v1",
     ):
         super().__init__(api_manager.compute(version), project)
@@ -80,7 +80,7 @@ def create_health_check(
         name: str,
         protocol: HealthCheckProtocol,
         *,
-        port: Optional[int] = None,
+        port: int | None = None,
     ) -> "GcpResource":
         if protocol is self.HealthCheckProtocol.TCP:
             health_check_field = "tcpHealthCheck"
@@ -117,7 +117,7 @@ def create_firewall_rule(
         network_url: str,
         source_ranges: List[str],
         ports: List[str],
-    ) -> Optional["GcpResource"]:
+    ) -> "GcpResource" | None:
         try:
             return self._insert_resource(
                 self.api.firewalls(),
@@ -146,11 +146,11 @@ def create_backend_service_traffic_director(
         self,
         name: str,
         health_check: "GcpResource",
-        affinity_header: Optional[str] = None,
-        protocol: Optional[BackendServiceProtocol] = None,
-        subset_size: Optional[int] = None,
-        locality_lb_policies: Optional[List[dict]] = None,
-        outlier_detection: Optional[dict] = None,
+        affinity_header: str | None = None,
+        protocol: BackendServiceProtocol | None = None,
+        subset_size: int | None = None,
+        locality_lb_policies: List[dict] | None = None,
+        outlier_detection: dict | None = None,
         enable_dualstack: bool = False,
     ) -> "GcpResource":
         if not isinstance(protocol, self.BackendServiceProtocol):
@@ -201,9 +201,9 @@ def backend_service_patch_backends(
         self,
         backend_service,
         backends,
-        max_rate_per_endpoint: Optional[int] = None,
+        max_rate_per_endpoint: int | None = None,
         *,
-        circuit_breakers: Optional[dict[str, int]] = None,
+        circuit_breakers: dict[str, int] | None = None,
     ):
         if max_rate_per_endpoint is None:
             max_rate_per_endpoint = 5
@@ -244,7 +244,7 @@ def create_url_map(
         matcher_name: str,
         src_hosts,
         dst_default_backend_service: "GcpResource",
-        dst_host_rule_match_backend_service: Optional["GcpResource"] = None,
+        dst_host_rule_match_backend_service: "GcpResource" | None = None,
     ) -> "GcpResource":
         if dst_host_rule_match_backend_service is None:
             dst_host_rule_match_backend_service = dst_default_backend_service