Stricter checking

.github/workflows/test.yml

@@ -44,6 +44,16 @@ jobs:
       - name: uv tree
         run: uv tree
 
+      #----------------------------------------------
+      #---- Linting and Static Analysis
+      #----------------------------------------------
+
+      - name: 🔎 Run Ruff
+        run: uv run ruff check . # Or 'ruff format --check .' if you want formatting checks
+
+      - name: 🐍 Mypy Static Type Checker
+        run: uv run mypy .
+
       #----------------------------------------------
       #---- Pre-Checks
       #----------------------------------------------

DictDataBase.code-workspace

@@ -7,10 +7,14 @@
 	"settings": {
 		"[python]": {
 			"editor.formatOnSave": true,
-			"editor.defaultFormatter": "charliermarsh.ruff"
-		},
-		"editor.codeActionsOnSave": {
-			"source.organizeImports": "explicit"
+			"editor.defaultFormatter": "charliermarsh.ruff",
+			"editor.codeActionsOnSave": {
+				"source.fixAll": "never",
+				"source.organizeImports": "explicit",
+			},
 		},
+		"python.analysis.typeCheckingMode": "standard",
+		"python.analysis.diagnosticMode": "workspace",
+		"python.terminal.activateEnvironment": false,
 	}
 }

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/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

@@ -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
@@ -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,20 +142,20 @@ 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", "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
 		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")
 
@@ -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:
@@ -61,7 +63,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,
@@ -88,7 +90,7 @@ def at(*path, key: str = None, where: Callable[[Any, Any], bool] = None) -> DDBM
 
 
 class DDBMethodChooser:
-	__slots__ = ("path", "key", "where", "op_type")
+	__slots__ = ("key", "op_type", "path", "where")
 
 	path: str
 	key: str
@@ -98,8 +100,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 = []
@@ -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
@@ -146,13 +149,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 = {}
@@ -163,10 +166,11 @@ 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) -> 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.
 
@@ -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)
@@ -209,7 +213,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 +232,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 +243,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)

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):