Upgade code with pyupgrade

Signed-off-by: Jean-Christophe Morin <[email protected]>

docs/source/conf.py

@@ -62,7 +62,7 @@
 # https://www.sphinx-doc.org/en/master/usage/configuration.html#options-for-the-linkcheck-builder
 
 linkcheck_allowed_redirects = {
-    r"https://github.com/JeanChristopheMorinPerso/rez-pip/issues/\d+": "https://github.com/JeanChristopheMorinPerso/rez-pip/pull/\d+"
+    r"https://github.com/JeanChristopheMorinPerso/rez-pip/issues/\d+": r"https://github.com/JeanChristopheMorinPerso/rez-pip/pull/\d+"
 }
 
 
@@ -214,7 +214,7 @@ def run(self) -> list[docutils.nodes.Node]:
 
                 # Add links to rez docs for known settings.
                 help_str = re.sub(
-                    "(.* \(default: configured )([a-zA-Z_]+)(.*)$",
+                    r"(.* \(default: configured )([a-zA-Z_]+)(.*)$",
                     r"\g<1> :external:data:`\g<2>`\g<3>",
                     help_str,
                 )
@@ -359,15 +359,6 @@ def autodoc_process_signature(
     signature: str,
     return_annotation,
 ):
-    for name in ["Sequence", "Mapping", "MutableSequence"]:
-        signature = signature.replace(
-            f"rez_pip.compat.{name}", f"~collections.abc.{name}"
-        )
-        if return_annotation:
-            return_annotation = return_annotation.replace(
-                f"rez_pip.compat.{name}", f"~collections.abc.{name}"
-            )
-
     signature = signature.replace(
         "rez_pip.compat.importlib_metadata", "~importlib.metadata"
     )

noxfile.py

@@ -14,7 +14,6 @@ def lint(session: nox.Session):
 @nox.session()
 def mypy(session: nox.Session):
     session.install("mypy")
-    session.install(".", "-c", "tests/constraints.txt")
 
     session.run("mypy")
 

scripts/get_pyside6_files.py

@@ -40,8 +40,8 @@ def __init__(
         self._length = int(head.headers["Content-Length"])
         self._file = tempfile.NamedTemporaryFile()
         self.truncate(self._length)
-        self._left: typing.List[int] = []
-        self._right: typing.List[int] = []
+        self._left: list[int] = []
+        self._right: list[int] = []
         if "bytes" not in head.headers.get("Accept-Ranges", "none"):
             raise ValueError("range request is not supported")
         self._check_zip()
@@ -101,7 +101,7 @@ def tell(self) -> int:
         """Return the current position."""
         return self._file.tell()
 
-    def truncate(self, size: typing.Optional[int] = None) -> int:
+    def truncate(self, size: int | None = None) -> int:
         """Resize the stream to the given size in bytes.
 
         If size is unspecified resize to the current position.
@@ -115,15 +115,15 @@ def writable(self) -> bool:
         """Return False."""
         return False
 
-    def __enter__(self) -> "LazyZipOverHTTP":
+    def __enter__(self) -> LazyZipOverHTTP:
         self._file.__enter__()
         return self
 
     def __exit__(self, *exc: Any) -> None:
         self._file.__exit__(*exc)
 
     @contextlib.contextmanager
-    def _stay(self) -> typing.Generator[None, None, None]:
+    def _stay(self) -> typing.Generator[None]:
         """Return a context manager keeping the position.
 
         At the end of the block, seek back to original position.
@@ -153,7 +153,7 @@ def _stream_response(
         self,
         start: int,
         end: int,
-        base_headers: typing.Dict[str, str] = {"Accept-Encoding": "identity"},
+        base_headers: dict[str, str] = {"Accept-Encoding": "identity"},
     ) -> requests.Response:
         """Return HTTP response to a range request from start to end."""
         headers = base_headers.copy()
@@ -164,7 +164,7 @@ def _stream_response(
 
     def _merge(
         self, start: int, end: int, left: int, right: int
-    ) -> typing.Generator[typing.Tuple[int, int], None, None]:
+    ) -> typing.Generator[tuple[int, int]]:
         """Return a generator of intervals to be fetched.
 
         Args:
@@ -276,7 +276,7 @@ def run():
     while len(versions) > 1:
         leftFile = f"patches/data/{versions[0]}/__init__.py"
         rightFile = f"patches/data/{versions[1]}/__init__.py"
-        with open(leftFile, "r") as lfh, open(rightFile, "r") as rfh:
+        with open(leftFile) as lfh, open(rightFile) as rfh:
             lhs = ast.parse(lfh.read())
             rhs = ast.parse(rfh.read())
 

src/rez_pip/cli.py

@@ -4,7 +4,6 @@
 import sys
 import json
 import shutil
-import typing
 import logging
 import argparse
 import textwrap
@@ -35,7 +34,7 @@
 __all__ = ["run"]
 
 
-def __dir__() -> typing.List[str]:
+def __dir__() -> list[str]:
     return __all__
 
 
@@ -134,8 +133,8 @@ def _createParser() -> argparse.ArgumentParser:
 
 
 def _parseArgs(
-    args: typing.List[str],
-) -> typing.Tuple[argparse.Namespace, typing.List[str]]:
+    args: list[str],
+) -> tuple[argparse.Namespace, list[str]]:
     parser = _createParser()
 
     knownArgs = []
@@ -170,7 +169,7 @@ def _validateArgs(args: argparse.Namespace) -> None:
         )
 
 
-def _run(args: argparse.Namespace, pipArgs: typing.List[str], pipWorkArea: str) -> None:
+def _run(args: argparse.Namespace, pipArgs: list[str], pipWorkArea: str) -> None:
     pythonVersions = rez_pip.rez.getPythonExecutables(
         args.python_version, packageFamily="python"
     )
@@ -207,9 +206,7 @@ def _run(args: argparse.Namespace, pipArgs: typing.List[str], pipWorkArea: str)
             )
 
         _LOG.info(f"Resolved {len(packages)} dependencies for python {pythonVersion}")
-        _packageGroups: typing.List[
-            rez_pip.pip.PackageGroup[rez_pip.pip.PackageInfo]
-        ] = list(
+        _packageGroups: list[rez_pip.pip.PackageGroup[rez_pip.pip.PackageInfo]] = list(
             itertools.chain(*rez_pip.plugins.getHook().groupPackages(packages=packages))  # type: ignore[arg-type]
         )
 
@@ -231,7 +228,7 @@ def _run(args: argparse.Namespace, pipArgs: typing.List[str], pipWorkArea: str)
         # TODO: Should we postpone downloading to the last minute if we can?
         _LOG.info("[bold]Downloading...")
 
-        packageGroups: typing.List[
+        packageGroups: list[
             rez_pip.pip.PackageGroup[rez_pip.pip.DownloadedArtifact]
         ] = rez_pip.download.downloadPackages(_packageGroups, wheelsDir)
 

src/rez_pip/compat.py

@@ -1,13 +1,9 @@
 import sys
 
-if sys.version_info <= (3, 8):
-    from typing import Sequence, MutableSequence, Mapping
-else:
-    from collections.abc import Sequence, MutableSequence, Mapping
 
 if sys.version_info >= (3, 10):
     import importlib.metadata as importlib_metadata
 else:
     import importlib_metadata
 
-__all__ = ["Sequence", "MutableSequence", "Mapping", "importlib_metadata"]
+__all__ = ["importlib_metadata"]

src/rez_pip/download.py

@@ -18,19 +18,19 @@
 
 
 def downloadPackages(
-    packageGroups: typing.List[rez_pip.pip.PackageGroup[rez_pip.pip.PackageInfo]],
+    packageGroups: list[rez_pip.pip.PackageGroup[rez_pip.pip.PackageInfo]],
     dest: str,
-) -> typing.List[rez_pip.pip.PackageGroup[rez_pip.pip.DownloadedArtifact]]:
+) -> list[rez_pip.pip.PackageGroup[rez_pip.pip.DownloadedArtifact]]:
     return asyncio.run(_downloadPackages(packageGroups, dest))
 
 
 async def _downloadPackages(
-    packageGroups: typing.List[rez_pip.pip.PackageGroup[rez_pip.pip.PackageInfo]],
+    packageGroups: list[rez_pip.pip.PackageGroup[rez_pip.pip.PackageInfo]],
     dest: str,
 ) -> list[rez_pip.pip.PackageGroup[rez_pip.pip.DownloadedArtifact]]:
-    newPackageGroups: typing.List[
-        rez_pip.pip.PackageGroup[rez_pip.pip.DownloadedArtifact]
-    ] = []
+    newPackageGroups: list[rez_pip.pip.PackageGroup[rez_pip.pip.DownloadedArtifact]] = (
+        []
+    )
     someFailed = False
 
     async with aiohttp.ClientSession() as session:
@@ -43,7 +43,7 @@ async def _downloadPackages(
             transient=True,
             console=rez_pip.utils.CONSOLE,
         ) as progress:
-            tasks: typing.Dict[str, rich.progress.TaskID] = {}
+            tasks: dict[str, rich.progress.TaskID] = {}
 
             # Create all the download tasks first
             numPackages = 0
@@ -114,7 +114,7 @@ async def _return_local(
                     someFailed = True
 
                 artifacts = typing.cast(
-                    typing.Tuple[rez_pip.pip.DownloadedArtifact], artifacts
+                    tuple[rez_pip.pip.DownloadedArtifact], artifacts
                 )
 
                 newPackageGroups.append(

src/rez_pip/install.py

@@ -11,19 +11,15 @@
 import shutil
 import typing
 import logging
-import fnmatch
 import pathlib
 import zipfile
 import sysconfig
+import collections.abc
 
 import rez_pip.exceptions
 
 if typing.TYPE_CHECKING:
-    if sys.version_info >= (3, 8):
-        from typing import Literal
-    else:
-        from typing_extensions import Literal
-    import rez_pip.compat
+    from typing import Literal
 
 import installer
 import installer.utils
@@ -65,7 +61,7 @@ def isWheelPure(dist: importlib_metadata.Distribution) -> bool:
 
 
 # Taken from https://github.com/pypa/installer/blob/main/src/installer/__main__.py#L49
-def getSchemeDict(name: str, target: str) -> typing.Dict[str, str]:
+def getSchemeDict(name: str, target: str) -> dict[str, str]:
     vars = {}
     vars["base"] = vars["platbase"] = installed_base = target
 
@@ -143,7 +139,7 @@ def installWheel(
 class CustomWheelDestination(installer.destinations.SchemeDictionaryDestination):
     # Exactly the same as SchemeDictionaryDestination, but uses our custom Script class.
     def write_script(
-        self, name: str, module: str, attr: str, section: "ScriptSection"
+        self, name: str, module: str, attr: str, section: ScriptSection
     ) -> installer.records.RecordEntry:
         """Write a script to invoke an entrypoint.
         :param name: name of the script
@@ -191,9 +187,7 @@ def write_script(
 
 # TODO: Document where this code comes from.
 class Script(installer.scripts.Script):
-    def generate(
-        self, executable: str, kind: "LauncherKind"
-    ) -> typing.Tuple[str, bytes]:
+    def generate(self, executable: str, kind: LauncherKind) -> tuple[str, bytes]:
         """Generate a launcher for this script.
         :param executable: Path to the executable to invoke.
         :param kind: Which launcher template should be used.
@@ -232,12 +226,14 @@ def cleanup(dist: importlib_metadata.Distribution, path: str) -> None:
     are made on the installs (the wheel install). We could move this somewhere
     else but it's not clear where.
     """
-    actionsGroups: rez_pip.compat.Sequence[
-        rez_pip.compat.Sequence[rez_pip.plugins.CleanupAction]
-    ] = rez_pip.plugins.getHook().cleanup(dist=dist, path=path)
+    actionsGroups: collections.abc.Sequence[
+        collections.abc.Sequence[rez_pip.plugins.CleanupAction]
+    ] = rez_pip.plugins.getHook().cleanup(
+        dist=dist, path=path
+    )  # type: ignore[assignment]
 
     # Flatten
-    actions: typing.List[rez_pip.plugins.CleanupAction] = [
+    actions: list[rez_pip.plugins.CleanupAction] = [
         action for group in actionsGroups for action in group
     ]
 
@@ -272,7 +268,7 @@ def cleanup(dist: importlib_metadata.Distribution, path: str) -> None:
 
 
 def deleteEntryFromRecord(
-    dist: importlib_metadata.Distribution, path: str, entries: typing.List[str]
+    dist: importlib_metadata.Distribution, path: str, entries: list[str]
 ) -> None:
     """
     Delete an entry from the record file.
@@ -283,7 +279,7 @@ def deleteEntryFromRecord(
     items = [
         os.fspath(item)
         for item in dist.files
-        if re.search("[a-zA-Z0-9._+]+\.dist-info/RECORD", os.fspath(item))
+        if re.search(r"[a-zA-Z0-9._+]+\.dist-info/RECORD", os.fspath(item))
     ]
 
     if not items:
@@ -292,7 +288,7 @@ def deleteEntryFromRecord(
     recordFilePathRel = items[0]
     recordFilePath = os.path.join(path, "python", recordFilePathRel)
 
-    with open(recordFilePath, "r") as f:
+    with open(recordFilePath) as f:
         lines = f.readlines()
 
     schemesRaw = getSchemeDict(dist.name, path)

src/rez_pip/patch.py

@@ -5,6 +5,7 @@
 import typing
 import logging
 import contextlib
+import collections.abc
 import logging.handlers
 
 import patch_ng
@@ -30,7 +31,7 @@ def getBuiltinPatchesDir() -> str:
 
 
 @contextlib.contextmanager
-def logIfErrorOrRaises() -> typing.Generator[None]:
+def logIfErrorOrRaises() -> typing.Generator[None, None, None]:
     """
     Log patch_ng logs if any error is logged or if the wrapped body raises.
     Very slightly inspired by https://docs.python.org/3/howto/logging-cookbook.html#buffering-logging-messages-and-outputting-them-conditionally
@@ -66,7 +67,7 @@ def logIfErrorOrRaises() -> typing.Generator[None]:
 def patch(dist: importlib_metadata.Distribution, path: str) -> None:
     """Patch an installed package (wheel)"""
     _LOG.debug(f"[bold]Attempting to patch {dist.name!r} at {path!r}")
-    patchesGroups: rez_pip.compat.Sequence[rez_pip.compat.Sequence[str]] = (
+    patchesGroups: collections.abc.Sequence[collections.abc.Sequence[str]] = (
         rez_pip.plugins.getHook().patches(dist=dist, path=path)
     )