Database file

Author: Youngestdev

ch06/planner/database/database.py

@@ -0,0 +1,59 @@
+from beanie import init_beanie, PydanticObjectId
+from motor.motor_asyncio import AsyncIOMotorClient
+from typing import Optional
+from pydantic import BaseSettings, BaseModel
+
+from models.users import User
+from models.events import Event
+
+class Settings(BaseSettings):
+    DATABASE_URL: Optional[str] = None
+
+    async def initialize_database(self):
+        client = AsyncIOMotorClient(self.DATABASE_URL)
+        await init_beanie(database=client.get_default_database(),
+                      document_models=[Event, User])
+
+    class Config:
+        env_file = ".env"
+
+
+class Database:
+    def __init__(self, model):
+        self.model = model
+    
+    async def save(self, document):
+        await document.create()
+        return
+
+    async def get(self, id: PydanticObjectId):
+        doc = await self.model.get(id)
+        if doc:
+            return doc
+        return False
+
+    async def get_all(self):
+        docs = await self.model.find_all().to_list()
+        return docs
+
+    async def update(self, id: PydanticObjectId, body: BaseModel):
+        doc_id = id
+        des_body = body.dict()
+
+        des_body = {k:v for k,v in des_body.items() if v is not None}
+        update_query = {"$set": {
+            field: value for field, value in des_body.items()
+        }}
+        
+        doc = await self.get(doc_id)
+        if not doc:
+            return False
+        await doc.update(update_query)
+        return doc
+
+    async def delete(self, id: PydanticObjectId):
+        doc = await self.get(id)
+        if not doc:
+            return False
+        await doc.delete()
+        return True