ruff fixes

Author: mkrd

dictdatabase/io_safe.py

@@ -1,9 +1,11 @@
+from __future__ import annotations
+
 import os
 
 from . import config, io_unsafe, locking, utils
 
 
-def read(file_name: str) -> dict:
+def read(file_name: str) -> dict | None:
 	"""
 	Read the content of a file as a dict.
 
@@ -20,7 +22,7 @@ def read(file_name: str) -> dict:
 		return io_unsafe.read(file_name)
 
 
-def partial_read(file_name: str, key: str) -> dict:
+def partial_read(file_name: str, key: str) -> dict | None:
 	"""
 	Read only the value of a key-value pair from a file.
 

dictdatabase/locking.py

@@ -79,7 +79,7 @@ class FileLocksSnapshot:
 	On init, orphaned locks are removed.
 	"""
 
-	__slots__ = ("any_has_locks", "any_write_locks", "any_has_write_locks", "locks")
+	__slots__ = ("any_has_locks", "any_has_write_locks", "any_write_locks", "locks")
 
 	locks: list[LockFileMeta]
 	any_has_locks: bool
@@ -142,15 +142,15 @@ class AbstractLock:
 	provides a blueprint for derived classes to implement.
 	"""
 
-	__slots__ = ("db_name", "need_lock", "has_lock", "snapshot", "mode", "is_alivekeep_alive_thread")
+	__slots__ = ("db_name", "has_lock", "is_alive", "keep_alive_thread", "mode", "need_lock", "snapshot")
 
 	db_name: str
 	need_lock: LockFileMeta
 	has_lock: LockFileMeta
 	snapshot: FileLocksSnapshot
 	mode: str
 	is_alive: bool
-	keep_alive_thread: threading.Thread
+	keep_alive_thread: threading.Thread | None
 
 	def __init__(self, db_name: str) -> None:
 		# Normalize db_name to avoid file naming conflicts
@@ -197,7 +197,8 @@ def _start_keep_alive_thread(self) -> None:
 		"""
 
 		if self.keep_alive_thread is not None:
-			raise RuntimeError("Keep alive thread already exists.")
+			msg = "Keep alive thread already exists."
+			raise RuntimeError(msg)
 
 		self.is_alive = True
 		self.keep_alive_thread = threading.Thread(target=self._keep_alive_thread, daemon=False)
@@ -227,7 +228,7 @@ def _unlock(self) -> None:
 	def __enter__(self) -> None:
 		self._lock()
 
-	def __exit__(self, exc_type, exc_val, exc_tb) -> None:  # noqa: ANN001
+	def __exit__(self, exc_type, exc_val, exc_tb) -> None:
 		self._unlock()
 
 
@@ -248,7 +249,8 @@ def _lock(self) -> None:
 		# If this thread already holds a read lock, raise an exception.
 		if self.snapshot.exists(self.has_lock):
 			os.unlink(self.need_lock.path)
-			raise RuntimeError("Thread already has a read lock. Do not try to obtain a read lock twice.")
+			msg = "Thread already has a read lock. Do not try to obtain a read lock twice."
+			raise RuntimeError(msg)
 
 		start_time = time.time()
 
@@ -264,7 +266,8 @@ def _lock(self) -> None:
 				return
 			time.sleep(SLEEP_TIMEOUT)
 			if time.time() - start_time > AQUIRE_LOCK_TIMEOUT:
-				raise RuntimeError("Timeout while waiting for read lock.")
+				msg = "Timeout while waiting for read lock."
+				raise RuntimeError(msg)
 			self.snapshot = FileLocksSnapshot(self.need_lock)
 
 
@@ -285,7 +288,8 @@ def _lock(self) -> None:
 		# If this thread already holds a write lock, raise an exception.
 		if self.snapshot.exists(self.has_lock):
 			os.unlink(self.need_lock.path)
-			raise RuntimeError("Thread already has a write lock. Do not try to obtain a write lock twice.")
+			msg = "Thread already has a write lock. Do not try to obtain a write lock twice."
+			raise RuntimeError(msg)
 
 		start_time = time.time()
 
@@ -299,5 +303,6 @@ def _lock(self) -> None:
 				return
 			time.sleep(SLEEP_TIMEOUT)
 			if time.time() - start_time > AQUIRE_LOCK_TIMEOUT:
-				raise RuntimeError("Timeout while waiting for write lock.")
+				msg = "Timeout while waiting for write lock."
+				raise RuntimeError(msg)
 			self.snapshot = FileLocksSnapshot(self.need_lock)

dictdatabase/models.py

@@ -36,9 +36,11 @@ def __init__(self, path: str, key: str, where: Callable) -> None:
 		self.key = key is not None
 
 		if self.key and self.where:
-			raise TypeError("Cannot specify both key and where")
+			msg = "Cannot specify both key and where"
+			raise TypeError(msg)
 		if self.key and self.dir:
-			raise TypeError("Cannot specify sub-key when selecting a folder. Specify the key in the path instead.")
+			msg = "Cannot specify sub-key when selecting a folder. Specify the key in the path instead."
+			raise TypeError(msg)
 
 	@property
 	def file_normal(self) -> bool:
@@ -88,7 +90,7 @@ def at(*path, key: str | None = None, where: Callable[[Any, Any], bool] | None =
 
 
 class DDBMethodChooser:
-	__slots__ = ("path", "key", "where", "op_type")
+	__slots__ = ("key", "op_type", "path", "where")
 
 	path: str
 	key: str
@@ -124,7 +126,8 @@ def exists(self) -> bool:
 		As long it exists as a key in any dict, it will be found.
 		"""
 		if self.where is not None:
-			raise RuntimeError("DDB.at(where=...).exists() cannot be used with the where parameter")
+			msg = "DDB.at(where=...).exists() cannot be used with the where parameter"
+			raise RuntimeError(msg)
 
 		if not utils.file_exists(self.path):
 			return False
@@ -163,7 +166,8 @@ def delete(self) -> None:
 		Delete the file at the selected path.
 		"""
 		if self.where is not None or self.key is not None:
-			raise RuntimeError("DDB.at().delete() cannot be used with the where or key parameters")
+			msg = "DDB.at().delete() cannot be used with the where or key parameters"
+			raise RuntimeError(msg)
 		io_safe.delete(self.path)
 
 	def read(self, as_type: Type[T] | None = None) -> dict | T | None:
@@ -180,7 +184,7 @@ def type_cast(value):
 				return value
 			return as_type(value)
 
-		data = {}
+		data: dict = {}
 
 		if self.op_type.file_normal:
 			data = io_safe.read(self.path)

dictdatabase/sessions.py

@@ -13,12 +13,12 @@ def type_cast(obj, as_type):
 	return obj if as_type is None else as_type(obj)
 
 
-class SessionBase:
+class SessionBase[T]:
 	in_session: bool
 	db_name: str
 	as_type: T
 
-	def __init__(self, db_name: str, as_type):
+	def __init__(self, db_name: str, as_type: T) -> None:
 		self.in_session = False
 		self.db_name = db_name
 		self.as_type = as_type
@@ -27,7 +27,7 @@ def __enter__(self):
 		self.in_session = True
 		self.data_handle = {}
 
-	def __exit__(self, type, value, tb):
+	def __exit__(self, type, value, tb) -> None:
 		write_lock = getattr(self, "write_lock", None)
 		if write_lock is not None:
 			if isinstance(write_lock, list):

dictdatabase/utils.py

@@ -83,7 +83,8 @@ def seek_index_through_value_bytes(json_bytes: bytes, index: int) -> int:
 			while True:
 				i = json_bytes.find(byte_codes.QUOTE, i + 1)
 				if i == -1:
-					raise TypeError("Invalid JSON")
+					msg = "Invalid JSON"
+					raise TypeError(msg)
 
 				j = i - 1
 				backslash_count = 0
@@ -173,9 +174,9 @@ def find_outermost_key_in_json_bytes(json_bytes: bytes, key: str) -> Tuple[int,
 	# TODO: Very strict. the key must have a colon directly after it
 	# For example {"a": 1} will work, but {"a" : 1} will not work!
 
-	key = f'"{key}":'.encode()
+	key_bytes = f'"{key}":'.encode()
 
-	if (curr_i := json_bytes.find(key, 0)) == -1:
+	if (curr_i := json_bytes.find(key_bytes, 0)) == -1:
 		return (-1, -1)
 
 	# Assert: Key was found and curr_i is the index of the first character of the key
@@ -184,8 +185,8 @@ def find_outermost_key_in_json_bytes(json_bytes: bytes, key: str) -> Tuple[int,
 	key_nest = [(curr_i, count_nesting_in_bytes(json_bytes, 0, curr_i))]
 
 	# As long as more keys are found, keep track of them and their nesting level
-	while (next_i := json_bytes.find(key, curr_i + len(key))) != -1:
-		nesting = count_nesting_in_bytes(json_bytes, curr_i + len(key), next_i)
+	while (next_i := json_bytes.find(key_bytes, curr_i + len(key_bytes))) != -1:
+		nesting = count_nesting_in_bytes(json_bytes, curr_i + len(key_bytes), next_i)
 		key_nest.append((next_i, nesting))
 		curr_i = next_i
 
@@ -195,7 +196,7 @@ def find_outermost_key_in_json_bytes(json_bytes: bytes, key: str) -> Tuple[int,
 	# Early exit if there is only one key
 	if len(key_nest) == 1:
 		index, level = key_nest[0]
-		return (index, index + len(key)) if level == 1 else (-1, -1)
+		return (index, index + len(key_bytes)) if level == 1 else (-1, -1)
 
 	# Relative to total nesting
 	for i in range(1, len(key_nest)):
@@ -205,7 +206,7 @@ def find_outermost_key_in_json_bytes(json_bytes: bytes, key: str) -> Tuple[int,
 	indices_at_index_one = [i for i, level in key_nest if level == 1]
 	if len(indices_at_index_one) != 1:
 		return (-1, -1)
-	return (indices_at_index_one[0], indices_at_index_one[0] + len(key))
+	return (indices_at_index_one[0], indices_at_index_one[0] + len(key_bytes))
 
 
 def detect_indentation_in_json_bytes(json_bytes: bytes, index: int) -> Tuple[int, str]:

pyproject.toml

@@ -57,9 +57,15 @@ ignore = [
     "ANN201", # Missing return type annotation for public function
     "ARG001", # Unused function argument
     "T201", # `print` found
+    "ERA001", # Found commented-out code
 ]
 
 
 [tool.ruff.format]
 indent-style = "tab"
 quote-style = "double"
+
+
+
+[tool.mypy]
+ignore_missing_imports = true

tests/benchmark/run_parallel.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import json
 import os
 import random
@@ -6,6 +8,7 @@
 from calendar import c
 from dataclasses import dataclass
 from multiprocessing import Pool
+from typing import Callable
 
 from path_dict import PathDict
 
@@ -14,7 +17,7 @@
 DDB.config.storage_directory = ".ddb_bench_multi"
 
 
-def benchmark(iterations, setup: callable = None):
+def benchmark(iterations, setup: Callable | None = None):
 	def decorator(function):
 		def wrapper(*args, **kwargs):
 			f_name = function.__name__

tests/system_checks/test_monotonic_over_threads.py

@@ -17,7 +17,7 @@
 }
 
 # Queue to store timestamps in order
-timestamps = queue.Queue()
+timestamps: queue.Queue = queue.Queue()
 
 
 def capture_time(i, clock_func: Callable) -> None: