75 lines
2.0 KiB
Python
75 lines
2.0 KiB
Python
"""Idea service for VoIdea."""
|
|
|
|
from typing import Optional
|
|
from uuid import uuid4
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.idea import Idea
|
|
|
|
|
|
class IdeaService:
|
|
def __init__(self, db: AsyncSession):
|
|
self.db = db
|
|
|
|
async def create(
|
|
self, user_id: str, title: str, content: str,
|
|
tags: Optional[list[str]] = None, is_public: bool = False,
|
|
) -> Idea:
|
|
idea = Idea(
|
|
id=uuid4(),
|
|
user_id=user_id,
|
|
title=title,
|
|
content=content,
|
|
tags=tags,
|
|
is_public=is_public,
|
|
status="draft",
|
|
)
|
|
self.db.add(idea)
|
|
await self.db.commit()
|
|
await self.db.refresh(idea)
|
|
return idea
|
|
|
|
async def get_by_id(self, idea_id: str) -> Optional[Idea]:
|
|
result = await self.db.execute(
|
|
select(Idea).where(Idea.id == idea_id)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
async def list_by_user(
|
|
self, user_id: str, skip: int = 0, limit: int = 50,
|
|
) -> list[Idea]:
|
|
result = await self.db.execute(
|
|
select(Idea)
|
|
.where(Idea.user_id == user_id)
|
|
.offset(skip)
|
|
.limit(limit)
|
|
.order_by(Idea.created_at.desc())
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
async def update(
|
|
self, idea_id: str, user_id: str, **kwargs,
|
|
) -> Optional[Idea]:
|
|
idea = await self.get_by_id(idea_id)
|
|
if not idea or str(idea.user_id) != user_id:
|
|
return None
|
|
|
|
for key, value in kwargs.items():
|
|
if value is not None and hasattr(idea, key):
|
|
setattr(idea, key, value)
|
|
|
|
await self.db.commit()
|
|
await self.db.refresh(idea)
|
|
return idea
|
|
|
|
async def delete(self, idea_id: str, user_id: str) -> bool:
|
|
idea = await self.get_by_id(idea_id)
|
|
if not idea or str(idea.user_id) != user_id:
|
|
return False
|
|
|
|
await self.db.delete(idea)
|
|
await self.db.commit()
|
|
return True
|