54 lines
1.9 KiB
Python
54 lines
1.9 KiB
Python
from typing import Optional, List
|
|
from bson import ObjectId
|
|
from motor.motor_asyncio import AsyncIOMotorClient
|
|
from models.Idea import Idea
|
|
|
|
class IdeaRepo:
|
|
def __init__(self, client: AsyncIOMotorClient, db_name="bot_db"):
|
|
self.collection = client[db_name]["ideas"]
|
|
|
|
async def create_idea(self, idea: Idea) -> str:
|
|
res = await self.collection.insert_one(idea.model_dump())
|
|
return str(res.inserted_id)
|
|
|
|
async def get_idea(self, idea_id: str) -> Optional[Idea]:
|
|
if not ObjectId.is_valid(idea_id):
|
|
return None
|
|
res = await self.collection.find_one({"_id": ObjectId(idea_id)})
|
|
if res:
|
|
res["id"] = str(res.pop("_id"))
|
|
return Idea(**res)
|
|
return None
|
|
|
|
async def get_ideas(self, project_id: str, limit: int = 20, offset: int = 0) -> List[Idea]:
|
|
filter = {"project_id": project_id, "is_deleted": False}
|
|
res = await self.collection.find(filter).sort("updated_at", -1).skip(offset).limit(limit).to_list(None)
|
|
ideas = []
|
|
for doc in res:
|
|
doc["id"] = str(doc.pop("_id"))
|
|
ideas.append(Idea(**doc))
|
|
return ideas
|
|
|
|
async def delete_idea(self, idea_id: str) -> bool:
|
|
if not ObjectId.is_valid(idea_id):
|
|
return False
|
|
res = await self.collection.update_one(
|
|
{"_id": ObjectId(idea_id)},
|
|
{"$set": {"is_deleted": True}}
|
|
)
|
|
return res.modified_count > 0
|
|
|
|
async def update_idea(self, idea: Idea) -> bool:
|
|
if not idea.id or not ObjectId.is_valid(idea.id):
|
|
return False
|
|
|
|
idea_dict = idea.model_dump()
|
|
if "id" in idea_dict:
|
|
del idea_dict["id"]
|
|
|
|
res = await self.collection.update_one(
|
|
{"_id": ObjectId(idea.id)},
|
|
{"$set": idea_dict}
|
|
)
|
|
return res.modified_count > 0
|