[internal] Run pyupgrade src/python/pants/[engine,help,init] (#13183)

src/python/pants/engine/console.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 import sys
-from typing import Callable, Optional, TextIO
+from typing import Callable, TextIO
 
 from colors import blue, cyan, green, magenta, red, yellow
 
@@ -23,7 +23,7 @@ def __init__(
         stdout: TextIO | None = None,
         stderr: TextIO | None = None,
         use_colors: bool = True,
-        session: Optional[SchedulerSession] = None,
+        session: SchedulerSession | None = None,
     ):
         """If a SchedulerSession is set, any running UI will be torn down before stdio is
         rendered."""

src/python/pants/engine/internals/graph.py

@@ -9,7 +9,7 @@
 import os.path
 from dataclasses import dataclass
 from pathlib import PurePath
-from typing import Dict, Iterable, List, NamedTuple, Optional, Sequence, Set, Tuple, Type, cast
+from typing import Iterable, NamedTuple, Sequence, cast
 
 from pants.base.deprecated import warn_or_error
 from pants.base.exceptions import ResolveError
@@ -219,7 +219,7 @@ async def resolve_targets(
 
 
 class CycleException(Exception):
-    def __init__(self, subject: Address, path: Tuple[Address, ...]) -> None:
+    def __init__(self, subject: Address, path: tuple[Address, ...]) -> None:
         path_string = "\n".join((f"-> {a}" if a == subject else f"   {a}") for a in path)
         super().__init__(
             f"The dependency graph contained a cycle:\n{path_string}\n\nTo fix this, first verify "
@@ -237,10 +237,10 @@ def __init__(self, subject: Address, path: Tuple[Address, ...]) -> None:
 
 
 def _detect_cycles(
-    roots: Tuple[Address, ...], dependency_mapping: Dict[Address, Tuple[Address, ...]]
+    roots: tuple[Address, ...], dependency_mapping: dict[Address, tuple[Address, ...]]
 ) -> None:
     path_stack: OrderedSet[Address] = OrderedSet()
-    visited: Set[Address] = set()
+    visited: set[Address] = set()
 
     def maybe_report_cycle(address: Address) -> None:
         # NB: File-level dependencies are cycle tolerant.
@@ -293,7 +293,7 @@ class _DependencyMappingRequest:
 
 @dataclass(frozen=True)
 class _DependencyMapping:
-    mapping: FrozenDict[Address, Tuple[Address, ...]]
+    mapping: FrozenDict[Address, tuple[Address, ...]]
     visited: FrozenOrderedSet[Target]
     roots_as_targets: Collection[Target]
 
@@ -308,7 +308,7 @@ async def transitive_dependency_mapping(request: _DependencyMappingRequest) -> _
     roots_as_targets = await Get(UnexpandedTargets, Addresses(request.tt_request.roots))
     visited: OrderedSet[Target] = OrderedSet()
     queued = FrozenOrderedSet(roots_as_targets)
-    dependency_mapping: dict[Address, Tuple[Address, ...]] = {}
+    dependency_mapping: dict[Address, tuple[Address, ...]] = {}
     while queued:
         direct_dependencies: tuple[Collection[Target], ...]
         if request.expanded_targets:
@@ -456,7 +456,7 @@ class InvalidOwnersOfArgs(Exception):
 class OwnersRequest:
     """A request for the owners of a set of file paths."""
 
-    sources: Tuple[str, ...]
+    sources: tuple[str, ...]
     owners_not_found_behavior: OwnersNotFoundBehavior = OwnersNotFoundBehavior.ignore
 
 
@@ -546,7 +546,7 @@ def extract_owners_not_found_behavior(global_options: GlobalOptions) -> OwnersNo
 def _log_or_raise_unmatched_owners(
     file_paths: Sequence[PurePath],
     owners_not_found_behavior: OwnersNotFoundBehavior,
-    ignore_option: Optional[str] = None,
+    ignore_option: str | None = None,
 ) -> None:
     option_msg = (
         f"\n\nIf you would like to ignore un-owned files, please pass `{ignore_option}`."
@@ -599,7 +599,7 @@ async def addresses_from_filesystem_specs(
     owners_per_include = await MultiGet(
         Get(Owners, OwnersRequest(sources=paths.files)) for paths in paths_per_include
     )
-    addresses: Set[Address] = set()
+    addresses: set[Address] = set()
     for spec, owners in zip(filesystem_specs.includes, owners_per_include):
         if (
             owners_not_found_behavior != OwnersNotFoundBehavior.ignore
@@ -684,12 +684,12 @@ class AmbiguousCodegenImplementationsException(Exception):
 
     def __init__(
         self,
-        generators: Iterable[Type["GenerateSourcesRequest"]],
+        generators: Iterable[type[GenerateSourcesRequest]],
         *,
-        for_sources_types: Iterable[Type["Sources"]],
+        for_sources_types: Iterable[type[Sources]],
     ) -> None:
         all_same_generator_paths = (
-            len(set((generator.input, generator.output) for generator in generators)) == 1
+            len({(generator.input, generator.output) for generator in generators}) == 1
         )
         example_generator = list(generators)[0]
         input = example_generator.input.__name__
@@ -747,7 +747,7 @@ async def hydrate_sources(
 
     # Now, determine if any of the `for_sources_types` may be used, either because the
     # sources_field is a direct subclass or can be generated into one of the valid types.
-    def compatible_with_sources_field(valid_type: Type[Sources]) -> bool:
+    def compatible_with_sources_field(valid_type: type[Sources]) -> bool:
         is_instance = isinstance(sources_field, valid_type)
         can_be_generated = (
             request.enable_codegen
@@ -813,8 +813,8 @@ def extract_subproject_roots(global_options: GlobalOptions) -> SubprojectRoots:
 
 
 class ParsedDependencies(NamedTuple):
-    addresses: List[AddressInput]
-    ignored_addresses: List[AddressInput]
+    addresses: list[AddressInput]
+    ignored_addresses: list[AddressInput]
 
 
 class TransitiveExcludesNotSupportedError(ValueError):
@@ -857,8 +857,8 @@ async def determine_explicitly_provided_dependencies(
         subproject_roots=subproject_roots,
     )
 
-    addresses: List[AddressInput] = []
-    ignored_addresses: List[AddressInput] = []
+    addresses: list[AddressInput] = []
+    ignored_addresses: list[AddressInput] = []
     for v in request.field.value or ():
         is_ignore = v.startswith("!")
         if is_ignore:
@@ -912,7 +912,7 @@ async def resolve_dependencies(
 
     # Infer any dependencies (based on `Sources` field).
     inference_request_types = union_membership.get(InferDependenciesRequest)
-    inferred: Tuple[InferredDependencies, ...] = ()
+    inferred: tuple[InferredDependencies, ...] = ()
     if inference_request_types:
         sources_field = tgt.get(Sources)
         relevant_inference_request_types = [
@@ -942,7 +942,7 @@ async def resolve_dependencies(
     # `files` and `packages` fields, then we possibly include those too. We don't want to always
     # include those dependencies because they should often be excluded from the result due to
     # being handled elsewhere in the calling code.
-    special_cased: Tuple[Address, ...] = ()
+    special_cased: tuple[Address, ...] = ()
     if request.include_special_cased_deps:
         # Unlike normal, we don't use `tgt.get()` because there may be >1 subclass of
         # SpecialCasedDependencies.
@@ -1009,7 +1009,7 @@ def __init__(
         specs: Specs,
         union_membership: UnionMembership,
         *,
-        applicable_target_types: Iterable[Type[Target]],
+        applicable_target_types: Iterable[type[Target]],
         goal_description: str,
     ) -> None:
         applicable_target_aliases = sorted(

src/python/pants/engine/internals/mapper.py

@@ -5,7 +5,7 @@
 
 import re
 from dataclasses import dataclass
-from typing import Callable, Dict, Iterable, Mapping, Optional, Pattern, Tuple
+from typing import Callable, Iterable, Mapping, Pattern
 
 from pants.base.exceptions import MappingError
 from pants.build_graph.address import Address, BuildFileAddress
@@ -25,7 +25,7 @@ class AddressMap:
     """Maps target adaptors from a byte source."""
 
     path: str
-    name_to_target_adaptor: Dict[str, TargetAdaptor]
+    name_to_target_adaptor: dict[str, TargetAdaptor]
 
     @classmethod
     def parse(
@@ -44,7 +44,7 @@ def parse(
             target_adaptors = parser.parse(filepath, build_file_content, extra_symbols)
         except Exception as e:
             raise MappingError(f"Failed to parse ./{filepath}:\n{e}")
-        name_to_target_adaptors: Dict[str, TargetAdaptor] = {}
+        name_to_target_adaptors: dict[str, TargetAdaptor] = {}
         for target_adaptor in target_adaptors:
             name = target_adaptor.name
             if name in name_to_target_adaptors:
@@ -78,7 +78,7 @@ class AddressFamily:
 
     # The directory from which the adaptors were parsed.
     namespace: str
-    name_to_target_adaptors: Dict[str, Tuple[str, TargetAdaptor]]
+    name_to_target_adaptors: dict[str, tuple[str, TargetAdaptor]]
 
     @classmethod
     def create(cls, spec_path: str, address_maps: Iterable[AddressMap]) -> AddressFamily:
@@ -97,7 +97,7 @@ def create(cls, spec_path: str, address_maps: Iterable[AddressMap]) -> AddressFa
                     f"but received: {address_map.path!r}"
                 )
 
-        name_to_target_adaptors: Dict[str, Tuple[str, TargetAdaptor]] = {}
+        name_to_target_adaptors: dict[str, tuple[str, TargetAdaptor]] = {}
         for address_map in address_maps:
             for name, target_adaptor in address_map.name_to_target_adaptor.items():
                 if name in name_to_target_adaptors:
@@ -121,7 +121,7 @@ def addresses_to_target_adaptors(self) -> Mapping[Address, TargetAdaptor]:
         }
 
     @memoized_property
-    def build_file_addresses(self) -> Tuple[BuildFileAddress, ...]:
+    def build_file_addresses(self) -> tuple[BuildFileAddress, ...]:
         return tuple(
             BuildFileAddress(
                 rel_path=path, address=Address(spec_path=self.namespace, target_name=name)
@@ -130,10 +130,10 @@ def build_file_addresses(self) -> Tuple[BuildFileAddress, ...]:
         )
 
     @property
-    def target_names(self) -> Tuple[str, ...]:
+    def target_names(self) -> tuple[str, ...]:
         return tuple(addr.target_name for addr in self.addresses_to_target_adaptors)
 
-    def get_target_adaptor(self, address: Address) -> Optional[TargetAdaptor]:
+    def get_target_adaptor(self, address: Address) -> TargetAdaptor | None:
         assert address.spec_path == self.namespace
         entry = self.name_to_target_adaptors.get(address.target_name)
         if entry is None:
@@ -156,20 +156,20 @@ def __repr__(self) -> str:
 class AddressSpecsFilter:
     """Filters addresses with the `--tags` and `--exclude-target-regexp` options."""
 
-    tags: Tuple[str, ...]
-    exclude_target_regexps: Tuple[str, ...]
+    tags: tuple[str, ...]
+    exclude_target_regexps: tuple[str, ...]
 
     def __init__(
         self,
         *,
-        tags: Optional[Iterable[str]] = None,
-        exclude_target_regexps: Optional[Iterable[str]] = None,
+        tags: Iterable[str] | None = None,
+        exclude_target_regexps: Iterable[str] | None = None,
     ) -> None:
         self.tags = tuple(tags or [])
         self.exclude_target_regexps = tuple(exclude_target_regexps or [])
 
     @memoized_property
-    def _exclude_regexps(self) -> Tuple[Pattern, ...]:
+    def _exclude_regexps(self) -> tuple[Pattern, ...]:
         return tuple(re.compile(pattern) for pattern in self.exclude_target_regexps)
 
     def _is_excluded_by_pattern(self, address: Address) -> bool:

src/python/pants/engine/internals/scheduler.py

@@ -9,7 +9,7 @@
 from dataclasses import dataclass
 from pathlib import PurePath
 from types import CoroutineType
-from typing import Any, Dict, Iterable, List, NoReturn, Optional, Sequence, Tuple, Type, cast
+from typing import Any, Dict, Iterable, NoReturn, Sequence, cast
 
 from typing_extensions import TypedDict
 
@@ -76,8 +76,8 @@
 
 
 class PolledWorkunits(TypedDict):
-    started: Tuple[Workunit, ...]
-    completed: Tuple[Workunit, ...]
+    started: tuple[Workunit, ...]
+    completed: tuple[Workunit, ...]
 
 
 @dataclass(frozen=True)
@@ -105,19 +105,19 @@ class Scheduler:
     def __init__(
         self,
         *,
-        ignore_patterns: List[str],
+        ignore_patterns: list[str],
         use_gitignore: bool,
         build_root: str,
         local_execution_root_dir: str,
         named_caches_dir: str,
-        ca_certs_path: Optional[str],
+        ca_certs_path: str | None,
         rules: Iterable[Rule],
         union_membership: UnionMembership,
         execution_options: ExecutionOptions,
         local_store_options: LocalStoreOptions,
         executor: PyExecutor,
         include_trace_on_error: bool = True,
-        visualize_to_dir: Optional[str] = None,
+        visualize_to_dir: str | None = None,
         validate_reachability: bool = True,
         watch_filesystem: bool = True,
     ) -> None:
@@ -265,12 +265,12 @@ def rule_graph_visualization(self):
     def rule_subgraph_visualization(self, root_subject_types: list[type], product_type: type):
         with temporary_file_path() as path:
             self.visualize_rule_subgraph_to_file(path, root_subject_types, product_type)
-            with open(path, "r") as fd:
+            with open(path) as fd:
                 for line in fd.readlines():
                     yield line.rstrip()
 
     def rule_graph_consumed_types(
-        self, root_subject_types: Sequence[Type], product_type: Type
+        self, root_subject_types: Sequence[type], product_type: type
     ) -> Sequence[type]:
         return native_engine.rule_graph_consumed_types(
             self.py_scheduler, root_subject_types, product_type
@@ -611,7 +611,7 @@ def run_local_interactive_process(
             self.py_scheduler, self.py_session, request
         )
 
-    def write_digest(self, digest: Digest, *, path_prefix: Optional[str] = None) -> None:
+    def write_digest(self, digest: Digest, *, path_prefix: str | None = None) -> None:
         """Write a digest to disk, relative to the build root."""
         if path_prefix and PurePath(path_prefix).is_absolute():
             raise ValueError(