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(

framework/infrastructure/k8s.py

@@ -733,7 +733,7 @@ def delete_pod(
         self,
         name: str,
         *,
-        grace_period: Optional[_timedelta] = DELETE_GRACE_PERIOD,
+        grace_period: _timedelta | None = DELETE_GRACE_PERIOD,
     ) -> None:
         delete_options = client.V1DeleteOptions(propagation_policy="Foreground")
 
@@ -872,8 +872,8 @@ def wait_for_service_account_deleted(
 
     def wait_for_namespace_deleted(
         self,
-        timeout_sec: Optional[int] = None,
-        wait_sec: Optional[int] = None,
+        timeout_sec: int | None = None,
+        wait_sec: int | None = None,
     ) -> None:
         if timeout_sec is None:
             if self.wait_for_namespace_deleted_timeout_sec is not None:
@@ -1076,8 +1076,8 @@ def port_forward_pod(
         self,
         pod: V1Pod,
         remote_port: int,
-        local_port: Optional[int] = None,
-        local_address: Optional[str] = None,
+        local_port: int | None = None,
+        local_address: str | None = None,
     ) -> k8s_port_forwarder.PortForwarder:
         destination = f"pod/{pod.metadata.name}"
         logger.info(
@@ -1138,7 +1138,7 @@ def pretty_format_status(
             return "No data"
 
         result = []
-        metadata: Optional[V1ObjectMeta] = None
+        metadata: V1ObjectMeta | None = None
         if isinstance(getattr(k8s_object, "metadata", None), V1ObjectMeta):
             # Parse the name if present.
             metadata: V1ObjectMeta = k8s_object.metadata

framework/infrastructure/k8s_internal/k8s_port_forwarder.py

@@ -33,7 +33,7 @@ def __init__(
         namespace: str,
         destination: str,
         remote_port: int,
-        local_port: Optional[int] = None,
+        local_port: int | None = None,
         local_address: Optional[str] = None,
     ):
         self.context = context

framework/infrastructure/traffic_director.py

@@ -88,9 +88,9 @@ class TrafficDirectorManager:  # pylint: disable=too-many-public-methods
     alternative_backends: set[NegGcpResource]
 
     # Backend Serivices
-    backend_service: Optional[GcpResource] = None
-    affinity_backend_service: Optional[GcpResource] = None
-    alternative_backend_service: Optional[GcpResource] = None
+    backend_service: GcpResource | None = None
+    affinity_backend_service: GcpResource | None = None
+    alternative_backend_service: GcpResource | None = None
 
     # TODO(sergiitk): move these flags to backend service dataclass
     backend_service_protocol: BackendServiceProtocol = _BackendUnset
@@ -127,19 +127,19 @@ def __init__(
         self.enable_dualstack: bool = enable_dualstack
 
         # Managed resources
-        self.health_check: Optional[GcpResource] = None
-        self.url_map: Optional[GcpResource] = None
-        self.alternative_url_map: Optional[GcpResource] = None
-        self.firewall_rule: Optional[GcpResource] = None
-        self.firewall_rule_ipv6: Optional[GcpResource] = None
-        self.target_proxy: Optional[GcpResource] = None
-        self.target_proxy_ipv6: Optional[GcpResource] = None
+        self.health_check: GcpResource | None = None
+        self.url_map: GcpResource | None = None
+        self.alternative_url_map: GcpResource | None = None
+        self.firewall_rule: GcpResource | None = None
+        self.firewall_rule_ipv6: GcpResource | None = None
+        self.target_proxy: GcpResource | None = None
+        self.target_proxy_ipv6: GcpResource | None = None
         # TODO(sergiitk): remove this flag once target proxy resource loaded
         self.target_proxy_is_http: bool = False
-        self.alternative_target_proxy: Optional[GcpResource] = None
-        self.forwarding_rule: Optional[GcpResource] = None
-        self.forwarding_rule_ipv6: Optional[GcpResource] = None
-        self.alternative_forwarding_rule: Optional[GcpResource] = None
+        self.alternative_target_proxy: GcpResource | None = None
+        self.forwarding_rule: GcpResource | None = None
+        self.forwarding_rule_ipv6: GcpResource | None = None
+        self.alternative_forwarding_rule: GcpResource | None = None
 
         # Backends.
         self.backends = set()
@@ -155,8 +155,8 @@ def setup_for_grpc(
         service_host,
         service_port,
         *,
-        backend_protocol: Optional[BackendServiceProtocol] = _BackendGRPC,
-        health_check_port: Optional[int] = None,
+        backend_protocol: BackendServiceProtocol | None = _BackendGRPC,
+        health_check_port: int | None = None,
     ):
         self.setup_backend_for_grpc(
             protocol=backend_protocol, health_check_port=health_check_port
@@ -166,8 +166,8 @@ def setup_for_grpc(
     def setup_backend_for_grpc(
         self,
         *,
-        protocol: Optional[BackendServiceProtocol] = _BackendGRPC,
-        health_check_port: Optional[int] = None,
+        protocol: BackendServiceProtocol | None = _BackendGRPC,
+        health_check_port: int | None = None,
     ):
         self.create_health_check(port=health_check_port)
         self.create_backend_service(protocol)
@@ -211,8 +211,8 @@ def make_resource_name(self, name: str) -> str:
     def create_health_check(
         self,
         *,
-        protocol: Optional[HealthCheckProtocol] = _HealthCheckGRPC,
-        port: Optional[int] = None,
+        protocol: HealthCheckProtocol | None = _HealthCheckGRPC,
+        port: int | None = None,
     ):
         if self.health_check:
             raise ValueError(
@@ -240,8 +240,8 @@ def delete_health_check(self, force=False):
 
     def create_backend_service(
         self,
-        protocol: Optional[BackendServiceProtocol] = _BackendGRPC,
-        subset_size: Optional[int] = None,
+        protocol: BackendServiceProtocol | None = _BackendGRPC,
+        subset_size: int | None = None,
         affinity_header: Optional[str] = None,
         locality_lb_policies: Optional[List[dict]] = None,
         outlier_detection: Optional[dict] = None,

framework/rpc/grpc_channelz.py

@@ -58,9 +58,7 @@
 class ChannelzServiceClient(framework.rpc.grpc.GrpcClientHelper):
     stub: channelz_pb2_grpc.ChannelzStub
 
-    def __init__(
-        self, channel: grpc.Channel, *, log_target: Optional[str] = ""
-    ):
+    def __init__(self, channel: grpc.Channel, *, log_target: str | None = ""):
         super().__init__(
             channel, channelz_pb2_grpc.ChannelzStub, log_target=log_target
         )

framework/rpc/grpc_csds.py

@@ -163,7 +163,7 @@ class CsdsClient(framework.rpc.grpc.GrpcClientHelper):
     DEFAULT_RPC_DEADLINE: Final[dt.timedelta] = dt.timedelta(seconds=30)
 
     def __init__(
-        self, channel: grpc.Channel, *, log_target: Optional[str] = ""
+        self, channel: grpc.Channel, *, log_target: str | None = ""
     ) -> None:
         super().__init__(channel, self.STUB_CLASS, log_target=log_target)
 

framework/rpc/grpc_testing.py

@@ -135,9 +135,7 @@ class LoadBalancerStatsServiceClient(framework.rpc.grpc.GrpcClientHelper):
     STATS_PARTIAL_RESULTS_TIMEOUT_SEC = 1200
     STATS_ACCUMULATED_RESULTS_TIMEOUT_SEC = 600
 
-    def __init__(
-        self, channel: grpc.Channel, *, log_target: Optional[str] = ""
-    ):
+    def __init__(self, channel: grpc.Channel, *, log_target: str | None = ""):
         super().__init__(
             channel,
             test_pb2_grpc.LoadBalancerStatsServiceStub,