docs/v1.0/server/changelogs/2026-01-28-attachment-service-spec-compliance.md

附件服务文档规范符合性修复

日期：2026-01-28
类型：文档修复
影响范围：附件服务文档

---

变更概述

对照 jointo-tech-stack 规范，全面修复附件服务文档，确保符合项目架构约束和编码规范。

---

修复内容

1. 数据库设计修复

移除物理外键约束

修复前：

project_id TEXT NOT NULL REFERENCES folders(id) ON DELETE CASCADE
uploaded_by INTEGER REFERENCES users(id)

修复后：

-- 无外键约束，应用层验证
project_id UUID,  -- 应用层验证
uploaded_by UUID NOT NULL,  -- 应用层验证

UUID 生成方式修正

修复前：

id TEXT PRIMARY KEY DEFAULT gen_uuid_v7()

修复后：

# 应用层生成 UUID v7
attachment_id: UUID = Field(
    sa_column=Column(
        PG_UUID(as_uuid=True),
        primary_key=True,
        default=generate_uuid  # 应用层生成
    )
)

主键类型修正

修复前：

attachment_id = Column(Integer, primary_key=True, autoincrement=True)

修复后：

attachment_id: UUID = Field(
    sa_column=Column(PG_UUID(as_uuid=True), primary_key=True, default=generate_uuid)
)

添加表和列注释

COMMENT ON TABLE attachments IS '附件表 - 应用层保证引用完整性';
COMMENT ON COLUMN attachments.project_id IS '项目 ID - 应用层验证';
COMMENT ON COLUMN attachments.uploaded_by IS '上传者用户 ID - 应用层验证';

优化索引策略

-- 组合索引
CREATE INDEX idx_attachments_project_category ON attachments (project_id, category);
CREATE INDEX idx_attachments_uploader_created ON attachments (uploaded_by, created_at DESC);

2. Model 定义修复

改用 SQLModel

修复前：

from app.core.database import Base
class Attachment(Base):

修复后：

from sqlmodel import SQLModel
class Attachment(SQLModel, table=True):

时间戳修正

修复前：

created_at = Column(DateTime, default=datetime.utcnow)

修复后：

created_at: datetime = Field(
    default_factory=lambda: datetime.now(timezone.utc)
)

添加 Relationship primaryjoin

修复前：

project = relationship("Project", back_populates="attachments")

修复后：

project: Optional["Project"] = Relationship(
    sa_relationship_kwargs={
        "primaryjoin": "Attachment.project_id == Project.project_id",
        "foreign_keys": "[Attachment.project_id]",
    }
)

补充 to_dict() 方法

def to_dict(self) -> dict:
    """转换为字典（camelCase 字段名）"""
    return {
        'id': str(self.attachment_id),
        'name': self.name,
        'category': AttachmentCategory(self.category).to_string(),
        'categoryDisplay': AttachmentCategory.get_display_name(self.category),
        # ...
    }

3. 枚举类型修复

补充 to_string() 方法

修复前：

class AttachmentCategory(IntEnum):
    @classmethod
    def from_string(cls, value: str) -> int:
        # 只有 from_string

修复后：

class AttachmentCategory(IntEnum):
    @classmethod
    def from_string(cls, value: str) -> "AttachmentCategory":
        mapping = {'document': cls.DOCUMENT, 'image': cls.IMAGE}
        return mapping.get(value.lower(), cls.DOCUMENT)
    
    def to_string(self) -> str:
        """枚举转字符串"""
        mapping = {self.DOCUMENT: 'document', self.IMAGE: 'image'}
        return mapping.get(self, 'document')
    
    @classmethod
    def get_display_name(cls, value: int) -> str:
        """获取显示名称"""
        names = {cls.DOCUMENT: '文档', cls.IMAGE: '图片'}
        return names.get(value, '未知类型')

4. Service 层修复

添加引用完整性验证

修复前：

async def upload_attachment(..., project_id: Optional[int] = None):
    # 直接使用 project_id，未验证
    attachment = Attachment(project_id=project_id, ...)

修复后：

async def upload_attachment(..., project_id: Optional[UUID] = None):
    # 1. 验证用户存在
    if not await self.repository.exists_user(user_id):
        raise NotFoundError(f"用户不存在: {user_id}")
    
    # 2. 验证项目存在（如果提供）
    if project_id:
        if not await self.repository.exists_project(project_id):
            raise NotFoundError(f"项目不存在: {project_id}")
        
        # 验证用户对项目的权限
        has_permission = await self.repository.check_project_permission(
            user_id, project_id, 'editor'
        )
        if not has_permission:
            raise PermissionError("没有权限在该项目下上传附件")

改用 AsyncSession

修复前：

def __init__(self, db: Session):
    self.db = db

修复后：

def __init__(self, session: AsyncSession):
    self.session = session

5. Repository 层补充

添加引用完整性验证方法

class AttachmentRepository:
    async def exists_user(self, user_id: UUID) -> bool:
        """检查用户是否存在（应用层引用完整性）"""
        from app.models.user import User
        result = await self.session.execute(
            select(User.user_id).where(
                User.user_id == user_id,
                User.deleted_at.is_(None)
            ).limit(1)
        )
        return result.scalar_one_or_none() is not None
    
    async def exists_project(self, project_id: UUID) -> bool:
        """检查项目是否存在（应用层引用完整性）"""
        from app.models.project import Project
        result = await self.session.execute(
            select(Project.project_id).where(
                Project.project_id == project_id,
                Project.deleted_at.is_(None)
            ).limit(1)
        )
        return result.scalar_one_or_none() is not None
    
    async def check_project_permission(
        self,
        user_id: UUID,
        project_id: UUID,
        required_role: str = 'viewer'
    ) -> bool:
        """检查用户对项目的权限"""
        from app.repositories.project_repository import ProjectRepository
        project_repo = ProjectRepository(self.session)
        return await project_repo.check_user_permission(
            user_id, project_id, required_role
        )

6. Schema 层补充

完整的 Pydantic Schema 定义

# 请求 Schema
class AttachmentUploadRequest(BaseModel):
    category: str = Field(default='document')
    project_id: Optional[UUID] = Field(None, alias="projectId")
    is_public: bool = Field(default=False, alias="isPublic")
    
    @field_validator('category')
    @classmethod
    def validate_category(cls, v: str) -> str:
        if v not in ['document', 'image']:
            raise ValueError('文件类别必须是 document 或 image')
        return v

# 响应 Schema
class AttachmentResponse(BaseModel):
    id: str
    name: str
    original_name: str = Field(alias="originalName")
    file_url: str = Field(alias="fileUrl")
    # ... 完整字段定义
    
    class Config:
        populate_by_name = True

7. API 层补充

FastAPI 路由实现

router = APIRouter(prefix="/attachments", tags=["attachments"])

@router.post("", response_model=ApiResponse[AttachmentResponse])
async def upload_attachment(
    file: Annotated[UploadFile, File(...)],
    category: Annotated[str, Form()] = 'document',
    project_id: Annotated[str | None, Form(alias="projectId")] = None,
    is_public: Annotated[bool, Form(alias="isPublic")] = False,
    current_user: Annotated[User, Depends(get_current_user)] = None,
    service: Annotated[AttachmentService, Depends(get_attachment_service)] = None
):
    """上传附件"""
    project_uuid = UUID(project_id) if project_id else None
    
    attachment = await service.upload_attachment(
        user_id=current_user.user_id,
        file=file,
        category=category,
        project_id=project_uuid,
        is_public=is_public
    )
    
    return ApiResponse.success(data=attachment.to_dict())

统一响应格式

修复前：

return await self.repository.create(attachment)

修复后：

return ApiResponse.success(data=attachment.to_dict())

---

规范符合度

修复前

• ❌ 使用物理外键约束
• ❌ 使用数据库 server_default 生成 UUID
• ❌ 使用 SQLAlchemy Base
• ❌ 使用 datetime.utcnow
• ❌ 缺少 Relationship primaryjoin
• ❌ 缺少引用完整性验证
• ❌ 枚举类型不完整
• ❌ 缺少 Schema 和 API 实现

修复后

• ✅ 应用层保证引用完整性
• ✅ 应用层生成 UUID v7
• ✅ 使用 SQLModel
• ✅ 使用 datetime.now(timezone.utc)
• ✅ Relationship 配置 primaryjoin
• ✅ Service 层验证引用存在
• ✅ 枚举类型完整实现
• ✅ 完整的 Schema 和 API 实现

---

影响范围

• ✅ 文档更新：docs/requirements/backend/04-services/resource/attachment-service.md
• ⚠️ 代码实现：需要按照更新后的文档实现代码
• ⚠️ 数据库迁移：需要创建符合规范的迁移脚本

---

后续工作

1. 按照更新后的文档实现附件服务代码
2. 创建数据库迁移脚本（无外键约束）
3. 编写单元测试和集成测试
4. 验证引用完整性保证机制

---

变更日期：2026-01-28
文档版本：v3.0