sqxp/backend/app/crud/attachment.py

"""附件CRUD"""
from typing import Optional, List
from sqlalchemy.orm import Session
from sqlalchemy import and_

from app.models.models import BizAttachment
from app.crud.base import CRUDBase
from app.core.utils import generate_id


class AttachmentCRUD(CRUDBase[BizAttachment]):
    """附件CRUD操作"""

    def get_by_business(
        self,
        db: Session,
        business_type: str,
        business_id: str
    ) -> List[BizAttachment]:
        """根据业务类型和业务ID获取附件列表"""
        return db.query(self.model).filter(
            and_(
                self.model.business_type == business_type,
                self.model.business_id == business_id,
                self.model.is_del == 0
            )
        ).order_by(self.model.create_at.desc()).all()

    def create_attachment(
        self,
        db: Session,
        business_type: str,
        business_id: str,
        file_name: str,
        file_url: str,
        file_size: int = None,
        file_type: str = None,
        upload_by: str = None,
        upload_by_name: str = None
    ) -> BizAttachment:
        """创建附件记录"""
        data = {
            "id": generate_id(),
            "business_type": business_type,
            "business_id": business_id,
            "file_name": file_name,
            "file_url": file_url,
            "file_size": file_size,
            "file_type": file_type,
            "upload_by": upload_by,
            "upload_by_name": upload_by_name,
            "create_by": upload_by,
            "create_name": upload_by_name
        }
        return self.create(db, data)


attachment_crud = AttachmentCRUD(BizAttachment)