xiaoxia
208deb4d39
新增 cover_templates 表及完整 CRUD 接口：
- GET /api/v1/cover-templates 列出系统+用户模板
- POST /api/v1/cover-templates 创建自定义模板
- PUT /api/v1/cover-templates/{id} 更新模板（系统模板不可改）
- DELETE /api/v1/cover-templates/{id} 删除自定义模板（系统模板不可删）
技术实现：
- 领域实体 CoverTemplate（packages/domain/cover_template.py）
- 仓储接口 + SQLAlchemy 实现 + InMemory 实现
- Schema 对齐前端 CoverTemplate 类型定义
- Alembic migration 055，含 8 个系统模板 seed 数据
- 28 个新单测，全量 13704 单测通过
106 lines
3.7 KiB
Python
106 lines
3.7 KiB
Python
"""封面模板 SQLAlchemy 仓储实现。"""
|
|
|
|
from sqlalchemy import or_
|
|
from sqlalchemy.orm import Session
|
|
|
|
from packages.adapters.sqlalchemy_impl.models import CoverTemplateModel
|
|
from packages.domain.cover_template import CoverTemplate
|
|
|
|
|
|
class SQLAlchemyCoverTemplateRepository:
|
|
"""封面模板仓储实现。"""
|
|
|
|
def __init__(self, session: Session):
|
|
self.session = session
|
|
|
|
def create(self, template: CoverTemplate) -> CoverTemplate:
|
|
model = CoverTemplateModel(
|
|
id=template.id,
|
|
user_id=template.user_id,
|
|
name=template.name,
|
|
thumbnail_url=template.thumbnail_url,
|
|
is_system=template.is_system,
|
|
config=template.config,
|
|
created_at=template.created_at,
|
|
updated_at=template.updated_at,
|
|
)
|
|
self.session.add(model)
|
|
self.session.commit()
|
|
return template
|
|
|
|
def get(self, template_id: str) -> CoverTemplate | None:
|
|
model = self.session.query(CoverTemplateModel).filter(CoverTemplateModel.id == template_id).first()
|
|
if model is None:
|
|
return None
|
|
return self._to_domain(model)
|
|
|
|
def list_for_user(self, user_id: str, skip: int = 0, limit: int = 100) -> list[CoverTemplate]:
|
|
"""列出用户可见的模板：系统模板 + 用户自己的模板。"""
|
|
models = (
|
|
self.session.query(CoverTemplateModel)
|
|
.filter(
|
|
or_(
|
|
CoverTemplateModel.is_system == True, # noqa: E712
|
|
CoverTemplateModel.user_id == user_id,
|
|
)
|
|
)
|
|
.order_by(CoverTemplateModel.is_system.desc(), CoverTemplateModel.created_at.desc())
|
|
.offset(skip)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
return [self._to_domain(m) for m in models]
|
|
|
|
def count_for_user(self, user_id: str) -> int:
|
|
return (
|
|
self.session.query(CoverTemplateModel)
|
|
.filter(
|
|
or_(
|
|
CoverTemplateModel.is_system == True, # noqa: E712
|
|
CoverTemplateModel.user_id == user_id,
|
|
)
|
|
)
|
|
.count()
|
|
)
|
|
|
|
def update(self, template: CoverTemplate) -> CoverTemplate:
|
|
model = self.session.query(CoverTemplateModel).filter(CoverTemplateModel.id == template.id).first()
|
|
if model is None:
|
|
raise ValueError(f"模板 {template.id} 不存在")
|
|
model.name = template.name
|
|
model.thumbnail_url = template.thumbnail_url
|
|
model.config = template.config
|
|
model.updated_at = template.updated_at
|
|
self.session.commit()
|
|
return template
|
|
|
|
def delete(self, template_id: str) -> bool:
|
|
model = self.session.query(CoverTemplateModel).filter(CoverTemplateModel.id == template_id).first()
|
|
if model is None:
|
|
return False
|
|
self.session.delete(model)
|
|
self.session.commit()
|
|
return True
|
|
|
|
def list_system_templates(self) -> list[CoverTemplate]:
|
|
models = (
|
|
self.session.query(CoverTemplateModel)
|
|
.filter(CoverTemplateModel.is_system == True) # noqa: E712
|
|
.order_by(CoverTemplateModel.created_at)
|
|
.all()
|
|
)
|
|
return [self._to_domain(m) for m in models]
|
|
|
|
@staticmethod
|
|
def _to_domain(model: CoverTemplateModel) -> CoverTemplate:
|
|
return CoverTemplate(
|
|
id=model.id,
|
|
user_id=model.user_id,
|
|
name=model.name,
|
|
thumbnail_url=model.thumbnail_url,
|
|
is_system=model.is_system,
|
|
config=model.config or {},
|
|
created_at=model.created_at,
|
|
updated_at=model.updated_at,
|
|
)
|