Cheng Zhou
60 lines
1.5 KiB
Python
60 lines
1.5 KiB
Python
import uuid
|
|
from typing import Sequence
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.attachment import Attachment
|
|
|
|
|
|
async def create_attachment(
|
|
db: AsyncSession,
|
|
*,
|
|
study_id: uuid.UUID,
|
|
entity_type: str,
|
|
entity_id: uuid.UUID,
|
|
filename: str,
|
|
file_path: str,
|
|
file_size: int,
|
|
content_type: str | None,
|
|
uploaded_by: uuid.UUID,
|
|
) -> Attachment:
|
|
attachment = Attachment(
|
|
study_id=study_id,
|
|
entity_type=entity_type,
|
|
entity_id=entity_id,
|
|
filename=filename,
|
|
file_path=file_path,
|
|
file_size=file_size,
|
|
content_type=content_type,
|
|
uploaded_by=uploaded_by,
|
|
)
|
|
db.add(attachment)
|
|
await db.commit()
|
|
await db.refresh(attachment)
|
|
return attachment
|
|
|
|
|
|
async def list_attachments(
|
|
db: AsyncSession,
|
|
study_id: uuid.UUID,
|
|
entity_type: str,
|
|
entity_id: uuid.UUID,
|
|
) -> Sequence[Attachment]:
|
|
result = await db.execute(
|
|
select(Attachment)
|
|
.where(
|
|
Attachment.study_id == study_id,
|
|
Attachment.entity_type == entity_type,
|
|
Attachment.entity_id == entity_id,
|
|
Attachment.is_deleted.is_(False),
|
|
)
|
|
.order_by(Attachment.uploaded_at.desc())
|
|
)
|
|
return result.scalars().all()
|
|
|
|
|
|
async def get_attachment(db: AsyncSession, attachment_id: uuid.UUID) -> Attachment | None:
|
|
result = await db.execute(select(Attachment).where(Attachment.id == attachment_id, Attachment.is_deleted.is_(False)))
|
|
return result.scalar_one_or_none()
|