ruff

Author: mkrd

dictdatabase/io_bytes.py

@@ -1,10 +1,12 @@
+from __future__ import annotations
+
 import os
 import zlib
 
 from . import config, utils
 
 
-def read(db_name: str, *, start: int = None, end: int = None) -> bytes:
+def read(db_name: str, *, start: int | None = None, end: int | None = None) -> bytes:
 	"""
 	Read the content of a file as bytes. Reading works even when the config
 	changes, so a compressed ddb file can also be read if compression is
@@ -33,7 +35,8 @@ def read(db_name: str, *, start: int = None, end: int = None) -> bytes:
 
 	if json_exists:
 		if ddb_exists:
-			raise FileExistsError(f'Inconsistent: "{db_name}" exists as .json and .ddb.' "Please remove one of them.")
+			msg = f'Inconsistent: "{db_name}" exists as .json and .ddb.Please remove one of them.'
+			raise FileExistsError(msg)
 		with open(json_path, "rb") as f:
 			if start is None and end is None:
 				return f.read()
@@ -43,7 +46,8 @@ def read(db_name: str, *, start: int = None, end: int = None) -> bytes:
 				return f.read()
 			return f.read(end - start)
 	if not ddb_exists:
-		raise FileNotFoundError(f'No database file exists for "{db_name}"')
+		msg = f'No database file exists for "{db_name}"'
+		raise FileNotFoundError(msg)
 	with open(ddb_path, "rb") as f:
 		json_bytes = zlib.decompress(f.read())
 		if start is None and end is None:
@@ -53,7 +57,7 @@ def read(db_name: str, *, start: int = None, end: int = None) -> bytes:
 		return json_bytes[start:end]
 
 
-def write(db_name: str, dump: bytes, *, start: int = None) -> None:
+def write(db_name: str, dump: bytes, *, start: int | None = None) -> None:
 	"""
 	Write the bytes to the file of the db_path. If the db was compressed but no
 	compression is enabled, remove the compressed file, and vice versa.
@@ -72,7 +76,8 @@ def write(db_name: str, dump: bytes, *, start: int = None) -> None:
 	remove_file = None
 	if config.use_compression:
 		if start is not None:
-			raise RuntimeError("Cannot write to compressed file at a specific index")
+			msg = "Cannot write to compressed file at a specific index"
+			raise RuntimeError(msg)
 		write_file = ddb_path
 		if json_exists:
 			remove_file = json_path

dictdatabase/locking.py

@@ -40,7 +40,7 @@ class LockFileMeta:
 	Metadata representation for a lock file.
 	"""
 
-	__slots__ = ("ddb_dir", "name", "id", "time_ns", "stage", "mode", "path")
+	__slots__ = ("ddb_dir", "id", "mode", "name", "path", "stage", "time_ns")
 
 	ddb_dir: str
 	name: str
@@ -142,7 +142,7 @@ class AbstractLock:
 	provides a blueprint for derived classes to implement.
 	"""
 
-	__slots__ = ("db_name", "need_lock", "has_lock", "snapshot", "mode", "is_alive" "keep_alive_thread")
+	__slots__ = ("db_name", "need_lock", "has_lock", "snapshot", "mode", "is_alivekeep_alive_thread")
 
 	db_name: str
 	need_lock: LockFileMeta
@@ -155,7 +155,7 @@ class AbstractLock:
 	def __init__(self, db_name: str) -> None:
 		# Normalize db_name to avoid file naming conflicts
 		self.db_name = db_name.replace("/", "___").replace(".", "____")
-		time_ns = time.time_ns()
+		time_ns = str(time.time_ns())
 		t_id = f"{threading.get_native_id()}"  # ID that's unique across processes and threads.
 		dir = os.path.join(config.storage_directory, ".ddb")
 

dictdatabase/models.py

@@ -61,7 +61,7 @@ def dir_where(self) -> bool:
 		return self.dir and self.where and not self.key
 
 
-def at(*path, key: str = None, where: Callable[[Any, Any], bool] = None) -> DDBMethodChooser:
+def at(*path, key: str | None = None, where: Callable[[Any, Any], bool] | None = None) -> DDBMethodChooser:
 	"""
 	Select a file or folder to perform an operation on.
 	If you want to select a specific key in a file, use the `key` parameter,
@@ -98,8 +98,8 @@ class DDBMethodChooser:
 	def __init__(
 		self,
 		path: tuple,
-		key: str = None,
-		where: Callable[[Any, Any], bool] = None,
+		key: str | None = None,
+		where: Callable[[Any, Any], bool] | None = None,
 	) -> None:
 		# Convert path to a list of strings
 		pc = []
@@ -146,13 +146,13 @@ def create(self, data: dict | None = None, force_overwrite: bool = False) -> Non
 		exists, defaults to False (optional).
 		"""
 		if self.where is not None or self.key is not None:
-			raise RuntimeError("DDB.at().create() cannot be used with the where or key parameters")
+			msg = "DDB.at().create() cannot be used with the where or key parameters"
+			raise RuntimeError(msg)
 
 		# Except if db exists and force_overwrite is False
 		if not force_overwrite and self.exists():
-			raise FileExistsError(
-				f"Database {self.path} already exists in {config.storage_directory}. Pass force_overwrite=True to overwrite."
-			)
+			msg = f"Database {self.path} already exists in {config.storage_directory}. Pass force_overwrite=True to overwrite."
+			raise FileExistsError(msg)
 		# Write db to file
 		if data is None:
 			data = {}
@@ -166,7 +166,7 @@ def delete(self) -> None:
 			raise RuntimeError("DDB.at().delete() cannot be used with the where or key parameters")
 		io_safe.delete(self.path)
 
-	def read(self, as_type: Type[T] = None) -> dict | T | None:
+	def read(self, as_type: Type[T] | None = None) -> dict | T | None:
 		"""
 		Reads a file or folder depending on previous `.at(...)` selection.
 
@@ -209,7 +209,7 @@ def type_cast(value):
 		return type_cast(data)
 
 	def session(
-		self, as_type: Type[T] = None
+		self, as_type: Type[T] | None = None
 	) -> SessionFileFull[T] | SessionFileKey[T] | SessionFileWhere[T] | SessionDirFull[T] | SessionDirWhere[T]:
 		"""
 		Opens a session to the selected file(s) or folder, depending on previous
@@ -228,6 +228,7 @@ def session(
 		Returns:
 		- Tuple of (session_object, data)
 		"""
+
 		if self.op_type.file_normal:
 			return SessionFileFull(self.path, as_type)
 		if self.op_type.file_key:
@@ -238,3 +239,6 @@ def session(
 			return SessionDirFull(self.path, as_type)
 		if self.op_type.dir_where:
 			return SessionDirWhere(self.path, self.where, as_type)
+
+		msg = "Invalid operation type"
+		raise RuntimeError(msg)

tests/benchmark/locking.py

@@ -15,24 +15,24 @@
 # 25.11.22: 4156ms
 with profiler.Profiler() as p:
 	for _ in range(25_000):
-		l = locking.ReadLock("db")
-		l._lock()
-		l._unlock()
+		read_lock = locking.ReadLock("db")
+		read_lock._lock()  # noqa: SLF001
+		read_lock._unlock()  # noqa: SLF001
 p.open_in_browser()
 
 
 # 05.11.22: 4884ms
 # 25.11.22: 4159ms
 with profiler.Profiler() as p:
 	for _ in range(25_000):
-		l = locking.WriteLock("db")
-		l._lock()
-		l._unlock()
+		write_lock = locking.WriteLock("db")
+		write_lock._lock()  # noqa: SLF001
+		write_lock._unlock()  # noqa: SLF001
 p.open_in_browser()
 
 
-l = locking.WriteLock("db/test.some")
-l._lock()
+write_lock = locking.WriteLock("db/test.some")
+write_lock._lock()  # noqa: SLF001
 
 
 shutil.rmtree(DDB.config.storage_directory)

tests/system_checks/test_monotonic_over_threads.py

@@ -1,6 +1,7 @@
 import queue
 import threading
 import time
+from typing import Callable
 
 # Number of threads
 NUM_THREADS = 64
@@ -19,14 +20,14 @@
 timestamps = queue.Queue()
 
 
-def capture_time(i, clock_func: callable) -> None:
+def capture_time(i, clock_func: Callable) -> None:
 	# Capture time using the given clock function and put it in the queue
 	for _ in range(1000):
 		# print(f"Thread {i} capturing time")
 		timestamps.put(clock_func())
 
 
-def check_monotonicity_for_clock(clock_name: str, clock_func: callable) -> None:
+def check_monotonicity_for_clock(clock_name: str, clock_func: Callable) -> None:
 	# Clear the queue for the next clock
 	while not timestamps.empty():
 		timestamps.get()

tests/system_checks/test_tick_rate.py

@@ -1,7 +1,8 @@
 import time
+from typing import Callable
 
 
-def get_tick_rate(clock_func: callable) -> float:
+def get_tick_rate(clock_func: Callable) -> float:
 	start_time = time.time()
 	measurements = [clock_func() for _ in range(2_000_000)]
 	end_time = time.time()
@@ -10,7 +11,8 @@ def get_tick_rate(clock_func: callable) -> float:
 	prev_value = measurements[0]
 	for current_value in measurements[1:]:
 		if current_value < prev_value:
-			raise RuntimeError("Clock function is not monotonic")
+			msg = "Clock function is not monotonic"
+			raise RuntimeError(msg)
 		if current_value != prev_value:
 			ticks += 1
 		prev_value = current_value