Cheng Zhou
71 lines
1.8 KiB
Python
71 lines
1.8 KiB
Python
import uuid
|
|
from typing import Sequence
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.fee_attachment import FeeAttachment
|
|
|
|
|
|
async def create_attachment(
|
|
db: AsyncSession,
|
|
*,
|
|
entity_type: str,
|
|
entity_id: uuid.UUID,
|
|
file_type: str,
|
|
filename: str,
|
|
mime_type: str | None,
|
|
size: int,
|
|
storage_key: str | None,
|
|
url: str | None,
|
|
uploaded_by: uuid.UUID,
|
|
) -> FeeAttachment:
|
|
attachment = FeeAttachment(
|
|
entity_type=entity_type,
|
|
entity_id=entity_id,
|
|
file_type=file_type,
|
|
filename=filename,
|
|
mime_type=mime_type,
|
|
size=size,
|
|
storage_key=storage_key,
|
|
url=url,
|
|
uploaded_by=uploaded_by,
|
|
)
|
|
db.add(attachment)
|
|
await db.commit()
|
|
await db.refresh(attachment)
|
|
return attachment
|
|
|
|
|
|
async def list_attachments(
|
|
db: AsyncSession,
|
|
*,
|
|
entity_type: str,
|
|
entity_id: uuid.UUID,
|
|
) -> Sequence[FeeAttachment]:
|
|
result = await db.execute(
|
|
select(FeeAttachment)
|
|
.where(
|
|
FeeAttachment.entity_type == entity_type,
|
|
FeeAttachment.entity_id == entity_id,
|
|
FeeAttachment.is_deleted.is_(False),
|
|
)
|
|
.order_by(FeeAttachment.uploaded_at.desc())
|
|
)
|
|
return result.scalars().all()
|
|
|
|
|
|
async def get_attachment(db: AsyncSession, attachment_id: uuid.UUID) -> FeeAttachment | None:
|
|
result = await db.execute(
|
|
select(FeeAttachment).where(FeeAttachment.id == attachment_id, FeeAttachment.is_deleted.is_(False))
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def soft_delete_attachment(db: AsyncSession, attachment: FeeAttachment) -> FeeAttachment:
|
|
attachment.is_deleted = True
|
|
db.add(attachment)
|
|
await db.commit()
|
|
await db.refresh(attachment)
|
|
return attachment
|