Replace Optional with | None

Author: sergiitk

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:

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

@@ -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,7 +63,7 @@ class Highlighter:
     formatter: Formatter
     lexer: Lexer
     color: bool
-    color_style: Optional[str] = None
+    color_style: str | None = None
 
     def __init__(
         self,

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,7 +134,7 @@ def constant_retryer(
 
 def _on_error_callback(
     *,
-    timeout: Optional[timedelta] = None,
+    timeout: timedelta | None = None,
     attempts: int = 0,
     check_result: Optional[CheckResultFn] = None,
     error_note: str = "",

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

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

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":

framework/infrastructure/gcp/network_services.py

@@ -37,8 +37,8 @@ class EndpointPolicy:
     endpoint_matcher: dict
     update_time: str
     create_time: str
-    http_filters: Optional[dict] = None
-    server_tls_policy: Optional[str] = None
+    http_filters: dict | None = None
+    server_tls_policy: str | None = None
 
     @classmethod
     def from_response(