Replace Optional: part 1, no nesting

bin/cleanup/cleanup.py

@@ -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

@@ -127,7 +127,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 +165,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

@@ -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

@@ -31,7 +31,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 +77,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

@@ -63,14 +63,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 +97,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

@@ -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

@@ -51,7 +51,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

@@ -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

@@ -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,
+        affinity_header: str | None = None,
+        protocol: BackendServiceProtocol | None = None,
+        subset_size: int | None = None,
         locality_lb_policies: Optional[List[dict]] = None,
-        outlier_detection: Optional[dict] = None,
+        outlier_detection: dict | None = None,
         enable_dualstack: bool = False,
     ) -> "GcpResource":
         if not isinstance(protocol, self.BackendServiceProtocol):
@@ -201,7 +201,7 @@ 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,
     ):
@@ -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

framework/infrastructure/gcp/iam.py

@@ -131,7 +131,7 @@ class Binding:
 
         role: str
         members: FrozenSet[str]
-        condition: Optional[Expr] = None
+        condition: Expr | None = None
 
         @classmethod
         def from_response(cls, response: Dict[str, Any]) -> "Policy.Binding":
@@ -155,12 +155,12 @@ def as_dict(self) -> Dict[str, Any]:
 
     bindings: FrozenSet[Binding]
     etag: str
-    version: Optional[int] = None
+    version: int | None = None
 
     @functools.lru_cache(maxsize=128)
     def find_binding_for_role(
-        self, role: str, condition: Optional[Expr] = None
-    ) -> Optional["Policy.Binding"]:
+        self, role: str, condition: Expr | None = None
+    ) -> "Policy.Binding" | None:
         results = (
             binding
             for binding in self.bindings
@@ -283,7 +283,7 @@ def add_service_account_iam_policy_binding(
         https://cloud.google.com/iam/docs/reference/rest/v1/projects.serviceAccounts/setIamPolicy
         """
         policy: Policy = self.get_service_account_iam_policy(account)
-        binding: Optional[Policy.Binding] = policy.find_binding_for_role(role)
+        binding: Policy.Binding | None = policy.find_binding_for_role(role)
         if binding and member in binding.members:
             logger.debug(
                 "Member %s already has role %s for Service Account %s",
@@ -325,7 +325,7 @@ def remove_service_account_iam_policy_binding(
         https://cloud.google.com/iam/docs/reference/rest/v1/projects.serviceAccounts/setIamPolicy
         """
         policy: Policy = self.get_service_account_iam_policy(account)
-        binding: Optional[Policy.Binding] = policy.find_binding_for_role(role)
+        binding: Policy.Binding | None = policy.find_binding_for_role(role)
 
         if binding is None:
             logger.debug(