move files

Author: mkrd

dictdatabase/orm.py dictdatabase/object_mapper.pydictdatabase/orm.py renamed to dictdatabase/object_mapper.py

@@ -1,9 +1,9 @@
 from __future__ import annotations
 from abc import ABC
-from typing import get_type_hints, get_origin, get_args, TypeVar, Tuple, Type
+from typing import get_type_hints, get_origin, get_args, TypeVar, Tuple, Type, Generic
 from types import UnionType, NoneType
 from . import io_safe
-from . sessions import SessionFileFull, SessionFileKey, SessionFileWhere, SessionDirFull, SessionDirWhere
+from .sessions import SessionFileFull, SessionFileKey, SessionFileWhere, SessionDirFull, SessionDirWhere
 
 
 
@@ -12,15 +12,13 @@
 
 
 def get_type_hints_excluding_internals(cls):
-	type_hints = {}
+	"""
+		Get type hints of the class, excluding double dunder variables.
+	"""
 	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
-
-
-
+		yield var_name, var_type
 
 
 
@@ -29,26 +27,31 @@ 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():
+	for var_name, var_type in get_type_hints_excluding_internals(cls):
+		var_type_args = get_args(var_type)
+		var_type_origin = get_origin(var_type)
 		# 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))
+			nullable = var_type_origin is UnionType and NoneType in var_type_args
 			if not nullable:
 				raise RuntimeError(f"Missing variable '{var_name}' in {cls.__name__}.")
-			else:
-				continue
+			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]
+		if var_type_origin is list and len(var_type_args) == 1:
+			item_type = var_type_args[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
 
 
 
+def fill_dict_from_object_using_type_hints(cls, obj):
+	raise NotImplementedError
+
+
+
 
 
 ########################################################################################
@@ -58,22 +61,24 @@ def fill_object_from_dict_using_type_hints(obj, cls, data: dict):
 
 
 
-class FileDictModel(ABC):
+class FileDictModel(ABC, Generic[T]):
 	"""
 		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_model__: T = None
+	__item_model__: Type[T]
 
 	@classmethod
 	def get_at_key(cls, key) -> T:
 		"""
 			Gets an item by key.
 			The data is partially read from the __file__.
 		"""
-		return cls.__item_model__.partial_read_by_key(key)
+		data = io_safe.partial_read(cls.__file__, key)
+		res: T = cls.__item_model__.from_key_value(key, data)
+		return res
 
 	@classmethod
 	def session_at_key(cls, key):
@@ -85,7 +90,7 @@ def items(cls) -> list[Tuple[str, T]]:
 			Gets all items as a list of tuples (key, ORM model of value).
 		"""
 		data = io_safe.read(cls.__file__)
-		return [(k, cls.__item_model__(k, v)) for k, v in data.items()]
+		return [(k, cls.__item_model__.from_key_value(k, v)) for k, v in data.items()]
 
 	@classmethod
 	def session(cls):
@@ -109,8 +114,6 @@ def get_where(cls, where: callable) -> list[Tuple[str, T]]:
 		"""
 		return [(k, v) for k, v in cls.items() if where(v)]
 
-	# Not Implemented:
-	# - select by filter callback (file_where): Not implemented because no performance advantage.
 
 
 
@@ -123,11 +126,6 @@ def from_key_value(cls: Type[T2], key, value) -> T2:
 		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)
-
 	@classmethod
 	def session(cls, key):
 		def partial_func(x):
@@ -152,7 +150,6 @@ def to_dict(self) -> dict:
 
 
 
-
 ########################################################################################
 # Scenario 2:
 # Add in a later version of DDB

test_orm.py

@@ -0,0 +1,122 @@
+from dictdatabase.object_mapper import DictModel, FileDictItemModel, FileDictModel
+import dictdatabase as DDB
+
+
+class WorkTime(DictModel):
+	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[User]):
+	__file__ = "users"
+	__item_model__ = User
+
+
+
+u = User.from_key_value("uid1", {
+	"first_name": "John",
+	"last_name": "Doe",
+	"none": "no",
+	"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)
+
+
+
+
+DDB.at("users").create({
+	"uid1": {
+		"first_name": "John",
+		"last_name": "Doe",
+		"none": "no",
+		"work_times": [
+			{"start": "08:00", "end": "12:00"},
+			{"start": "13:00", "end": "17:00"},
+		]
+	},
+	"uid2": {
+		"first_name": "Jane",
+		"last_name": "Smith",
+		"none": "no",
+		"work_times": [
+			{"start": "08:00", "end": "12:00"},
+			{"start": "13:00", "end": "17:00"},
+		]
+	},
+	"uid3": {
+		"first_name": "Pete",
+		"last_name": "Griffin",
+		"none": "no",
+		"work_times": [
+			{"start": "08:00", "end": "12:00"},
+			{"start": "13:00", "end": "17:00"},
+		]
+	}
+}, force_overwrite=True)
+
+
+u1 = Users.get_at_key("uid1")
+assert u1.first_name == "John"
+assert u1.last_name == "Doe"
+assert u1.full_name() == "John Doe"
+assert u1.work_times[0].start == "08:00"
+assert u1.work_times[0].end == "12:00"
+assert u1.work_times[1].start == "13:00"
+assert u1.work_times[1].end == "17:00"
+assert len(u1.work_times) == 2
+
+
+
+u2 = Users.get_at_key("uid2")
+
+
+
+for uid, u in Users.items():
+	print(u.full_name())
+
+
+
+# # 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_at_key("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]
+# 	...