Replace 'Optional[...]' by '|' on type hints.

PiperOrigin-RevId: 658178117

Author: dorellang

perfkitbenchmarker/app_service.py

@@ -3,7 +3,7 @@
 import logging
 import threading
 import time
-from typing import Any, Dict, List, Optional, Type, TypeVar
+from typing import Any, Dict, List, Type, TypeVar
 
 from absl import flags
 from perfkitbenchmarker import errors
@@ -108,7 +108,7 @@ def _GetOptionDecoderConstructions(cls) -> Dict[str, Any]:
 AppServiceChild = TypeVar('AppServiceChild', bound='BaseAppService')
 
 
-def GetAppServiceClass(service: str) -> Optional[Type[AppServiceChild]]:
+def GetAppServiceClass(service: str) -> Type[AppServiceChild] | None:
   """Returns class of the given AppService.
 
   Args:
@@ -150,7 +150,7 @@ def __init__(self, base_app_service_spec: BaseAppServiceSpec):
     self.region: str = base_app_service_spec.appservice_region
     self.backend: str = base_app_service_spec.appservice_backend
     self.builder: Any = None
-    self.endpoint: Optional[str] = None
+    self.endpoint: str | None = None
     self.poller: http_poller.HttpPoller = http_poller.HttpPoller()
     self.vm: virtual_machine.BaseVirtualMachine
     # update metadata

perfkitbenchmarker/cloud_harmony_util.py

@@ -4,7 +4,7 @@
 """
 
 import io
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List
 
 from absl import flags
 import pandas as pd
@@ -130,7 +130,7 @@ def ParseCsvResultsFromString(
   return results
 
 
-def GetCommonMetadata(custom_metadata: Optional[Dict[str, Any]] = None) -> str:
+def GetCommonMetadata(custom_metadata: Dict[str, Any] | None = None) -> str:
   """Returns pkb metadata associated with this run as cloudharmony metadata.
 
   Cloudharmony benchmarks take in benchmark setup configurations as inputs and

perfkitbenchmarker/configs/auto_registry.py

@@ -15,7 +15,7 @@
 import itertools
 import logging
 import typing
-from typing import Any, Optional
+from typing import Any
 from perfkitbenchmarker import errors
 
 
@@ -26,7 +26,7 @@
 def GetRegisteredClass(
     registry: dict[tuple[Any], type[T]],
     base_class: type[T],
-    default_class: Optional[type[T]] = None,
+    default_class: type[T] | None = None,
     **kwargs
 ) -> type[T]:
   """Returns the subclass with the corresponding attributes.
@@ -79,8 +79,8 @@ def GetRegisteredClass(
 def RegisterClass(
     registry: dict[tuple[Any], type[T]],
     cls: type[T],
-    required_attrs: Optional[list[str]],
-    cls_type: Optional[str],
+    required_attrs: list[str] | None,
+    cls_type: str | None,
 ):
   """Adds the class with its attributes to the registry.
 

perfkitbenchmarker/configs/container_spec.py

@@ -17,7 +17,7 @@
 configuration files.
 """
 
-from typing import Any, Optional
+from typing import Any
 
 from absl import flags
 from perfkitbenchmarker import custom_virtual_machine_spec
@@ -40,7 +40,7 @@ class ContainerSpec(spec.BaseSpec):
   def __init__(
       self,
       component_full_name: str,
-      flag_values: Optional[flags.FlagValues] = None,
+      flag_values: flags.FlagValues | None = None,
       **kwargs: Any,
   ):
     super().__init__(component_full_name, flag_values, **kwargs)
@@ -107,16 +107,16 @@ class ContainerRegistrySpec(spec.BaseSpec):
   def __init__(
       self,
       component_full_name: str,
-      flag_values: Optional[flags.FlagValues] = None,
+      flag_values: flags.FlagValues | None = None,
       **kwargs: Any,
   ):
     super().__init__(component_full_name, flag_values=flag_values, **kwargs)
-    self.spec: Optional[dict[str, Any]]
+    self.spec: dict[str, Any] | None
     self.cloud: str
     registry_spec = getattr(self.spec, self.cloud, {})
-    self.project: Optional[str] = registry_spec.get('project')
-    self.zone: Optional[str] = registry_spec.get('zone')
-    self.name: Optional[str] = registry_spec.get('name')
+    self.project: str | None = registry_spec.get('project')
+    self.zone: str | None = registry_spec.get('zone')
+    self.name: str | None = registry_spec.get('name')
     self.cpus: float
     self.memory: int
     self.command: list[str]
@@ -240,7 +240,7 @@ def __init__(
     )
     self.vm_count: int
     self.vm_spec: spec.PerCloudConfigSpec
-    self.sandbox_config: Optional[SandboxSpec]
+    self.sandbox_config: SandboxSpec | None
 
   @classmethod
   def _GetOptionDecoderConstructions(cls):

perfkitbenchmarker/configs/freeze_restore_spec.py

@@ -13,8 +13,6 @@
 # limitations under the License.
 """Benchmark-configurable options for freezing and restoring resources."""
 
-from typing import Optional
-
 from absl import flags
 from perfkitbenchmarker.configs import option_decoders
 from perfkitbenchmarker.configs import spec
@@ -42,7 +40,7 @@ class FreezeRestoreSpec(spec.BaseSpec):
   def __init__(
       self,
       component_full_name: str,
-      flag_values: Optional[flags.FlagValues] = None,
+      flag_values: flags.FlagValues | None = None,
       **kwargs
   ):
     super().__init__(component_full_name, flag_values=flag_values, **kwargs)

perfkitbenchmarker/configs/spec.py

@@ -16,7 +16,7 @@
 import collections
 import logging
 import threading
-from typing import Any, Optional
+from typing import Any
 
 from absl import flags
 from perfkitbenchmarker import errors
@@ -77,7 +77,7 @@ class BaseSpec(six.with_metaclass(BaseSpecMetaClass, object)):
   def __init__(
       self,
       component_full_name: str,
-      flag_values: Optional[flags.FlagValues] = None,
+      flag_values: flags.FlagValues | None = None,
       **kwargs: Any,
   ):
     """Initializes a BaseSpec.
@@ -200,7 +200,7 @@ def _DecodeAndInit(
       decoders: collections.OrderedDict[
           str, option_decoders.ConfigOptionDecoder
       ],
-      flag_values: Optional[flags.FlagValues],
+      flag_values: flags.FlagValues | None,
   ):
     """Initializes spec attributes from provided config option values.
 

perfkitbenchmarker/configs/vm_group_decoders.py

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 """Module containing a VM group spec and related decoders."""
-from typing import Optional
+
 from perfkitbenchmarker import disk
 from perfkitbenchmarker import errors
 from perfkitbenchmarker import os_types
@@ -56,7 +56,7 @@ class VmGroupSpec(spec.BaseSpec):
   vm_count: int
   vm_spec: virtual_machine.BaseVmSpec
   vm_as_nfs: bool
-  vm_as_nfs_disk_spec: Optional[disk.BaseNFSDiskSpec]
+  vm_as_nfs_disk_spec: disk.BaseNFSDiskSpec | None
   placement_group_name: str
   cidr: str
 

perfkitbenchmarker/container_service.py

@@ -31,7 +31,7 @@
 import os
 import re
 import time
-from typing import Any, Optional, Sequence
+from typing import Any, Sequence
 
 from absl import flags
 import jinja2
@@ -166,7 +166,7 @@ class BaseContainer(resource.BaseResource):
   """Class representing a single container."""
 
   def __init__(
-      self, container_spec: Optional[container_spec_lib.ContainerSpec] = None
+      self, container_spec: container_spec_lib.ContainerSpec | None = None
   ):
     # Hack to make container_spec a kwarg
     assert container_spec
@@ -175,7 +175,7 @@ def __init__(
     self.memory: int = container_spec.memory
     self.command: list[str] = container_spec.command
     self.image: str = container_spec.image
-    self.ip_address: Optional[str] = None
+    self.ip_address: str | None = None
 
   def WaitForExit(self, timeout: int = 1200) -> dict[str, Any]:
     """Gets the successfully finished container.
@@ -204,9 +204,9 @@ def __init__(self, container_spec: container_spec_lib.ContainerSpec):
     self.command: list[str] = container_spec.command
     self.image: str = container_spec.image
     self.container_port: int = container_spec.container_port
-    self.ip_address: Optional[str] = None
-    self.port: Optional[int] = None
-    self.host_header: Optional[str] = None
+    self.ip_address: str | None = None
+    self.port: int | None = None
+    self.host_header: str | None = None
 
 
 class ContainerImage:
@@ -396,14 +396,14 @@ def __init__(self, vm_config: virtual_machine.BaseVirtualMachine, name: str):
     self.machine_type = vm_config.machine_type
     self.name = NodePoolName(name)
     self.zone: str = vm_config.zone
-    self.sandbox_config: Optional[container_spec_lib.SandboxSpec] = None
+    self.sandbox_config: container_spec_lib.SandboxSpec | None = None
     self.num_nodes: int
     self.disk_type: str
     self.disk_size: int
     # Defined by google_kubernetes_engine
-    self.max_local_disks: Optional[int]
-    self.gpu_type: Optional[str]
-    self.gpu_count: Optional[int]
+    self.max_local_disks: int | None
+    self.gpu_type: str | None
+    self.gpu_count: int | None
     self.threads_per_core: int
     self.gce_tags: list[str]
     self.min_cpu_platform: str
@@ -631,7 +631,7 @@ def _GetPod(self) -> dict[str, Any]:
     self.ip_address = pod.get('status', {}).get('podIP')
     return pod
 
-  def WaitForExit(self, timeout: Optional[int] = None) -> dict[str, Any]:
+  def WaitForExit(self, timeout: int | None = None) -> dict[str, Any]:
     """Gets the finished running container."""
 
     @vm_util.Retry(
@@ -881,7 +881,7 @@ def WaitForResource(
       self,
       resource_name: str,
       condition_name: str,
-      namespace: Optional[str] = None,
+      namespace: str | None = None,
       timeout: int = vm_util.DEFAULT_TIMEOUT,
   ):
     """Waits for a condition on a Kubernetes resource (eg: deployment, pod)."""
@@ -962,7 +962,7 @@ def CreateConfigMap(self, name: str, from_file_dir: str):
     )
 
   def CreateServiceAccount(
-      self, name: str, clusterrole: Optional[str] = None, namespace='default'
+      self, name: str, clusterrole: str | None = None, namespace='default'
   ):
     """Create a k8s service account and cluster-role-binding."""
     RunKubectlCommand(
@@ -1101,11 +1101,11 @@ class KubernetesEvent:
   resource: KubernetesEventResource
   message: str
   # Reason is actually more of a machine readable message.
-  reason: Optional[str]
+  reason: str | None
   timestamp: float
 
   @classmethod
-  def FromDict(cls, yaml_data: dict[str, Any]) -> Optional['KubernetesEvent']:
+  def FromDict(cls, yaml_data: dict[str, Any]) -> 'KubernetesEvent | None':
     """Parse Kubernetes Event YAML output."""
     if 'message' not in yaml_data:
       return

perfkitbenchmarker/disk_strategies.py

@@ -24,7 +24,7 @@
 import json
 import logging
 import time
-from typing import Any, Optional, Union
+from typing import Any, Union
 
 from absl import flags
 from perfkitbenchmarker import background_tasks
@@ -252,7 +252,7 @@ def __init__(
       self,
       vm,
       disk_spec: disk.BaseDiskSpec,
-      unmanaged_nfs_service: Optional[nfs_service.BaseNfsService] = None,
+      unmanaged_nfs_service: nfs_service.BaseNfsService | None = None,
   ):
     super().__init__(vm, disk_spec)
     if unmanaged_nfs_service: