modernize: use Python 3.8+ features

src/borg/archiver/serve_cmd.py

@@ -42,7 +42,7 @@ def check_rest_restrictions(backend, restrict_to_paths, restrict_to_repositories
             return
         if not backend.startswith("FILE:"):
             raise PathNotAllowed("only FILE: backends can be restricted")
-        path = os.path.realpath(os.path.expanduser(backend[len("FILE:") :]))
+        path = os.path.realpath(os.path.expanduser(backend.removeprefix("FILE:")))
         path_with_sep = os.path.join(path, "")  # ensure trailing slash for prefix checks
         if restrict_to_paths:
             for p in restrict_to_paths:

src/borg/archiver/tar_cmds.py

@@ -293,10 +293,7 @@ def _import_tar(self, args, repository, manifest, key, cache, tarstream):
 
         tar = tarfile.open(fileobj=tarstream, mode="r|", ignore_zeros=args.ignore_zeros)
 
-        while True:
-            tarinfo = tar.next()
-            if not tarinfo:
-                break
+        while tarinfo := tar.next():
             if tarinfo.isreg():
                 status = tfo.process_file(tarinfo=tarinfo, status="A", type=stat.S_IFREG, tar=tar)
                 archive.stats.nfiles += 1

src/borg/chunkers/buzhash.pyi

@@ -1,4 +1,5 @@
-from typing import List, Iterator, BinaryIO
+from collections.abc import Iterator
+from typing import BinaryIO
 
 from .reader import fmap_entry
 
@@ -15,4 +16,4 @@ class Chunker:
         hash_window_size: int,
         sparse: bool = False,
     ) -> None: ...
-    def chunkify(self, fd: BinaryIO = None, fh: int = -1, fmap: List[fmap_entry] = None) -> Iterator: ...
+    def chunkify(self, fd: BinaryIO = None, fh: int = -1, fmap: list[fmap_entry] = None) -> Iterator: ...

src/borg/chunkers/buzhash64.pyi

@@ -1,10 +1,11 @@
-from typing import List, Iterator, BinaryIO
+from collections.abc import Iterator
+from typing import BinaryIO
 
 from .reader import fmap_entry
 
 def buzhash64(data: bytes, key: bytes) -> int: ...
 def buzhash64_update(sum: int, remove: int, add: int, len: int, key: bytes) -> int: ...
-def buzhash64_get_table(key: bytes) -> List[int]: ...
+def buzhash64_get_table(key: bytes) -> list[int]: ...
 
 class ChunkerBuzHash64:
     def __init__(
@@ -16,4 +17,4 @@ class ChunkerBuzHash64:
         hash_window_size: int,
         sparse: bool = False,
     ) -> None: ...
-    def chunkify(self, fd: BinaryIO = None, fh: int = -1, fmap: List[fmap_entry] = None) -> Iterator: ...
+    def chunkify(self, fd: BinaryIO = None, fh: int = -1, fmap: list[fmap_entry] = None) -> Iterator: ...

src/borg/chunkers/failing.py

@@ -1,6 +1,7 @@
 import os
 import errno
-from typing import BinaryIO, Iterator
+from collections.abc import Iterator
+from typing import BinaryIO
 
 from ..constants import CH_DATA
 from .reader import Chunk

src/borg/chunkers/fixed.py

@@ -1,4 +1,5 @@
-from typing import Iterator, BinaryIO
+from collections.abc import Iterator
+from typing import BinaryIO
 
 
 import time

src/borg/chunkers/reader.pyi

@@ -1,16 +1,17 @@
-from typing import NamedTuple, Tuple, Dict, List, Any, Type, BinaryIO, Iterator
+from collections.abc import Iterator
+from typing import Any, BinaryIO, NamedTuple
 
 has_seek_hole: bool
 
 class _Chunk(NamedTuple):
     data: bytes | None
-    meta: Dict[str, Any]
+    meta: dict[str, Any]
 
-def Chunk(data: bytes | None, **meta) -> Type[_Chunk]: ...
+def Chunk(data: bytes | None, **meta) -> type[_Chunk]: ...
 
-fmap_entry = Tuple[int, int, bool]
+fmap_entry = tuple[int, int, bool]
 
-def sparsemap(fd: BinaryIO = None, fh: int = -1) -> List[fmap_entry]: ...
+def sparsemap(fd: BinaryIO = None, fh: int = -1) -> list[fmap_entry]: ...
 
 class FileFMAPReader:
     def __init__(
@@ -20,9 +21,9 @@ class FileFMAPReader:
         fh: int = -1,
         read_size: int = 0,
         sparse: bool = False,
-        fmap: List[fmap_entry] = None,
+        fmap: list[fmap_entry] = None,
     ) -> None: ...
-    def _build_fmap(self) -> List[fmap_entry]: ...
+    def _build_fmap(self) -> list[fmap_entry]: ...
     def blockify(self) -> Iterator: ...
 
 class FileReader:
@@ -33,7 +34,7 @@ class FileReader:
         fh: int = -1,
         read_size: int = 0,
         sparse: bool = False,
-        fmap: List[fmap_entry] = None,
+        fmap: list[fmap_entry] = None,
     ) -> None: ...
     def _fill_buffer(self) -> bool: ...
-    def read(self, size: int) -> Type[_Chunk]: ...
+    def read(self, size: int) -> type[_Chunk]: ...

src/borg/cockpit/runner.py

@@ -6,7 +6,7 @@
 import logging
 import os
 import sys
-from typing import Callable
+from collections.abc import Callable
 
 
 class BorgRunner:
@@ -44,10 +44,7 @@ async def start(self):
             )
 
             async def read_stream(stream, stream_name):
-                while True:
-                    line = await stream.readline()
-                    if not line:
-                        break
+                while line := await stream.readline():
                     decoded_line = line.decode("utf-8", errors="replace").rstrip()
                     if decoded_line:
                         self.log_callback({"type": "stream_line", "stream": stream_name, "line": decoded_line})

src/borg/compress.pyi

@@ -1,11 +1,11 @@
-from typing import Any, Type, Dict, Tuple
+from typing import Any
 
 def get_compressor(name: str, **kwargs) -> Any: ...
 
 class Compressor:
     def __init__(self, name: Any = ..., **kwargs) -> None: ...
-    def compress(self, meta: Dict, data: bytes) -> Tuple[Dict, bytes]: ...
-    def decompress(self, meta: Dict, data: bytes) -> Tuple[Dict, bytes]: ...
+    def compress(self, meta: dict, data: bytes) -> tuple[dict, bytes]: ...
+    def decompress(self, meta: dict, data: bytes) -> tuple[dict, bytes]: ...
     @staticmethod
     def detect(data: bytes) -> Any: ...
 
@@ -51,7 +51,7 @@ class ZSTD(DecidingCompressor):
     def __init__(self, level: int = ..., **kwargs) -> None: ...
     level: int
 
-LZ4_COMPRESSOR: Type[LZ4]
-NONE_COMPRESSOR: Type[CNONE]
+LZ4_COMPRESSOR: type[LZ4]
+NONE_COMPRESSOR: type[CNONE]
 
 COMPRESSOR_TABLE: dict

src/borg/crypto/low_level.pyi

@@ -1,8 +1,6 @@
 # Type stubs for borg.crypto.low_level
 # This file provides type hints for the Cython extension module
 
-from typing import Optional, Union
-
 # Module-level functions
 def num_cipher_blocks(length: int, blocksize: int = 16) -> int:
     """Return the number of cipher blocks required to encrypt/decrypt <length> bytes of data."""
@@ -31,23 +29,18 @@ class UNENCRYPTED:
     """Unencrypted cipher suite (no encryption, no MAC)."""
 
     header_len: int
-    iv: Optional[Union[int, bytes]]
+    iv: int | bytes | None
 
     def __init__(
-        self,
-        mac_key: None,
-        enc_key: None,
-        iv: Optional[Union[int, bytes]] = None,
-        header_len: int = 1,
-        aad_offset: int = 1,
+        self, mac_key: None, enc_key: None, iv: int | bytes | None = None, header_len: int = 1, aad_offset: int = 1
     ) -> None: ...
     def encrypt(
-        self, data: bytes, header: bytes = b"", iv: Optional[Union[int, bytes]] = None, aad: Optional[bytes] = None
+        self, data: bytes, header: bytes = b"", iv: int | bytes | None = None, aad: bytes | None = None
     ) -> bytes: ...
-    def decrypt(self, envelope: bytes, aad: Optional[bytes] = None) -> memoryview: ...
+    def decrypt(self, envelope: bytes, aad: bytes | None = None) -> memoryview: ...
     def block_count(self, length: int) -> int: ...
-    def set_iv(self, iv: Union[int, bytes]) -> None: ...
-    def next_iv(self) -> Union[int, bytes]: ...
+    def set_iv(self, iv: int | bytes) -> None: ...
+    def next_iv(self) -> int | bytes: ...
     def extract_iv(self, envelope: bytes) -> int: ...
 
 class AES256_CTR_BASE:
@@ -56,44 +49,29 @@ class AES256_CTR_BASE:
     @classmethod
     def requirements_check(cls) -> None: ...
     def __init__(
-        self,
-        mac_key: bytes,
-        enc_key: bytes,
-        iv: Optional[Union[int, bytes]] = None,
-        header_len: int = 1,
-        aad_offset: int = 1,
+        self, mac_key: bytes, enc_key: bytes, iv: int | bytes | None = None, header_len: int = 1, aad_offset: int = 1
     ) -> None: ...
     def encrypt(
-        self, data: bytes, header: bytes = b"", iv: Optional[Union[int, bytes]] = None, aad: Optional[bytes] = None
+        self, data: bytes, header: bytes = b"", iv: int | bytes | None = None, aad: bytes | None = None
     ) -> bytes: ...
-    def decrypt(self, envelope: bytes, aad: Optional[bytes] = None) -> bytes: ...
+    def decrypt(self, envelope: bytes, aad: bytes | None = None) -> bytes: ...
     def block_count(self, length: int) -> int: ...
-    def set_iv(self, iv: Union[int, bytes]) -> None: ...
+    def set_iv(self, iv: int | bytes) -> None: ...
     def next_iv(self) -> int: ...
     def extract_iv(self, envelope: bytes) -> int: ...
 
 class AES256_CTR_HMAC_SHA256(AES256_CTR_BASE):
     """AES-256-CTR with HMAC-SHA256 authentication."""
 
     def __init__(
-        self,
-        mac_key: bytes,
-        enc_key: bytes,
-        iv: Optional[Union[int, bytes]] = None,
-        header_len: int = 1,
-        aad_offset: int = 1,
+        self, mac_key: bytes, enc_key: bytes, iv: int | bytes | None = None, header_len: int = 1, aad_offset: int = 1
     ) -> None: ...
 
 class AES256_CTR_BLAKE2b(AES256_CTR_BASE):
     """AES-256-CTR with BLAKE2b authentication."""
 
     def __init__(
-        self,
-        mac_key: bytes,
-        enc_key: bytes,
-        iv: Optional[Union[int, bytes]] = None,
-        header_len: int = 1,
-        aad_offset: int = 1,
+        self, mac_key: bytes, enc_key: bytes, iv: int | bytes | None = None, header_len: int = 1, aad_offset: int = 1
     ) -> None: ...
 
 class _AEAD_BASE:
@@ -104,43 +82,35 @@ class _AEAD_BASE:
         """Check whether library requirements for this ciphersuite are satisfied."""
         ...
 
-    def __init__(
-        self, key: bytes, iv: Optional[Union[int, bytes]] = None, header_len: int = 0, aad_offset: int = 0
-    ) -> None: ...
-    def encrypt(
-        self, data: bytes, header: bytes = b"", iv: Optional[Union[int, bytes]] = None, aad: bytes = b""
-    ) -> bytes: ...
+    def __init__(self, key: bytes, iv: int | bytes | None = None, header_len: int = 0, aad_offset: int = 0) -> None: ...
+    def encrypt(self, data: bytes, header: bytes = b"", iv: int | bytes | None = None, aad: bytes = b"") -> bytes: ...
     def decrypt(self, envelope: bytes, aad: bytes = b"") -> bytes: ...
     def block_count(self, length: int) -> int: ...
-    def set_iv(self, iv: Union[int, bytes]) -> None: ...
+    def set_iv(self, iv: int | bytes) -> None: ...
     def next_iv(self) -> int: ...
 
 class AES256_OCB(_AEAD_BASE):
     """AES-256-OCB AEAD cipher suite."""
 
     @classmethod
     def requirements_check(cls) -> None: ...
-    def __init__(
-        self, key: bytes, iv: Optional[Union[int, bytes]] = None, header_len: int = 0, aad_offset: int = 0
-    ) -> None: ...
+    def __init__(self, key: bytes, iv: int | bytes | None = None, header_len: int = 0, aad_offset: int = 0) -> None: ...
 
 class CHACHA20_POLY1305(_AEAD_BASE):
     """ChaCha20-Poly1305 AEAD cipher suite."""
 
     @classmethod
     def requirements_check(cls) -> None: ...
-    def __init__(
-        self, key: bytes, iv: Optional[Union[int, bytes]] = None, header_len: int = 0, aad_offset: int = 0
-    ) -> None: ...
+    def __init__(self, key: bytes, iv: int | bytes | None = None, header_len: int = 0, aad_offset: int = 0) -> None: ...
 
 class AES:
     """A thin wrapper around the OpenSSL EVP cipher API - for legacy code, like key file encryption."""
 
-    def __init__(self, enc_key: bytes, iv: Optional[Union[int, bytes]] = None) -> None: ...
-    def encrypt(self, data: bytes, iv: Optional[Union[int, bytes]] = None) -> bytes: ...
+    def __init__(self, enc_key: bytes, iv: int | bytes | None = None) -> None: ...
+    def encrypt(self, data: bytes, iv: int | bytes | None = None) -> bytes: ...
     def decrypt(self, data: bytes) -> bytes: ...
     def block_count(self, length: int) -> int: ...
-    def set_iv(self, iv: Union[int, bytes]) -> None: ...
+    def set_iv(self, iv: int | bytes) -> None: ...
     def next_iv(self) -> int: ...
 
 class CSPRNG:

src/borg/hashindex.pyi

@@ -1,4 +1,5 @@
-from typing import NamedTuple, Tuple, Type, IO, Iterator, Any, MutableMapping
+from collections.abc import Iterator, MutableMapping
+from typing import IO, Any, NamedTuple
 
 PATH_OR_FILE = str | IO
 
@@ -11,7 +12,7 @@ class ChunkIndexEntry(NamedTuple):
     obj_offset: int
     obj_size: int
 
-CIE = Tuple[int, int, bytes, int, int] | Type[ChunkIndexEntry]
+CIE = tuple[int, int, bytes, int, int] | type[ChunkIndexEntry]
 
 class ChunkIndex:
     F_NONE: int
@@ -25,7 +26,7 @@ class ChunkIndex:
     def iteritems(self, *, only_new: bool = ...) -> Iterator: ...
     def clear_new(self) -> None: ...
     def __contains__(self, key: bytes) -> bool: ...
-    def __getitem__(self, key: bytes) -> Type[ChunkIndexEntry]: ...
+    def __getitem__(self, key: bytes) -> type[ChunkIndexEntry]: ...
     def __setitem__(self, key: bytes, value: CIE) -> None: ...
 
 class FuseVersionsIndexEntry(NamedTuple):

src/borg/helpers/nanorst.py

@@ -65,10 +65,7 @@ def rst_to_text(text, state_hook=None, references=None):
 
     inline_single = ("*", "`")
 
-    while True:
-        char = text.read(1)
-        if not char:
-            break
+    while char := text.read(1):
         next = text.peek(1)  # type: str
 
         if state == "text":