typing: annotate repoobj module boundaries and enable strict mypy checks (refs #10021)

pyproject.toml

@@ -175,13 +175,27 @@ module = [
     "msgpack.*",
     "llfuse",
     "pyfuse3",
+    "mfusepy",
     "trio",
+    "borghash.*",
+    "borgstore.*",
+    "textual.*",
+    "backports.*",
     "borg.crypto.low_level",
     "borg.legacy.crypto.low_level",
     "borg.platform.*",
 ]
 ignore_missing_imports = true
 
+[[tool.mypy.overrides]]
+module = [
+    "borg.repoobj",
+    "borg.legacy.repoobj",
+]
+disallow_untyped_defs = true
+disallow_incomplete_defs = true
+check_untyped_defs = true
+
 [tool.tox]
 requires = ["tox>=4.19", "pkgconfig", "cython", "wheel", "setuptools-scm[toml]"]
 # Important: when adding/removing Python versions here,

src/borg/compress.pyi

@@ -10,20 +10,22 @@ def forget_zstd_compressors(keep_alive: bool = ...) -> None: ...
 
 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 | memoryview) -> tuple[dict, bytes]: ...
+    def decompress(self, meta: dict, data: bytes | memoryview) -> tuple[dict, bytes]: ...
     @staticmethod
     def detect(data: bytes) -> Any: ...
 
 class CompressorBase:
     ID: int = ...
     name: str = ...
+    level: int = ...
+    legacy_mode: bool = ...
     @classmethod
     def detect(self, data: bytes) -> bool: ...
     def __init__(self, level: int = ..., **kwargs) -> None: ...
     def decide(self, data: bytes) -> Any: ...
-    def compress(self, data: bytes) -> bytes: ...
-    def decompress(self, data: bytes) -> bytes: ...
+    def compress(self, meta: dict, data: bytes | memoryview) -> tuple[dict, bytes]: ...
+    def decompress(self, meta: dict, data: bytes | memoryview) -> tuple[dict, bytes]: ...
 
 class Auto(CompressorBase):
     def __init__(self, compressor: Any) -> None: ...
@@ -57,7 +59,7 @@ class ZSTD(DecidingCompressor):
     def __init__(self, level: int = ..., **kwargs) -> None: ...
     level: int
 
-LZ4_COMPRESSOR: type[LZ4]
-NONE_COMPRESSOR: type[CNONE]
+LZ4_COMPRESSOR: LZ4
+NONE_COMPRESSOR: CNONE
 
 COMPRESSOR_TABLE: dict

src/borg/legacy/repoobj.py

@@ -3,6 +3,8 @@
 Moved from borg.repoobj as part of the legacy code separation.
 """
 
+from typing import Any
+
 from ..constants import *  # NOQA
 from ..helpers import workarounds
 from ..compress import Compressor, get_compressor
@@ -12,6 +14,9 @@
 
 
 class RepoObj1:  # legacy
+    key: Any
+    compressor: Any
+
     def set_assert_id_place(self, place: str) -> None:
         # borg 1.x crypto has no AEAD with the chunk id in the AAD, so the id check is a needed part of
         # the read path here: parse always does it and there is nothing to configure (see BORG_ASSERT_ID).
@@ -22,7 +27,7 @@ def extract_crypted_data(cls, data: bytes) -> bytes:
         # used for crypto type detection
         return data
 
-    def __init__(self, key):
+    def __init__(self, key: Any) -> None:
         self.key = key
         self.compressor = get_compressor("lz4", legacy_mode=True)
 
@@ -32,13 +37,13 @@ def id_hash(self, data: bytes) -> bytes:
     def format(
         self,
         id: bytes,
-        meta: dict,
-        data: bytes,
+        meta: dict[str, Any],
+        data: bytes | memoryview,
         compress: bool = True,
-        size: int = None,
-        ctype: int = None,
-        clevel: int = None,
-        ro_type: str = None,
+        size: int | None = None,
+        ctype: int | None = None,
+        clevel: int | None = None,
+        ro_type: str | None = None,
     ) -> bytes:
         assert isinstance(id, bytes)
         assert meta == {}
@@ -54,20 +59,25 @@ def format(
         data_encrypted = self.key.encrypt(id, data_compressed)
         return data_encrypted
 
-    def parse_meta(self, id: bytes, cdata: bytes) -> dict:
+    def parse_meta(self, id: bytes, cdata: bytes | memoryview) -> dict[str, Any]:
         raise NotImplementedError("parse_meta is not available for RepoObj1")
 
     def parse(
-        self, id: bytes, cdata: bytes, decompress: bool = True, want_compressed: bool = False, ro_type: str = None
-    ) -> tuple[dict, bytes]:
+        self,
+        id: bytes,
+        cdata: bytes | memoryview,
+        decompress: bool = True,
+        want_compressed: bool = False,
+        ro_type: str | None = None,
+    ) -> tuple[dict[str, Any] | None, bytes | None]:
         assert not (not decompress and not want_compressed), "invalid parameter combination!"
         assert isinstance(id, bytes)
-        assert isinstance(cdata, bytes)
+        assert isinstance(cdata, (bytes, memoryview))
         assert ro_type is not None
         data_compressed = self.key.decrypt(id, cdata)
         compressor_cls, compression_level = Compressor.detect(data_compressed[:2])
         compressor = compressor_cls(level=compression_level, legacy_mode=True)
-        meta_compressed = {}
+        meta_compressed: dict[str, Any] = {}
         meta_compressed["ctype"] = compressor.ID
         meta_compressed["clevel"] = compressor.level
         meta_compressed["csize"] = len(data_compressed)

src/borg/repoobj.py

@@ -1,11 +1,12 @@
 import os
-from collections import namedtuple
+from collections.abc import Callable
 from struct import Struct
+from typing import Any, NamedTuple
 
 from .constants import *  # NOQA
 from .helpers import msgpack, workarounds
 from .helpers.errors import Error, IntegrityError
-from .compress import Compressor, LZ4_COMPRESSOR
+from .compress import Compressor, CompressorBase, LZ4_COMPRESSOR
 from .crypto.low_level import IntegrityError as IntegrityErrorBase
 
 # Workaround for lost passphrase or key in the "authenticated-*" modes
@@ -31,7 +32,7 @@
 ASSERT_ID_PLACE_DEFAULT = "read"
 
 
-def get_assert_id_places():
+def get_assert_id_places() -> frozenset[str]:
     """Determine the configurable places that shall verify the chunk id, see the BORG_ASSERT_ID docs."""
     value = os.environ.get("BORG_ASSERT_ID")
     if value is None:
@@ -75,9 +76,20 @@ def get_assert_id_places():
 
 
 class RepoObj:
+    class ObjHeader(NamedTuple):
+        magic: bytes
+        version: int
+        chunk_id: bytes
+        meta_size: int
+        data_size: int
+
     # Object header: magic (8b), format version (1b), chunk_id (32b), meta size (4b), data size (4b).
-    obj_header = Struct("<8sB32sII")
-    ObjHeader = namedtuple("ObjHeader", "magic version chunk_id meta_size data_size")
+    obj_header: Struct = Struct("<8sB32sII")
+
+    key: Any
+    compressor: CompressorBase
+    assert_id_places: frozenset[str]
+    assert_id_place: str
 
     @classmethod
     def parse_header(cls, buf: bytes | memoryview) -> tuple["RepoObj.ObjHeader | None", str | None]:
@@ -97,7 +109,7 @@ def parse_header(cls, buf: bytes | memoryview) -> tuple["RepoObj.ObjHeader | Non
             return None, f"unsupported object version {hdr.version}"
         return hdr, None
 
-    def __init__(self, key):
+    def __init__(self, key: Any) -> None:
         self.key = key
         # Some commands write new chunks (e.g. rename) but don't take a --compression argument. This duplicates
         # the default used by those commands who do take a --compression argument.
@@ -120,13 +132,13 @@ def id_hash(self, data: bytes) -> bytes:
     def format(
         self,
         id: bytes,
-        meta: dict,
-        data: bytes,
+        meta: dict[str, Any],
+        data: bytes | memoryview,
         compress: bool = True,
-        size: int = None,
-        ctype: int = None,
-        clevel: int = None,
-        ro_type: str = None,
+        size: int | None = None,
+        ctype: int | None = None,
+        clevel: int | None = None,
+        ro_type: str | None = None,
     ) -> bytes:
         assert isinstance(ro_type, str)
         assert ro_type != ROBJ_DONTCARE
@@ -136,6 +148,7 @@ def format(
         assert isinstance(meta, dict)
         assert isinstance(data, (bytes, memoryview))
         assert compress or size is not None and ctype is not None and clevel is not None
+        data_compressed: bytes | memoryview
         if compress:
             assert size is None or size == len(data)
             meta, data_compressed = self.compressor.compress(meta, data)
@@ -156,7 +169,7 @@ def format(
         hdr_packed = self.obj_header.pack(*hdr)
         return hdr_packed + meta_encrypted + data_encrypted
 
-    def parse_meta(self, id: bytes, cdata: bytes | memoryview, ro_type: str) -> dict:
+    def parse_meta(self, id: bytes, cdata: bytes | memoryview, ro_type: str) -> dict[str, Any]:
         # when calling parse_meta, enough cdata needs to be supplied to contain completely the
         # meta_len_hdr and the encrypted, packed metadata. it is allowed to provide more cdata.
         assert isinstance(id, bytes)
@@ -187,9 +200,9 @@ def parse(
         cdata: bytes | memoryview,
         decompress: bool = True,
         want_compressed: bool = False,
-        ro_type: str = None,
-        assert_id_place: str = None,
-    ) -> tuple[dict, bytes]:
+        ro_type: str | None = None,
+        assert_id_place: str | None = None,
+    ) -> tuple[dict[str, Any] | None, bytes | None]:
         """
         Parse a repo object into metadata and data (decrypt it, maybe decompress, maybe verify if the chunk plaintext
         corresponds to the chunk id via assert_id()).
@@ -256,7 +269,7 @@ def parse(
         return meta_compressed if want_compressed else meta, data_compressed if want_compressed else data
 
 
-def object_validator(repo_objs):
+def object_validator(repo_objs: RepoObj) -> Callable[[bytes, bytes | memoryview], bool]:
     """Return validate(chunk_id, obj): True if obj is the repo object with id chunk_id.
 
     obj is an object's header plus its metadata slot. Parsing that slot verifies its tag, which is
@@ -272,7 +285,7 @@ def object_validator(repo_objs):
     hdr_size = RepoObj.obj_header.size
     overhead = repo_objs.key.PAYLOAD_OVERHEAD  # the envelope adds a fixed number of bytes to the payload
 
-    def validate(chunk_id, obj):
+    def validate(chunk_id: bytes, obj: bytes | memoryview) -> bool:
         try:
             meta = repo_objs.parse_meta(chunk_id, obj, ro_type=ROBJ_DONTCARE)
         except (IntegrityErrorBase, msgpack.UnpackException):
@@ -296,7 +309,7 @@ class ObjectsNotAuthenticatable(Error):
     """Objects can not be authenticated with BORG_WORKAROUNDS=authenticated_no_key."""
 
 
-def whole_object_authenticator(repo_objs):
+def whole_object_authenticator(repo_objs: RepoObj) -> Callable[[bytes, bytes | memoryview], bool]:
     """Return authenticate(chunk_id, obj): True if obj is the whole repo object with id chunk_id.
 
     obj is an object's header, metadata slot and data slot. Parsing it checks that the header's sizes
@@ -313,7 +326,7 @@ def whole_object_authenticator(repo_objs):
     if AUTHENTICATED_NO_KEY and isinstance(repo_objs.key, MACKeyBase):
         raise ObjectsNotAuthenticatable
 
-    def authenticate(chunk_id, obj):
+    def authenticate(chunk_id: bytes, obj: bytes | memoryview) -> bool:
         try:
             repo_objs.parse(chunk_id, obj, decompress=False, want_compressed=True, ro_type=ROBJ_DONTCARE)
         except (IntegrityErrorBase, msgpack.UnpackException):