Change linter to Ruff (#42)

Author: Schamper

dissect/esedb/__init__.py

@@ -7,9 +7,9 @@
 )
 
 __all__ = [
-    "EseDB",
-    "Error",
     "CompressedTaggedDataError",
+    "Error",
+    "EseDB",
     "InvalidDatabase",
     "KeyNotFoundError",
     "NoNeighbourPageError",

dissect/esedb/c_esedb.py

@@ -1,8 +1,9 @@
+from __future__ import annotations
+
 import datetime
 import struct
 import uuid
-from collections import namedtuple
-from typing import Union
+from typing import NamedTuple, Union
 
 from dissect.cstruct import cstruct
 
@@ -493,7 +494,12 @@ def checksum_xor(data: bytes, initial: int = 0x89ABCDEF) -> int:
     return digest
 
 
-ColumnType = namedtuple("ColumnType", ["value", "name", "size", "parse"])
+class ColumnType(NamedTuple):
+    value: int
+    name: str
+    size: int | None
+    parse: callable | None
+
 
 COLUMN_TYPES = [
     ColumnType(JET_coltyp.Nil, "NULL", 0, None),

dissect/esedb/compression.py

@@ -1,5 +1,6 @@
+from __future__ import annotations
+
 import struct
-from typing import Optional
 
 from dissect.util.compression import lzxpress, sevenbit
 
@@ -18,18 +19,17 @@ def decompress(buf: bytes) -> bytes:
     identifier = buf[0] >> 3
     if identifier == COMPRESSION_SCHEME.COMPRESS_7BITASCII:
         return sevenbit.decompress(buf[1:])
-    elif identifier == COMPRESSION_SCHEME.COMPRESS_7BITUNICODE:
+    if identifier == COMPRESSION_SCHEME.COMPRESS_7BITUNICODE:
         return sevenbit.decompress(buf[1:], wide=True)
-    elif identifier == COMPRESSION_SCHEME.COMPRESS_XPRESS:
+    if identifier == COMPRESSION_SCHEME.COMPRESS_XPRESS:
         return lzxpress.decompress(buf[3:])
-    elif identifier in (COMPRESSION_SCHEME.COMPRESS_XPRESS9, COMPRESSION_SCHEME.COMPRESS_XPRESS10):
+    if identifier in (COMPRESSION_SCHEME.COMPRESS_XPRESS9, COMPRESSION_SCHEME.COMPRESS_XPRESS10):
         raise NotImplementedError(f"Compression not yet implemented: {COMPRESSION_SCHEME(identifier)}")
-    else:
-        # Not compressed
-        return buf
+    # Not compressed
+    return buf
 
 
-def decompress_size(buf: bytes) -> Optional[int]:
+def decompress_size(buf: bytes) -> int | None:
     """Return the decompressed size of the given bytes according to the encoded compression scheme.
 
     Args:
@@ -41,10 +41,10 @@ def decompress_size(buf: bytes) -> Optional[int]:
     identifier = buf[0] >> 3
     if identifier == COMPRESSION_SCHEME.COMPRESS_7BITASCII:
         return ((buf[0] & 7) + (8 * len(buf))) // 7
-    elif identifier == COMPRESSION_SCHEME.COMPRESS_7BITUNICODE:
+    if identifier == COMPRESSION_SCHEME.COMPRESS_7BITUNICODE:
         return 2 * (((buf[0] & 7) + (8 * len(buf))) // 7)
-    elif identifier == COMPRESSION_SCHEME.COMPRESS_XPRESS:
+    if identifier == COMPRESSION_SCHEME.COMPRESS_XPRESS:
         return struct.unpack("<H", buf[1:2])[0]
-    elif identifier in (COMPRESSION_SCHEME.COMPRESS_XPRESS9, COMPRESSION_SCHEME.COMPRESS_XPRESS10):
+    if identifier in (COMPRESSION_SCHEME.COMPRESS_XPRESS9, COMPRESSION_SCHEME.COMPRESS_XPRESS10):
         raise NotImplementedError(f"Compression not yet implemented: {COMPRESSION_SCHEME(identifier)}")
     return None

dissect/esedb/cursor.py

@@ -1,12 +1,12 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Union
+from typing import TYPE_CHECKING
 
 from dissect.esedb.exceptions import KeyNotFoundError, NoNeighbourPageError
-from dissect.esedb.page import Node, Page
 
 if TYPE_CHECKING:
     from dissect.esedb.esedb import EseDB
+    from dissect.esedb.page import Node, Page
 
 
 class Cursor:
@@ -17,7 +17,7 @@ class Cursor:
         page: The page to open a cursor on.
     """
 
-    def __init__(self, esedb: EseDB, page: Union[int, Page]):
+    def __init__(self, esedb: EseDB, page: int | Page):
         self.esedb = esedb
 
         if isinstance(page, int):

dissect/esedb/esedb.py

@@ -1,15 +1,20 @@
 # Extensible Storage Engine (ESE) Database implementation
 # Combination of pre-source release reverse engineering and post-source release cleanup
 # Reference: https://github.com/microsoft/Extensible-Storage-Engine
+from __future__ import annotations
 
+import io
 from functools import cached_property, lru_cache
-from typing import BinaryIO, Iterator
+from typing import TYPE_CHECKING, BinaryIO
 
 from dissect.esedb.c_esedb import c_esedb, pgnoFDPMSO, ulDAEMagic
 from dissect.esedb.exceptions import InvalidDatabase
 from dissect.esedb.page import Page
 from dissect.esedb.table import Catalog, Table
 
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
 
 class EseDB:
     """EseDB class.
@@ -98,10 +103,6 @@ def page(self, num: int) -> Page:
 
     def pages(self) -> Iterator[Page]:
         """Iterate over all pages."""
-        num = 1
-        while True:
-            try:
-                yield self.page(num)
-                num += 1
-            except IndexError:
-                break
+        size = self.fh.seek(0, io.SEEK_END)
+        for i in range(1, (size // self.page_size) - 1):
+            yield self.page(i)

dissect/esedb/index.py

@@ -3,22 +3,22 @@
 import struct
 import uuid
 from functools import cached_property
-from typing import TYPE_CHECKING, Union
+from typing import TYPE_CHECKING
 
 from dissect.esedb.c_esedb import CODEPAGE, JET_bitIndex, JET_coltyp, RecordValue
 from dissect.esedb.cursor import Cursor
 from dissect.esedb.lcmapstring import map_string
-from dissect.esedb.page import Node, Page
 from dissect.esedb.record import Record
 
 if TYPE_CHECKING:
+    from dissect.esedb.page import Node, Page
     from dissect.esedb.table import Column, Table
 
 
 JET_cbKeyMost_OLD = 255
 
 
-class Index(object):
+class Index:
     """Represents an index on a table.
 
     This is still very much WIP but works for basic indexes.
@@ -266,7 +266,7 @@ def _encode_text(index: Index, column: Column, value: str, max_size: int) -> byt
     return bytes(key)
 
 
-def _encode_guid(value: Union[str, uuid.UUID]) -> bytes:
+def _encode_guid(value: str | uuid.UUID) -> bytes:
     if isinstance(value, str):
         value = uuid.UUID(value)
     guid_bytes = value.bytes_le
@@ -278,6 +278,5 @@ def _flip_bits(value: int, size: int) -> int:
     if value & (1 << (size - 1)):
         # If the high bit is set, all bits are flipped
         return ~value & ((1 << size) - 1)
-    else:
-        # Otherwise only the high bit is flipped
-        return value ^ (1 << (size - 1))
+    # Otherwise only the high bit is flipped
+    return value ^ (1 << (size - 1))

dissect/esedb/lcmapstring.py

@@ -181,7 +181,7 @@ def map_string(value: str, flags: MapFlags, locale: str) -> bytes:
             key_case.append(case_weight)
             continue
 
-        if script_member == SCRIPT.DIGIT and 0:
+        if script_member == SCRIPT.DIGIT:
             raise NotImplementedError(SCRIPT.DIGIT)
 
         # else
@@ -191,10 +191,7 @@ def map_string(value: str, flags: MapFlags, locale: str) -> bytes:
         key_case.append(case_weight)
 
     key_diacritic = _filter_weights(key_diacritic)
-    if flags & (MapFlags.NORM_IGNORECASE | MapFlags.NORM_IGNOREWIDTH):
-        key_case = []
-    else:
-        key_case = _filter_weights(key_case)
+    key_case = [] if flags & (MapFlags.NORM_IGNORECASE | MapFlags.NORM_IGNOREWIDTH) else _filter_weights(key_case)
 
     return bytes(
         [
@@ -212,7 +209,7 @@ def map_string(value: str, flags: MapFlags, locale: str) -> bytes:
     )
 
 
-def _filter_weights(weights):
+def _filter_weights(weights: list[int]) -> list[int]:
     i = len(weights)
     while i > 0:
         if weights[i - 1] > 2:

dissect/esedb/page.py

@@ -2,11 +2,13 @@
 
 import struct
 from functools import cached_property
-from typing import TYPE_CHECKING, Iterator, Optional, Union
+from typing import TYPE_CHECKING
 
 from dissect.esedb.c_esedb import PAGE_FLAG, TAG_FLAG, c_esedb
 
 if TYPE_CHECKING:
+    from collections.abc import Iterator
+
     from dissect.esedb.esedb import EseDB
 
 
@@ -81,10 +83,12 @@ def is_branch(self) -> bool:
         return not self.is_leaf
 
     @cached_property
-    def key_prefix(self) -> Optional[bytes]:
+    def key_prefix(self) -> bytes | None:
         if not self.is_root:
             return bytes(self.tag(0).data)
 
+        return None
+
     def tag(self, num: int) -> Tag:
         """Retrieve a tag by index.
 
@@ -104,7 +108,7 @@ def tags(self) -> Iterator[Tag]:
         for i in range(1, self.tag_count):
             yield self.tag(i)
 
-    def node(self, num: int) -> Union[BranchNode, LeafNode]:
+    def node(self, num: int) -> BranchNode | LeafNode:
         """Retrieve a node by index.
 
         Nodes are just tags, but indexed from the first tag.
@@ -123,7 +127,7 @@ def node(self, num: int) -> Union[BranchNode, LeafNode]:
 
         return self._node_cache[num]
 
-    def nodes(self) -> Iterator[Union[BranchNode, LeafNode]]:
+    def nodes(self) -> Iterator[BranchNode | LeafNode]:
         """Yield all nodes."""
         for i in range(self.node_count):
             yield self.node(i)
@@ -152,8 +156,7 @@ def iter_leaf_nodes(self) -> Iterator[LeafNode]:
                     yield leaf
 
         if self.is_root and leaf and leaf.tag.page.next_page:
-            for leaf in esedb.page(leaf.tag.page.next_page).iter_leaf_nodes():
-                yield leaf
+            yield from esedb.page(leaf.tag.page.next_page).iter_leaf_nodes()
 
     def __repr__(self) -> str:
         return f"<Page num={self.num:d}>"
@@ -167,7 +170,7 @@ class Tag:
         num: The tag number to parse.
     """
 
-    __slots__ = ("page", "num", "tag", "offset", "size", "data", "flags")
+    __slots__ = ("data", "flags", "num", "offset", "page", "size", "tag")
 
     def __init__(self, page: Page, num: int):
         self.page = page
@@ -210,7 +213,7 @@ class Node:
         tag: The :class:`Tag` to parse a node from.
     """
 
-    __slots__ = ("tag", "num", "key", "key_prefix", "key_suffix", "data")
+    __slots__ = ("data", "key", "key_prefix", "key_suffix", "num", "tag")
 
     def __init__(self, tag: Tag):
         self.tag = tag