51 lines
1.9 KiB
Python
51 lines
1.9 KiB
Python
from typing import List, Optional
|
|
|
|
from bson import ObjectId
|
|
from motor.motor_asyncio import AsyncIOMotorClient
|
|
|
|
from models.Character import Character
|
|
|
|
|
|
class CharacterRepo:
|
|
def __init__(self, client: AsyncIOMotorClient, db_name="bot_db"):
|
|
self.collection = client[db_name]["characters"]
|
|
|
|
async def add_character(self, character: Character) -> Character:
|
|
op = await self.collection.insert_one(character.model_dump())
|
|
character.id = str(op.inserted_id)
|
|
return character
|
|
|
|
async def get_character(self, character_id: str, with_image_data: bool = False) -> Character | None:
|
|
args = {}
|
|
if not with_image_data:
|
|
args["character_image_data"] = 0
|
|
res = await self.collection.find_one({"_id": ObjectId(character_id)}, args)
|
|
if res is None:
|
|
return None
|
|
else:
|
|
res["id"] = str(res.pop("_id"))
|
|
return Character(**res)
|
|
|
|
async def get_all_characters(self, created_by: Optional[str] = None, project_id: Optional[str] = None) -> List[Character]:
|
|
filter = {}
|
|
if created_by:
|
|
filter["created_by"] = created_by
|
|
if project_id:
|
|
filter["project_id"] = project_id
|
|
|
|
args = {"character_image_data": 0} # don't return image data for list
|
|
res = await self.collection.find(filter, args).to_list(None)
|
|
chars = []
|
|
for doc in res:
|
|
doc["id"] = str(doc.pop("_id"))
|
|
chars.append(Character(**doc))
|
|
return chars
|
|
|
|
async def update_char(self, char_id: str, character: Character) -> bool:
|
|
result = await self.collection.update_one({"_id": ObjectId(char_id)}, {"$set": character.model_dump()})
|
|
return result.modified_count > 0
|
|
|
|
async def delete_character(self, char_id: str) -> bool:
|
|
result = await self.collection.delete_one({"_id": ObjectId(char_id)})
|
|
return result.deleted_count > 0
|