wip

Author: mkrd

dictdatabase/orm.py

@@ -1,90 +1,160 @@
 from __future__ import annotations
 from abc import ABC
-from typing import get_type_hints, get_origin, get_args, TypeVar
-from types import UnionType, NoneType,
+from typing import get_type_hints, get_origin, get_args, TypeVar, Tuple, Type
+from types import UnionType, NoneType
 from . import io_safe
 
-T = TypeVar('T')
 
 
+T = TypeVar("T")
+T2 = TypeVar("T2")
 
-class FolderBase(ABC):
-	__folder__ = None
-	__file__base__ = None
 
+def get_type_hints_excluding_internals(cls):
+	type_hints = {}
+	for var_name, var_type in get_type_hints(cls).items():
+		if var_name.startswith("__") and var_name.endswith("__"):
+			continue
+		type_hints[var_name] = var_type
+	return type_hints
 
-class FileDictBase(ABC):
+
+
+
+
+
+def fill_object_from_dict_using_type_hints(obj, cls, data: dict):
+	"""
+		Attributes of obj are set using the data dict.
+		The type hints of the class cls are used to determine which attributes to set.
+	"""
+	for var_name, var_type in get_type_hints_excluding_internals(cls).items():
+		# Check if variable is nullable (e.g. email: str | None)
+		nullable = get_origin(var_type) is UnionType and NoneType in get_args(var_type)
+		# When it is not nullable but not in the data, raise an error
+		if var_name not in data:
+			print(var_name, get_origin(var_type), get_args(var_type))
+			if not nullable:
+				raise RuntimeError(f"Missing variable '{var_name}' in {cls.__name__}.")
+			else:
+				continue
+		# When it is a list, fill the list with the items
+		if get_origin(var_type) is list and len(arg := get_args(var_type)) == 1:
+			item_type = arg[0]
+			setattr(obj, var_name, [item_type.from_dict(x) for x in data[var_name]])
+		else:
+			setattr(obj, var_name, data.get(var_name, None))
+	return obj
+
+
+
+
+
+########################################################################################
+# Scenario 1:
+# Model a single file with FileDictModel, which is a dict at the top level.
+# Each key-value item is modeled by a DictItemModel.
+
+
+
+class FileDictModel(ABC):
 	"""
 		A file base refers to a file that is stored in the database.
+		At the top level the file must contain a dictionary with strings as keys.
 	"""
 
 	__file__ = None
-	__item_base__: T = None
-	__create_file_on_write__ = False
-
+	__item_model__: T = None
 
 	@classmethod
 	def get_by_key(cls, key) -> T:
 		"""
 			Gets an item by key.
+			The data is partially read from the __file__.
 		"""
-		return cls.__item_base__.get_by_key(key)
-
+		return cls.__item_model__.partial_read_by_key(key)
 
+	@classmethod
+	def items(cls) -> list[Tuple[str, T]]:
+		"""
+			Gets all items.
+		"""
+		data = io_safe.read(cls.__file__)
+		return [cls.__item_model__(key, value) for key, value in data.items()]
 
+	@classmethod
+	def session(cls):
+		...
 
 
-class FileDictItemBase(ABC):
 	@classmethod
-	def get_by_key(cls, key):
-		data = io_safe.partial_read(cls.__file__, key)
-		return cls.from_data(data)
+	def session_at_key(cls, key):
+		...
+
 
 
 
 	@classmethod
-	def from_data(cls, data):
+	def get_where(cls, where: callable) -> list[T]:
+		"""
+			Gets all items where the where function returns True.
+			The where function takes an object of type __item_model__.
+		"""
+		return [item for item in cls.get_all() if where(item)]
 
-		instance = cls()
+	# Not Implemented:
+	# - select by filter callback (file_where): Not implemented because no performance advantage.
 
-		for var_name, var_type in get_type_hints(cls).items():
-			# Check if variable is nullable (e.g. email: str | None)
-			nullable = get_origin(var_type) is UnionType and NoneType in get_args(var_name)
-			# When it is not nullable but not in the data, raise an error
-			if var_name not in data and not nullable:
-				raise RuntimeError(f"Missing variable '{var_name}' in file '{cls.__file__}'.")
 
-			setattr(instance, var_name, data.get(var_name, None))
 
-		return instance
+class FileDictItemModel(ABC):
+	__key__: str
 
+	@classmethod
+	def from_key_value(cls: Type[T2], key, value) -> T2:
+		obj = fill_object_from_dict_using_type_hints(cls(), cls, value)
+		obj.__key__ = key
+		return obj
 
+	@classmethod
+	def read_by_key(cls, key) -> T:
+		data = io_safe.partial_read(cls.__file__, key)
+		return cls.from_key_value(key, data)
 
 
 
 
-class User(FileDictItemBase):
-	first_name: str | None
-	last_name: str
-	email: str
+class DictModel(ABC):
+
+	@classmethod
+	def from_dict(cls, data) -> DictModel:
+		obj = cls()
+		return fill_object_from_dict_using_type_hints(obj, cls, data)
+
 
-	def full_name(self):
-		return f"{self.first_name} {self.last_name}"
+	def to_dict(self) -> dict:
+		res = {}
+		for var_name in get_type_hints(self).keys():
+			if (value := getattr(self, var_name)) is not None:
+				res[var_name] = value
+		return res
 
 
-class Users(FileDictBase):
-	__file__ = "users"
-	__item_base__ = User
 
 
+########################################################################################
+# Scenario 2:
+# Add in a later version of DDB
+# A folder containing multiple files, each containing json.
 
-# Iterate FileDictBase
-# for user_id, user: User in Users.items():
-# 	print(user_id, user.first_name, user.last_name, user.email)
+# class FolderBase(ABC):
+# 	__folder__ = None
+# 	__file_model__: FileInFolderModel = None
 
-# # Get one item
-# user: User = Users.get("user_id")
 
+# class FileInFolderModel(ABC):
 
-# Get by lambda
-# users: Users = Users.where(lambda user: user.first_name != "John")
+# 	@classmethod
+# 	def get_by_name(cls, file_name: str) -> FileInFolderModel:
+# 		data = io_safe.read(f"{cls.__folder__}/{file_name}")
+# 		return cls(**data)

test_orm.py

@@ -0,0 +1,69 @@
+from dictdatabase.orm import DictModel, FileDictItemModel, FileDictModel
+
+
+class WorkTime(DictModel):
+	"""
+		Represents a work time.
+	"""
+	start: str
+	end: str
+
+
+class User(FileDictItemModel):
+	first_name: str
+	last_name: str
+	email: str | None
+
+	work_times: list[WorkTime]
+
+
+	def full_name(self):
+		return f"{self.first_name} {self.last_name}"
+
+
+class Users(FileDictModel):
+	__file__ = "users"
+	__item_model__ = User
+
+
+
+u = User.from_key_value("uid1", {
+	"first_name": "John",
+	"last_name": "Doe",
+	"work_times": [
+		{"start": "08:00", "end": "12:00"},
+		{"start": "13:00", "end": "17:00"},
+	]
+})
+
+
+assert u.first_name == "John"
+assert u.last_name == "Doe"
+assert u.full_name() == "John Doe"
+assert u.work_times[0].start == "08:00"
+assert u.work_times[0].end == "12:00"
+assert u.work_times[1].start == "13:00"
+assert u.work_times[1].end == "17:00"
+assert len(u.work_times) == 2
+
+print("check")
+print(u)
+
+
+# # Iterate FileDictModel
+# for user_id, user in Users.items():
+# 	print(user_id, user.first_name, user.last_name, user.email)
+
+# # Get one item
+# user: User = Users.get("user_id")
+
+
+# # Get by lambda
+# users: Users = Users.where(lambda user: user.first_name != "John")
+
+
+# with Users.session_at_key(user_id) as (session, user):
+# 	...
+
+# with Users.session() as (session, users): Dict[str, User]
+# 	...