Apply pyupgrade

Author: AlexWaygood

misc/actions_stubs.py

@@ -20,7 +20,7 @@ def apply_all(
     directory: str,
     extension: str,
     to_extension: str = "",
-    exclude: Tuple[str] = ("",),
+    exclude: tuple[str] = ("",),
     recursive: bool = True,
     debug: bool = False,
 ) -> None:
@@ -100,7 +100,7 @@ def main(
     directory: str,
     extension: str,
     to_extension: str,
-    exclude: Tuple[str],
+    exclude: tuple[str],
     not_recursive: bool,
 ) -> None:
     """

misc/analyze_cache.py

@@ -75,18 +75,18 @@ def pluck(name: str, chunks: Iterable[JsonDict]) -> Iterable[JsonDict]:
     return (chunk for chunk in chunks if chunk[".class"] == name)
 
 
-def report_counter(counter: Counter, amount: Optional[int] = None) -> None:
+def report_counter(counter: Counter, amount: int | None = None) -> None:
     for name, count in counter.most_common(amount):
         print(f"    {count: <8} {name}")
     print()
 
 
-def report_most_common(chunks: List[JsonDict], amount: Optional[int] = None) -> None:
+def report_most_common(chunks: list[JsonDict], amount: int | None = None) -> None:
     report_counter(Counter(str(chunk) for chunk in chunks), amount)
 
 
 def compress(chunk: JsonDict) -> JsonDict:
-    cache: Dict[int, JsonDict] = {}
+    cache: dict[int, JsonDict] = {}
     counter = 0
 
     def helper(chunk: Any) -> Any:
@@ -119,7 +119,7 @@ def helper(chunk: Any) -> Any:
 
 
 def decompress(chunk: JsonDict) -> JsonDict:
-    cache: Dict[int, JsonDict] = {}
+    cache: dict[int, JsonDict] = {}
 
     def helper(chunk: Any) -> Any:
         if not isinstance(chunk, dict):

misc/diff-cache.py

@@ -26,7 +26,7 @@ def make_cache(input_dir: str, sqlite: bool) -> MetadataStore:
         return FilesystemMetadataStore(input_dir)
 
 
-def merge_deps(all: Dict[str, Set[str]], new: Dict[str, Set[str]]) -> None:
+def merge_deps(all: dict[str, set[str]], new: dict[str, set[str]]) -> None:
     for k, v in new.items():
         all.setdefault(k, set()).update(v)
 
@@ -70,13 +70,13 @@ def main() -> None:
     cache1 = make_cache(args.input_dir1, args.sqlite)
     cache2 = make_cache(args.input_dir2, args.sqlite)
 
-    type_misses: Dict[str, int] = defaultdict(int)
-    type_hits: Dict[str, int] = defaultdict(int)
+    type_misses: dict[str, int] = defaultdict(int)
+    type_hits: dict[str, int] = defaultdict(int)
 
-    updates: Dict[str, Optional[str]] = {}
+    updates: dict[str, str | None] = {}
 
-    deps1: Dict[str, Set[str]] = {}
-    deps2: Dict[str, Set[str]] = {}
+    deps1: dict[str, set[str]] = {}
+    deps2: dict[str, set[str]] = {}
 
     misses = hits = 0
     cache1_all = list(cache1.list_all())

misc/dump-ast.py

@@ -15,7 +15,7 @@
 from mypy.parse import parse
 
 
-def dump(fname: str, python_version: Tuple[int, int], quiet: bool = False) -> None:
+def dump(fname: str, python_version: tuple[int, int], quiet: bool = False) -> None:
     options = Options()
     options.python_version = python_version
     with open(fname, "rb") as f:

misc/incremental_checker.py

@@ -66,7 +66,7 @@ def delete_folder(folder_path: str) -> None:
         shutil.rmtree(folder_path)
 
 
-def execute(command: List[str], fail_on_error: bool = True) -> Tuple[str, str, int]:
+def execute(command: list[str], fail_on_error: bool = True) -> tuple[str, str, int]:
     proc = subprocess.Popen(
         " ".join(command), stderr=subprocess.PIPE, stdout=subprocess.PIPE, shell=True
     )
@@ -98,7 +98,7 @@ def initialize_repo(repo_url: str, temp_repo_path: str, branch: str) -> None:
         execute(["git", "-C", temp_repo_path, "checkout", branch])
 
 
-def get_commits(repo_folder_path: str, commit_range: str) -> List[Tuple[str, str]]:
+def get_commits(repo_folder_path: str, commit_range: str) -> list[tuple[str, str]]:
     raw_data, _stderr, _errcode = execute(
         ["git", "-C", repo_folder_path, "log", "--reverse", "--oneline", commit_range]
     )
@@ -109,25 +109,25 @@ def get_commits(repo_folder_path: str, commit_range: str) -> List[Tuple[str, str
     return output
 
 
-def get_commits_starting_at(repo_folder_path: str, start_commit: str) -> List[Tuple[str, str]]:
+def get_commits_starting_at(repo_folder_path: str, start_commit: str) -> list[tuple[str, str]]:
     print(f"Fetching commits starting at {start_commit}")
     return get_commits(repo_folder_path, f"{start_commit}^..HEAD")
 
 
-def get_nth_commit(repo_folder_path: str, n: int) -> Tuple[str, str]:
+def get_nth_commit(repo_folder_path: str, n: int) -> tuple[str, str]:
     print(f"Fetching last {n} commits (or all, if there are fewer commits than n)")
     return get_commits(repo_folder_path, f"-{n}")[0]
 
 
 def run_mypy(
-    target_file_path: Optional[str],
+    target_file_path: str | None,
     mypy_cache_path: str,
-    mypy_script: Optional[str],
+    mypy_script: str | None,
     *,
     incremental: bool = False,
     daemon: bool = False,
     verbose: bool = False,
-) -> Tuple[float, str, Dict[str, Any]]:
+) -> tuple[float, str, dict[str, Any]]:
     """Runs mypy against `target_file_path` and returns what mypy prints to stdout as a string.
 
     If `incremental` is set to True, this function will use store and retrieve all caching data
@@ -136,7 +136,7 @@ def run_mypy(
 
     If `daemon` is True, we use daemon mode; the daemon must be started and stopped by the caller.
     """
-    stats: Dict[str, Any] = {}
+    stats: dict[str, Any] = {}
     if daemon:
         command = DAEMON_CMD + ["check", "-v"]
     else:
@@ -162,8 +162,8 @@ def run_mypy(
     return runtime, output, stats
 
 
-def filter_daemon_stats(output: str) -> Tuple[str, Dict[str, Any]]:
-    stats: Dict[str, Any] = {}
+def filter_daemon_stats(output: str) -> tuple[str, dict[str, Any]]:
+    stats: dict[str, Any] = {}
     lines = output.splitlines()
     output_lines = []
     for line in lines:
@@ -208,12 +208,12 @@ def save_cache(cache: JsonDict, incremental_cache_path: str = CACHE_PATH) -> Non
 
 
 def set_expected(
-    commits: List[Tuple[str, str]],
+    commits: list[tuple[str, str]],
     cache: JsonDict,
     temp_repo_path: str,
-    target_file_path: Optional[str],
+    target_file_path: str | None,
     mypy_cache_path: str,
-    mypy_script: Optional[str],
+    mypy_script: str | None,
 ) -> None:
     """Populates the given `cache` with the expected results for all of the given `commits`.
 
@@ -241,13 +241,13 @@ def set_expected(
 
 
 def test_incremental(
-    commits: List[Tuple[str, str]],
+    commits: list[tuple[str, str]],
     cache: JsonDict,
     temp_repo_path: str,
-    target_file_path: Optional[str],
+    target_file_path: str | None,
     mypy_cache_path: str,
     *,
-    mypy_script: Optional[str] = None,
+    mypy_script: str | None = None,
     daemon: bool = False,
     exit_on_error: bool = False,
 ) -> None:
@@ -258,7 +258,7 @@ def test_incremental(
     """
     print("Note: first commit is evaluated twice to warm up cache")
     commits = [commits[0]] + commits
-    overall_stats: Dict[str, float] = {}
+    overall_stats: dict[str, float] = {}
     for commit_id, message in commits:
         print(f'Now testing commit {commit_id}: "{message}"')
         execute(["git", "-C", temp_repo_path, "checkout", commit_id])
@@ -286,10 +286,10 @@ def test_incremental(
         print("Overall stats:", overall_stats)
 
 
-def combine_stats(overall_stats: Dict[str, float], new_stats: Dict[str, Any]) -> Dict[str, float]:
+def combine_stats(overall_stats: dict[str, float], new_stats: dict[str, Any]) -> dict[str, float]:
     INTERESTING_KEYS = ["build_time", "gc_time"]
     # For now, we only support float keys
-    relevant_stats: Dict[str, float] = {}
+    relevant_stats: dict[str, float] = {}
     for key in INTERESTING_KEYS:
         if key in new_stats:
             value = float(new_stats[key])
@@ -306,7 +306,7 @@ def cleanup(temp_repo_path: str, mypy_cache_path: str) -> None:
 def test_repo(
     target_repo_url: str,
     temp_repo_path: str,
-    target_file_path: Optional[str],
+    target_file_path: str | None,
     mypy_path: str,
     incremental_cache_path: str,
     mypy_cache_path: str,

misc/perf_checker.py

@@ -28,7 +28,7 @@ def delete_folder(folder_path: str) -> None:
         shutil.rmtree(folder_path)
 
 
-def execute(command: List[str]) -> None:
+def execute(command: list[str]) -> None:
     proc = subprocess.Popen(
         " ".join(command), stderr=subprocess.PIPE, stdout=subprocess.PIPE, shell=True
     )
@@ -45,7 +45,7 @@ def execute(command: List[str]) -> None:
         raise RuntimeError("Unexpected error from external tool.")
 
 
-def trial(num_trials: int, command: Command) -> List[float]:
+def trial(num_trials: int, command: Command) -> list[float]:
     trials = []
     for i in range(num_trials):
         command.setup()
@@ -56,7 +56,7 @@ def trial(num_trials: int, command: Command) -> List[float]:
     return trials
 
 
-def report(name: str, times: List[float]) -> None:
+def report(name: str, times: list[float]) -> None:
     print(f"{name}:")
     print(f"  Times: {times}")
     print(f"  Mean:  {statistics.mean(times)}")

misc/proper_plugin.py

@@ -33,7 +33,7 @@ class ProperTypePlugin(Plugin):
     all these became dangerous because typ may be e.g. an alias to union.
     """
 
-    def get_function_hook(self, fullname: str) -> Optional[Callable[[FunctionContext], Type]]:
+    def get_function_hook(self, fullname: str) -> Callable[[FunctionContext], Type] | None:
         if fullname == "builtins.isinstance":
             return isinstance_proper_hook
         if fullname == "mypy.types.get_proper_type":

misc/sync-typeshed.py

@@ -28,7 +28,7 @@ def check_state() -> None:
         sys.exit('error: Output of "git status -s mypy/typeshed" must be empty')
 
 
-def update_typeshed(typeshed_dir: str, commit: Optional[str]) -> str:
+def update_typeshed(typeshed_dir: str, commit: str | None) -> str:
     """Update contents of local typeshed copy.
 
     Return the normalized typeshed commit hash.

misc/test_case_to_actual.py

@@ -10,7 +10,7 @@ class Chunk:
     def __init__(self, header_type: str, args: str) -> None:
         self.header_type = header_type
         self.args = args
-        self.lines: List[str] = []
+        self.lines: list[str] = []
 
 
 def is_header(line: str) -> bool:
@@ -36,7 +36,7 @@ def produce_chunks(lines: Iterator[str]) -> Iterator[Chunk]:
         yield current_chunk
 
 
-def write_out(filename: str, lines: List[str]) -> None:
+def write_out(filename: str, lines: list[str]) -> None:
     os.makedirs(os.path.dirname(filename), exist_ok=True)
     with open(filename, "w") as stream:
         stream.write("\n".join(lines))

misc/touch_checker.py

@@ -24,7 +24,7 @@ def delete_folder(folder_path: str) -> None:
         shutil.rmtree(folder_path)
 
 
-def execute(command: List[str]) -> None:
+def execute(command: list[str]) -> None:
     proc = subprocess.Popen(
         " ".join(command), stderr=subprocess.PIPE, stdout=subprocess.PIPE, shell=True
     )
@@ -53,7 +53,7 @@ def test(setup: Command, command: Command, teardown: Command) -> float:
     return end
 
 
-def make_touch_wrappers(filename: str) -> Tuple[Command, Command]:
+def make_touch_wrappers(filename: str) -> tuple[Command, Command]:
     def setup() -> None:
         execute(["touch", filename])
 
@@ -63,8 +63,8 @@ def teardown() -> None:
     return setup, teardown
 
 
-def make_change_wrappers(filename: str) -> Tuple[Command, Command]:
-    copy: Optional[str] = None
+def make_change_wrappers(filename: str) -> tuple[Command, Command]:
+    copy: str | None = None
 
     def setup() -> None:
         nonlocal copy

misc/upload-pypi.py

@@ -29,14 +29,14 @@ def is_whl_or_tar(name: str) -> bool:
     return name.endswith(".tar.gz") or name.endswith(".whl")
 
 
-def get_release_for_tag(tag: str) -> Dict[str, Any]:
+def get_release_for_tag(tag: str) -> dict[str, Any]:
     with urlopen(f"{BASE}/{REPO}/releases/tags/{tag}") as f:
         data = json.load(f)
     assert data["tag_name"] == tag
     return data
 
 
-def download_asset(asset: Dict[str, Any], dst: Path) -> Path:
+def download_asset(asset: dict[str, Any], dst: Path) -> Path:
     name = asset["name"]
     download_url = asset["browser_download_url"]
     assert is_whl_or_tar(name)
@@ -46,7 +46,7 @@ def download_asset(asset: Dict[str, Any], dst: Path) -> Path:
     return dst / name
 
 
-def download_all_release_assets(release: Dict[str, Any], dst: Path) -> None:
+def download_all_release_assets(release: dict[str, Any], dst: Path) -> None:
     print(f"Downloading assets...")
     with ThreadPoolExecutor() as e:
         for asset in e.map(lambda asset: download_asset(asset, dst), release["assets"]):
@@ -92,7 +92,7 @@ def tmp_twine() -> Iterator[Path]:
 def upload_dist(dist: Path, dry_run: bool = True) -> None:
     with tmp_twine() as twine:
         files = [item for item in dist.iterdir() if is_whl_or_tar(item.name)]
-        cmd: List[Any] = [twine, "upload"]
+        cmd: list[Any] = [twine, "upload"]
         cmd += files
         if dry_run:
             print("[dry run] " + " ".join(map(str, cmd)))

mypy/api.py

@@ -50,7 +50,7 @@
 from typing import Callable, List, TextIO, Tuple
 
 
-def _run(main_wrapper: Callable[[TextIO, TextIO], None]) -> Tuple[str, str, int]:
+def _run(main_wrapper: Callable[[TextIO, TextIO], None]) -> tuple[str, str, int]:
 
     stdout = StringIO()
     stderr = StringIO()
@@ -64,7 +64,7 @@ def _run(main_wrapper: Callable[[TextIO, TextIO], None]) -> Tuple[str, str, int]
     return stdout.getvalue(), stderr.getvalue(), exit_status
 
 
-def run(args: List[str]) -> Tuple[str, str, int]:
+def run(args: list[str]) -> tuple[str, str, int]:
     # Lazy import to avoid needing to import all of mypy to call run_dmypy
     from mypy.main import main
 
@@ -73,7 +73,7 @@ def run(args: List[str]) -> Tuple[str, str, int]:
     )
 
 
-def run_dmypy(args: List[str]) -> Tuple[str, str, int]:
+def run_dmypy(args: list[str]) -> tuple[str, str, int]:
     from mypy.dmypy.client import main
 
     # A bunch of effort has been put into threading stdout and stderr

mypy/applytype.py

@@ -27,7 +27,7 @@ def get_target_type(
     report_incompatible_typevar_value: Callable[[CallableType, Type, str, Context], None],
     context: Context,
     skip_unsatisfied: bool,
-) -> Optional[Type]:
+) -> Type | None:
     if isinstance(tvar, ParamSpecType):
         return type
     if isinstance(tvar, TypeVarTupleType):
@@ -68,7 +68,7 @@ def get_target_type(
 
 def apply_generic_arguments(
     callable: CallableType,
-    orig_types: Sequence[Optional[Type]],
+    orig_types: Sequence[Type | None],
     report_incompatible_typevar_value: Callable[[CallableType, Type, str, Context], None],
     context: Context,
     skip_unsatisfied: bool = False,
@@ -88,7 +88,7 @@ def apply_generic_arguments(
     # Check that inferred type variable values are compatible with allowed
     # values and bounds.  Also, promote subtype values to allowed values.
     # Create a map from type variable id to target type.
-    id_to_type: Dict[TypeVarId, Type] = {}
+    id_to_type: dict[TypeVarId, Type] = {}
 
     for tvar, type in zip(tvars, orig_types):
         assert not isinstance(type, PartialType), "Internal error: must never apply partial type"