Cheng Zhou
49 lines
1.1 KiB
Python
49 lines
1.1 KiB
Python
import uuid
|
|
from typing import Sequence
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.comment import Comment
|
|
from app.schemas.comment import CommentCreate
|
|
|
|
|
|
async def create_comment(
|
|
db: AsyncSession,
|
|
study_id: uuid.UUID,
|
|
entity_type: str,
|
|
entity_id: uuid.UUID,
|
|
comment_in: CommentCreate,
|
|
created_by: uuid.UUID,
|
|
) -> Comment:
|
|
comment = Comment(
|
|
study_id=study_id,
|
|
entity_type=entity_type,
|
|
entity_id=entity_id,
|
|
content=comment_in.content,
|
|
created_by=created_by,
|
|
)
|
|
db.add(comment)
|
|
await db.commit()
|
|
await db.refresh(comment)
|
|
return comment
|
|
|
|
|
|
async def list_comments(
|
|
db: AsyncSession,
|
|
study_id: uuid.UUID,
|
|
entity_type: str,
|
|
entity_id: uuid.UUID,
|
|
) -> Sequence[Comment]:
|
|
result = await db.execute(
|
|
select(Comment)
|
|
.where(
|
|
Comment.study_id == study_id,
|
|
Comment.entity_type == entity_type,
|
|
Comment.entity_id == entity_id,
|
|
Comment.is_deleted.is_(False),
|
|
)
|
|
.order_by(Comment.created_at.asc())
|
|
)
|
|
return result.scalars().all()
|