feat: add type hints & mypy validation

Author: Rogdham

.github/workflows/build.yml

@@ -26,7 +26,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        env: [generate-integration-files, lint, format]
+        env: [generate-integration-files, type, lint, format]
     steps:
       - uses: actions/checkout@v2
       - name: Setup Python

.pylintrc

@@ -1,3 +1,12 @@
+[BASIC]
+good-names =
+    i,
+    j,
+    k,
+    ex,
+    _,
+    T,
+
 [MESSAGES CONTROL]
 disable =
     missing-class-docstring,

CHANGELOG.md

@@ -9,6 +9,23 @@ adheres to [Semantic Versioning](https://semver.org/).
 
 [unreleased]: https://github.com/rogdham/python-xz/compare/v0.2.0...HEAD
 
+### :boom: Breaking changes
+
+- The `filename` argument of `XZFile` is now mandatory; this change should have very
+  limited impact as not providing it makes no sense and would have raised a `TypeError`,
+  plus it was already mandatory on `xz.open`
+
+### :rocket: Added
+
+- Type hints
+
+### :house: Internal
+
+- Type validation with mypy
+- Distribute `py.typed` file in conformance with [PEP 561]
+
+[pep 561]: https://www.python.org/dev/peps/pep-0561/
+
 ## [0.2.0] - 2021-10-23
 
 [0.2.0]: https://github.com/rogdham/python-xz/releases/tag/v0.2.0

README.md

@@ -6,7 +6,7 @@ Pure Python implementation of the XZ file format with random access support
 
 _Leveraging the lzma module for fast (de)compression_
 
-[![GitHub build status](https://img.shields.io/github/workflow/status/rogdham/python-xz/build/master)](https://github.com/rogdham/python-xz/actions?query=branch:master) [![Release on PyPI](https://img.shields.io/pypi/v/python-xz)](https://pypi.org/project/python-xz/) [![Code coverage](https://img.shields.io/badge/coverage-100%25-brightgreen)](https://github.com/rogdham/python-xz/search?q=fail+under&type=Code) [![MIT License](https://img.shields.io/pypi/l/python-xz)](https://github.com/Rogdham/python-xz/blob/master/LICENSE.txt)
+[![GitHub build status](https://img.shields.io/github/workflow/status/rogdham/python-xz/build/master)](https://github.com/rogdham/python-xz/actions?query=branch:master) [![Release on PyPI](https://img.shields.io/pypi/v/python-xz)](https://pypi.org/project/python-xz/) [![Code coverage](https://img.shields.io/badge/coverage-100%25-brightgreen)](https://github.com/rogdham/python-xz/search?q=fail+under&type=Code) [![Mypy type checker](https://img.shields.io/badge/type_checker-mypy-informational)](https://mypy.readthedocs.io/) [![MIT License](https://img.shields.io/pypi/l/python-xz)](https://github.com/Rogdham/python-xz/blob/master/LICENSE.txt)
 
 ---
 

dev-requirements.txt

@@ -1,6 +1,9 @@
 # install + dependencies
 -e .
 
+# typing
+mypy
+
 # tests
 coverage
 pytest

mypy.ini

@@ -0,0 +1,49 @@
+[mypy]
+# section names refer to the documentation
+# https://mypy.readthedocs.io/en/stable/config_file.html
+
+# Import discovery
+files = src
+ignore_missing_imports = False
+follow_imports = normal
+
+# Platform configuration
+python_version = 3.9
+
+# Disallow dynamic typing
+disallow_any_unimported = True
+disallow_any_decorated = True
+disallow_any_generics = True
+disallow_subclassing_any = True
+
+# Untyped definitions and calls
+disallow_untyped_calls = True
+disallow_untyped_defs = True
+disallow_incomplete_defs = True
+check_untyped_defs = True
+disallow_untyped_decorators = True
+
+# None and Optional handling
+no_implicit_optional = True
+strict_optional = True
+
+# Configuring warning
+warn_redundant_casts = True
+warn_unused_ignores = True
+warn_no_return = True
+warn_return_any = True
+warn_unreachable = True
+
+# Supressing errors
+show_none_errors = True
+ignore_errors = False
+
+# Miscellaneous strictness flags
+strict_equality = True
+
+# Configuring error messages
+show_error_context = True
+show_error_codes = True
+
+# Miscellaneous
+warn_unused_configs = True

setup.cfg

@@ -23,6 +23,7 @@ classifiers =
     Topic :: System :: Archiving :: Compression
 
 [options]
+include_package_data = True
 package_dir = =src
 packages = xz
 setup_requires =

src/xz/block.py

@@ -1,5 +1,6 @@
 from io import DEFAULT_BUFFER_SIZE, SEEK_SET
 from lzma import FORMAT_XZ, LZMACompressor, LZMADecompressor, LZMAError
+from typing import Tuple, Union
 
 from xz.common import (
     XZError,
@@ -9,12 +10,19 @@
     parse_xz_index,
 )
 from xz.io import IOAbstract, IOCombiner, IOStatic
+from xz.typing import _LZMAFiltersType, _LZMAPresetType
 
 
 class BlockRead:
     read_size = DEFAULT_BUFFER_SIZE
 
-    def __init__(self, fileobj, check, unpadded_size, uncompressed_size):
+    def __init__(
+        self,
+        fileobj: IOAbstract,
+        check: int,
+        unpadded_size: int,
+        uncompressed_size: int,
+    ) -> None:
         self.length = uncompressed_size
         self.fileobj = IOCombiner(
             IOStatic(create_xz_header(check)),
@@ -25,12 +33,12 @@ def __init__(self, fileobj, check, unpadded_size, uncompressed_size):
         )
         self.reset()
 
-    def reset(self):
+    def reset(self) -> None:
         self.fileobj.seek(0, SEEK_SET)
         self.pos = 0
         self.decompressor = LZMADecompressor(format=FORMAT_XZ)
 
-    def decompress(self, pos, size):
+    def decompress(self, pos: int, size: int) -> bytes:
         if pos < self.pos:
             self.reset()
 
@@ -63,24 +71,30 @@ def decompress(self, pos, size):
 
 
 class BlockWrite:
-    def __init__(self, fileobj, check, preset, filters):
+    def __init__(
+        self,
+        fileobj: IOAbstract,
+        check: int,
+        preset: _LZMAPresetType,
+        filters: _LZMAFiltersType,
+    ) -> None:
         self.fileobj = fileobj
         self.check = check
         self.compressor = LZMACompressor(FORMAT_XZ, check, preset, filters)
         self.pos = 0
         if self.compressor.compress(b"") != create_xz_header(check):
             raise XZError("block: compressor header")
 
-    def _write(self, data):
+    def _write(self, data: bytes) -> None:
         if data:
             self.fileobj.seek(self.pos)
             self.fileobj.write(data)
             self.pos += len(data)
 
-    def compress(self, data):
+    def compress(self, data: bytes) -> None:
         self._write(self.compressor.compress(data))
 
-    def finish(self):
+    def finish(self) -> Tuple[int, int]:
         data = self.compressor.flush()
 
         # footer
@@ -102,26 +116,26 @@ def finish(self):
 class XZBlock(IOAbstract):
     def __init__(
         self,
-        fileobj,
-        check,
-        unpadded_size,
-        uncompressed_size,
-        preset=None,
-        filters=None,
+        fileobj: IOAbstract,
+        check: int,
+        unpadded_size: int,
+        uncompressed_size: int,
+        preset: _LZMAPresetType = None,
+        filters: _LZMAFiltersType = None,
     ):
         super().__init__(uncompressed_size)
         self.fileobj = fileobj
         self.check = check
         self.preset = preset
         self.filters = filters
         self.unpadded_size = unpadded_size
-        self.operation = None
+        self.operation: Union[BlockRead, BlockWrite, None] = None
 
     @property
-    def uncompressed_size(self):
+    def uncompressed_size(self) -> int:
         return self._length
 
-    def _read(self, size):
+    def _read(self, size: int) -> bytes:
         # enforce read mode
         if not isinstance(self.operation, BlockRead):
             self._write_end()
@@ -138,10 +152,10 @@ def _read(self, size):
         except LZMAError as ex:
             raise XZError(f"block: error while decompressing: {ex}") from ex
 
-    def writable(self):
+    def writable(self) -> bool:
         return isinstance(self.operation, BlockWrite) or not self._length
 
-    def _write(self, data):
+    def _write(self, data: bytes) -> int:
         # enforce write mode
         if not isinstance(self.operation, BlockWrite):
             self.operation = BlockWrite(
@@ -155,14 +169,14 @@ def _write(self, data):
         self.operation.compress(data)
         return len(data)
 
-    def _write_after(self):
+    def _write_after(self) -> None:
         if isinstance(self.operation, BlockWrite):
             self.unpadded_size, uncompressed_size = self.operation.finish()
             if uncompressed_size != self.uncompressed_size:
                 raise XZError("block: compressor uncompressed size")
             self.operation = None
 
-    def _truncate(self, size):
+    def _truncate(self, size: int) -> None:
         # thanks to the writable method, we are sure that length is zero
         # so we don't need to handle the case of truncating in middle of the block
         self.seek(size)

src/xz/common.py

@@ -1,6 +1,7 @@
 from binascii import crc32 as crc32int
 import lzma
 from struct import pack, unpack
+from typing import List, Tuple, cast
 
 HEADER_MAGIC = b"\xfd7zXZ\x00"
 FOOTER_MAGIC = b"YZ"
@@ -10,7 +11,7 @@ class XZError(Exception):
     pass
 
 
-def encode_mbi(value):
+def encode_mbi(value: int) -> bytes:
     data = bytearray()
     while value >= 0x80:
         data.append((value & 0x7F) | 0x80)
@@ -19,7 +20,7 @@ def encode_mbi(value):
     return data
 
 
-def decode_mbi(data):
+def decode_mbi(data: bytes) -> Tuple[int, int]:
     value = 0
     for size, byte in enumerate(data):
         value |= (byte & 0x7F) << (size * 7)
@@ -28,30 +29,30 @@ def decode_mbi(data):
     raise XZError("invalid mbi")
 
 
-def crc32(data):
+def crc32(data: bytes) -> bytes:
     return pack("<I", crc32int(data))
 
 
-def round_up(value):
+def round_up(value: int) -> int:
     remainder = value % 4
     if remainder:
         return value - remainder + 4
     return value
 
 
-def pad(value):
+def pad(value: int) -> bytes:
     return b"\x00" * (round_up(value) - value)
 
 
-def create_xz_header(check):
+def create_xz_header(check: int) -> bytes:
     if not 0 <= check <= 0xF:
         raise XZError("header check")
     # stream header
     flags = pack("<BB", 0, check)
     return HEADER_MAGIC + flags + crc32(flags)
 
 
-def create_xz_index_footer(check, records):
+def create_xz_index_footer(check: int, records: List[Tuple[int, int]]) -> bytes:
     if not 0 <= check <= 0xF:
         raise XZError("footer check")
     # index
@@ -70,20 +71,23 @@ def create_xz_index_footer(check, records):
     return index + footer
 
 
-def parse_xz_header(header):
+def parse_xz_header(header: bytes) -> int:
     if len(header) != 12:
         raise XZError("header length")
     if header[:6] != HEADER_MAGIC:
         raise XZError("header magic")
     if crc32(header[6:8]) != header[8:12]:
         raise XZError("header crc32")
-    flag_first_byte, check = unpack("<BB", header[6:8])
+    flag_first_byte, check = cast(
+        Tuple[int, int],
+        unpack("<BB", header[6:8]),
+    )
     if flag_first_byte or not 0 <= check <= 0xF:
         raise XZError("header flags")
     return check
 
 
-def parse_xz_index(index):
+def parse_xz_index(index: bytes) -> List[Tuple[int, int]]:
     if len(index) < 8 or len(index) % 4:
         raise XZError("index length")
     index = memoryview(index)
@@ -115,14 +119,17 @@ def parse_xz_index(index):
     return records
 
 
-def parse_xz_footer(footer):
+def parse_xz_footer(footer: bytes) -> Tuple[int, int]:
     if len(footer) != 12:
         raise XZError("footer length")
     if footer[10:12] != FOOTER_MAGIC:
         raise XZError("footer magic")
     if crc32(footer[4:10]) != footer[:4]:
         raise XZError("footer crc32")
-    backward_size, flag_first_byte, check = unpack("<IBB", footer[4:10])
+    backward_size, flag_first_byte, check = cast(
+        Tuple[int, int, int],
+        unpack("<IBB", footer[4:10]),
+    )
     backward_size = (backward_size + 1) * 4
     if flag_first_byte or not 0 <= check <= 0xF:
         raise XZError("footer flags")