Better type handling

Author: Eeems

ext4/_compat.py

@@ -1,12 +1,27 @@
+from typing import Protocol
+from typing import runtime_checkable
+
 try:
-    from typing import override
+    from typing import override  # pyright: ignore[reportAssignmentType]
 
 except ImportError:
     from typing import Callable
     from typing import Any
 
-    def override(fn: Callable[..., Any]):
+    def override(fn: Callable[..., Any]):  # pyright: ignore[reportExplicitAny]
         return fn
 
 
-__all__ = ["override"]
+@runtime_checkable
+class ReadableStream(Protocol):
+    def read(self, size: int | None = -1, /) -> bytes: ...
+
+    def tell(self) -> int: ...
+
+
+@runtime_checkable
+class PeekableStream(ReadableStream, Protocol):
+    def peek(self, size: int = 0, /) -> bytes: ...
+
+
+__all__ = ["override", "ReadableStream"]

ext4/block.py

@@ -91,8 +91,8 @@ def tell(self) -> int:
         return self.cursor
 
     @override
-    def read(self, size: int = -1) -> bytes:
-        if size < 0:
+    def read(self, size: int | None = -1) -> bytes:
+        if size is None or size < 0:
             size = len(self) - self.cursor
 
         data = self.peek(size)

ext4/inode.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from codecs import _ReadableStream
 import io
 import warnings
 
@@ -9,7 +10,10 @@
 from ctypes import c_uint16
 from ctypes import sizeof
 
+from typing import cast
+
 from ._compat import override
+from ._compat import ReadableStream
 
 from .struct import crc32c
 from .struct import Ext4Struct
@@ -208,11 +212,11 @@ def block_size(self):
         return self.volume.block_size
 
     @property
-    def i_size(self):
+    def i_size(self) -> int:
         return self.i_size_high << 32 | self.i_size_lo
 
     @property
-    def i_file_acl(self):
+    def i_file_acl(self) -> int:
         return self.osd2.linux2.l_i_file_acl_high << 32 | self.i_file_acl_lo
 
     @property
@@ -303,24 +307,35 @@ def headers(self):
     def indices(self):
         return self.tree.indices
 
-    def _open(self, mode: str = "rb", encoding: None = None, newline: None = None):
+    def _open(
+        self, mode: str = "rb", encoding: None = None, newline: None = None
+    ) -> ReadableStream:
         if mode != "rb" or encoding is not None or newline is not None:
             raise NotImplementedError()
 
         if self.is_inline:
             self.volume.seek(self.offset + Inode.i_block.offset)
-            data = self.volume.read(self.i_size)
+            data = cast(bytes, self.volume.read(self.i_size))
             return io.BytesIO(data)
 
         return BlockIO(self)
 
-    def open(self, mode="rb", encoding=None, newline=None):
+    def open(
+        self,
+        mode: str = "rb",  # pyright: ignore[reportUnusedParameter]
+        encoding: None = None,  # pyright: ignore[reportUnusedParameter]
+        newline: None = None,  # pyright: ignore[reportUnusedParameter]
+    ) -> io.RawIOBase:
         raise NotImplementedError()
 
     @property
     def xattrs(self):
-        inline_offset = self.offset + self.EXT2_GOOD_OLD_INODE_SIZE + self.i_extra_isize
-        inline_size = self.offset + self.superblock.s_inode_size - inline_offset
+        inline_offset = cast(
+            int, self.offset + self.EXT2_GOOD_OLD_INODE_SIZE + self.i_extra_isize
+        )
+        inline_size = cast(
+            int, self.offset + self.superblock.s_inode_size - inline_offset
+        )
         if inline_size > sizeof(ExtendedAttributeIBodyHeader):
             try:
                 header = ExtendedAttributeIBodyHeader(self, inline_offset, inline_size)
@@ -364,7 +379,7 @@ class File(Inode):
     @override
     def open(
         self, mode: str = "rb", encoding: None = None, newline: None = None
-    ) -> io.BytesIO | BlockIO:
+    ) -> io.RawIOBase:
         return self._open(mode, encoding, newline)
 
 

ext4/struct.py

@@ -25,10 +25,10 @@ def to_hex(data):
 
 
 class Ext4Struct(LittleEndianStructure):
-    def __init__(self, volume, offset):
+    def __init__(self, volume, offset: int):
         super().__init__()
         self.volume = volume
-        self.offset = offset
+        self.offset: int = offset
         self.read_from_volume()
         self.verify()
 

ext4/volume.py

@@ -6,11 +6,11 @@
 
 from uuid import UUID
 from pathlib import PurePosixPath
-from typing import Callable
 
 from cachetools import cached
 from cachetools import LRUCache
 
+from ._compat import PeekableStream
 from .enum import EXT4_INO
 from .superblock import Superblock
 from .inode import Inode
@@ -41,7 +41,7 @@ def group(self, index):
         return group_index, table_entry_index
 
     @cached(cache=LRUCache(maxsize=32))
-    def offset(self, index):
+    def offset(self, index) -> int:
         group_index, table_entry_index = self.group(index)
         table_offset = (
             self.volume.group_descriptors[group_index].bg_inode_table * self.block_size
@@ -57,7 +57,7 @@ def __getitem__(self, index):
 class Volume(object):
     def __init__(
         self,
-        stream: io.Reader[bytes],
+        stream: PeekableStream,
         offset=0,
         ignore_flags=False,
         ignore_magic=False,
@@ -69,7 +69,7 @@ def __init__(
             if not hasattr(stream, name):
                 errors.append(f"{name} method missing")
 
-            elif not isinstance(getattr(stream, name), Callable):
+            elif not callable(getattr(stream, name)):  # pyright: ignore[reportAny]
                 errors.append(f"{name} is not a method")
 
         if errors:

ext4/xattr.py

@@ -5,6 +5,8 @@
 from ctypes import c_uint8
 from ctypes import sizeof
 
+from typing import cast
+
 from ._compat import override
 
 from .enum import EXT4_FEATURE_INCOMPAT
@@ -69,15 +71,15 @@ def __iter__(self):
                     warnings.warn(message, RuntimeWarning)
 
                 # TODO determine if e_value_size or i_size are required to limit results?
-                value = inode.open().read()
+                value = cast(bytes, inode.open().read())
 
             elif entry.e_value_size != 0:
                 value_offset = self.value_offset(entry)
                 if value_offset + entry.e_value_size > self.offset + self.data_size:
                     value = b""
                 else:
                     self.volume.seek(value_offset)
-                    value = self.volume.read(entry.e_value_size)
+                    value = cast(bytes, self.volume.read(entry.e_value_size))
             else:
                 value = b""
 
@@ -160,7 +162,7 @@ def size(self):
         return sizeof(self) + self.e_name_len
 
     @property
-    def name_str(self):
+    def name_str(self) -> str:
         name_index = self.e_name_index
         if 0 > name_index or name_index >= len(ExtendedAttributeEntry.NAME_INDICES):
             msg = f"Unknown attribute prefix {self.e_name_index:d}"

test.py

@@ -2,10 +2,11 @@
 
 import os
 import sys
-import ext4
+import ext4  # pyright: ignore[reportImplicitRelativeImport]
 
 from typing import cast
 from typing import Callable
+from typing import Any
 
 FAILED = False
 
@@ -21,29 +22,29 @@ def test_path_tuple(path: str | bytes, expected: tuple[bytes, ...]):
         print("pass")
 
     except Exception as e:
-        FAILED = True
+        FAILED = True  # pyright: ignore[reportConstantRedefinition]
         print("fail")
         print("  ", end="")
         print(e)
 
 
-def _assert(source: str, debug: Callable[[], str] | None = None):
+def _assert(source: str, debug: Callable[[], Any] | None = None):  # pyright: ignore[reportExplicitAny]
     global FAILED
     print(f"check {source}: ", end="")
     if eval(source):
         print("pass")
         return
 
-    FAILED = True
+    FAILED = True  # pyright: ignore[reportConstantRedefinition]
     print("fail")
     if debug is not None:
         print(f"  {debug()}")
 
 
 print("check ext4.Volume stream validation", end="")
 try:
-    ext4.Volume(1)
-    FAILED = True
+    _ = ext4.Volume(1)  # pyright: ignore[reportArgumentType]
+    FAILED = True  # pyright: ignore[reportConstantRedefinition]
     print("fail")
 
 except ext4.InvalidStreamException:
@@ -63,14 +64,14 @@ def _assert(source: str, debug: Callable[[], str] | None = None):
         try:
             print("check MagicError: ", end="")
             _ = ext4.Volume(f, offset=0)
-            FAILED = True
+            FAILED = True  # pyright: ignore[reportConstantRedefinition]
             print("fail")
             print("  MagicError not raised")
         except ext4.struct.MagicError:
             print("pass")
 
         except Exception as e:
-            FAILED = True
+            FAILED = True  # pyright: ignore[reportConstantRedefinition]
             print("fail")
             print("  ", end="")
             print(e)
@@ -84,7 +85,7 @@ def _assert(source: str, debug: Callable[[], str] | None = None):
             print("pass")
 
         except ext4.struct.ChecksumError as e:
-            FAILED = True
+            FAILED = True  # pyright: ignore[reportConstantRedefinition]
             print("fail")
             print("  ", end="")
             print(e)
@@ -111,7 +112,7 @@ def _assert(source: str, debug: Callable[[], str] | None = None):
         b = inode.open()
         data = b"hello world1\n"
         for x in range(1, 15):
-            _assert(f"b.peek({x}) == {data[:x]}", lambda: b.peek(x))
+            _assert(f"b.read({x}) == {data[:x]}", lambda: b.read(x))
 
 if FAILED:
     sys.exit(1)