Add aggregate(), insert_many() and save() Queries

panther/db/connections.py

@@ -21,7 +21,7 @@
     from pymongo.database import Database
 
 
-class DatabaseConnection:
+class BaseDatabaseConnection:
     def __init__(self, *args, **kwargs):
         """Initialized in application startup"""
         self.init(*args, **kwargs)
@@ -36,7 +36,7 @@ def session(self):
         pass
 
 
-class MongoDBConnection(DatabaseConnection):
+class MongoDBConnection(BaseDatabaseConnection):
     def init(
             self,
             host: str = 'localhost',
@@ -73,7 +73,7 @@ def session(self):
         return self._database
 
 
-class PantherDBConnection(DatabaseConnection):
+class PantherDBConnection(BaseDatabaseConnection):
     def init(self, path: str | None = None, encryption: bool = False):
         params = {'db_name': path, 'return_dict': True}
         if encryption:
@@ -90,7 +90,7 @@ def session(self):
         return self._connection
 
 
-class DatabaseSession(Singleton):
+class DatabaseConnection(Singleton):
     @property
     def session(self):
         return config['database'].session
@@ -130,5 +130,5 @@ def create_connection_for_websocket(self) -> _Redis:
         return self.websocket_connection
 
 
-db: DatabaseSession = DatabaseSession()
+db: DatabaseConnection = DatabaseConnection()
 redis: RedisConnection = RedisConnection()

panther/db/cursor.py

@@ -16,8 +16,9 @@ class Cursor(_Cursor):
     models = {}
 
     def __init__(self, collection, *args, cls=None, **kwargs):
+        # cls.__name__ and collection.name are equal.
         if cls:
-            self.models[cls.__name__] = cls
+            self.models[collection.name] = cls
             self.cls = cls
         else:
             self.cls = self.models[collection.name]

panther/db/models.py

@@ -56,7 +56,7 @@ def dict(self, *args, **kwargs) -> dict:
 class BaseUser(Model):
     first_name: str = Field('', max_length=64)
     last_name: str = Field('', max_length=64)
-    last_login: datetime = None
+    last_login: datetime | None = None
 
     async def update_last_login(self) -> None:
         await self.update(last_login=datetime.now())

panther/db/queries/base_queries.py

@@ -0,0 +1,126 @@
+import operator
+from abc import abstractmethod
+from functools import reduce
+from sys import version_info
+
+from pydantic_core._pydantic_core import ValidationError
+
+from panther.db.cursor import Cursor
+from panther.db.utils import prepare_id_for_query
+from panther.exceptions import DatabaseError
+
+if version_info >= (3, 11):
+    from typing import Self, Iterator
+else:
+    from typing import TypeVar
+
+    Self = TypeVar('Self', bound='BaseQuery')
+
+
+class BaseQuery:
+    @classmethod
+    def _merge(cls, *args, is_mongo: bool = False) -> dict:
+        prepare_id_for_query(*args, is_mongo=is_mongo)
+        return reduce(operator.ior, filter(None, args), {})
+
+    @classmethod
+    def _clean_error_message(cls, validation_error: ValidationError, is_updating: bool = False) -> str:
+        error = ', '.join(
+            '{field}="{error}"'.format(
+                field='.'.join(loc for loc in e['loc']),
+                error=e['msg']
+            )
+            for e in validation_error.errors()
+            if not is_updating or e['type'] != 'missing'
+        )
+        return f'{cls.__name__}({error})' if error else ''
+
+    @classmethod
+    def _validate_data(cls, *, data: dict, is_updating: bool = False):
+        """Validate document before inserting to collection"""
+        try:
+            cls(**data)
+        except ValidationError as validation_error:
+            if error := cls._clean_error_message(validation_error=validation_error, is_updating=is_updating):
+                raise DatabaseError(error)
+
+    @classmethod
+    def _create_model_instance(cls, document: dict):
+        """Prevent getting errors from document insertion"""
+        try:
+            return cls(**document)
+        except ValidationError as validation_error:
+            if error := cls._clean_error_message(validation_error=validation_error):
+                raise DatabaseError(error)
+
+    @classmethod
+    @abstractmethod
+    async def find_one(cls, *args, **kwargs) -> Self | None:
+        raise NotImplementedError
+
+    @classmethod
+    @abstractmethod
+    async def find(cls, *args, **kwargs) -> list[Self] | Cursor:
+        raise NotImplementedError
+
+    @classmethod
+    @abstractmethod
+    async def first(cls, *args, **kwargs) -> Self | None:
+        raise NotImplementedError
+
+    @classmethod
+    @abstractmethod
+    async def last(cls, *args, **kwargs):
+        raise NotImplementedError
+
+    @classmethod
+    @abstractmethod
+    async def aggregate(cls, *args, **kwargs) -> Iterator[dict]:
+        raise NotImplementedError
+
+    # # # # # Count # # # # #
+    @classmethod
+    @abstractmethod
+    async def count(cls, *args, **kwargs) -> int:
+        raise NotImplementedError
+
+    # # # # # Insert # # # # #
+    @classmethod
+    @abstractmethod
+    async def insert_one(cls, *args, **kwargs) -> Self:
+        raise NotImplementedError
+
+    @classmethod
+    @abstractmethod
+    async def insert_many(cls, *args, **kwargs) -> list[Self]:
+        raise NotImplementedError
+
+    # # # # # Delete # # # # #
+    @abstractmethod
+    async def delete(self) -> None:
+        raise NotImplementedError
+
+    @classmethod
+    @abstractmethod
+    async def delete_one(cls, *args, **kwargs) -> bool:
+        raise NotImplementedError
+
+    @classmethod
+    @abstractmethod
+    async def delete_many(cls, *args, **kwargs) -> int:
+        raise NotImplementedError
+
+    # # # # # Update # # # # #
+    @abstractmethod
+    async def update(self, *args, **kwargs) -> None:
+        raise NotImplementedError
+
+    @classmethod
+    @abstractmethod
+    async def update_one(cls, *args, **kwargs) -> bool:
+        raise NotImplementedError
+
+    @classmethod
+    @abstractmethod
+    async def update_many(cls, *args, **kwargs) -> int:
+        raise NotImplementedError