feat(phase8): 任务 2.03 — API 模板管理 CRUD #144

Closed
xiaoxia wants to merge 1 commits from feature/phase8-task203-template-api into develop
4 changed files with 706 additions and 1 deletions
+6
View File
@@ -6,6 +6,7 @@ from app.api.routes.chunked_upload import router as chunked_upload_router
from app.api.routes.classification_jobs import router as classification_jobs_router
from app.api.routes.dashboard import router as dashboard_router
from app.api.routes.duplication import router as duplication_router
from app.api.routes.edit_templates import router as edit_templates_router
from app.api.routes.generated_videos import router as generated_videos_router
from app.api.routes.generation_tasks import router as generation_tasks_router
from app.api.routes.health import router as health_check_router
@@ -116,3 +117,8 @@ api_router.include_router(
prefix="/dashboard",
tags=["Dashboard"],
)
api_router.include_router(
edit_templates_router,
prefix="/edit-templates",
tags=["EditTemplate"],
)
+267
View File
@@ -0,0 +1,267 @@
"""模板管理 API — Phase 8 模板编排引擎.
RESTful CRUD for EditTemplate:
- GET /api/v1/edit-templates 列表(分页 + 类型筛选)
- GET /api/v1/edit-templates/{id} 详情
- POST /api/v1/edit-templates 创建(管理员)
- PUT /api/v1/edit-templates/{id} 更新
- DELETE /api/v1/edit-templates/{id} 删除(软删除 → inactive
"""
from __future__ import annotations
import logging
from datetime import datetime
from typing import Any, List, Optional
from app.auth import AuthenticatedUser, get_current_user
from app.dependencies import get_db_session
from fastapi import APIRouter, Depends, HTTPException, Query, status
from pydantic import BaseModel, Field
from sqlalchemy.orm import Session
from packages.adapters.sqlalchemy_impl import SQLAlchemyEditTemplateRepository
from packages.domain.edit_template import EditTemplate, EditTemplateStatus
logger = logging.getLogger(__name__)
router = APIRouter()
# ── Pydantic Schemas ─────────────────────────────────────────────────────────
class EditTemplateCreateRequest(BaseModel):
"""创建模板请求体"""
name: str = Field(..., min_length=1, max_length=200, description="模板名称")
description: str = Field(default="", max_length=2000, description="模板描述")
template_type: str = Field(default="default", max_length=50, description="模板类型")
config: dict[str, Any] = Field(default_factory=dict, description="模板配置 (JSON)")
preview_url: str = Field(default="", max_length=500, description="预览地址")
sort_weight: int = Field(default=0, ge=0, le=9999, description="排序权重")
class EditTemplateUpdateRequest(BaseModel):
"""更新模板请求体"""
name: Optional[str] = Field(default=None, min_length=1, max_length=200, description="模板名称")
description: Optional[str] = Field(default=None, max_length=2000, description="模板描述")
template_type: Optional[str] = Field(default=None, max_length=50, description="模板类型")
config: Optional[dict[str, Any]] = Field(default=None, description="模板配置 (JSON)")
preview_url: Optional[str] = Field(default=None, max_length=500, description="预览地址")
sort_weight: Optional[int] = Field(default=None, ge=0, le=9999, description="排序权重")
status: Optional[str] = Field(default=None, description="状态: active / inactive")
class EditTemplateResponse(BaseModel):
"""模板响应体"""
id: str
name: str
description: str
template_type: str
config: dict[str, Any]
preview_url: str
sort_weight: int
status: str
created_at: datetime
updated_at: datetime
model_config = {"from_attributes": True}
class EditTemplateListResponse(BaseModel):
"""模板列表响应体"""
items: List[EditTemplateResponse]
total: int
page: int
page_size: int
# ── Helpers ───────────────────────────────────────────────────────────────────
def _to_response(t: EditTemplate) -> EditTemplateResponse:
return EditTemplateResponse(
id=t.id,
name=t.name,
description=t.description,
template_type=t.template_type,
config=t.config,
preview_url=t.preview_url,
sort_weight=t.sort_weight,
status=t.status.value if hasattr(t.status, "value") else t.status,
created_at=t.created_at,
updated_at=t.updated_at,
)
# ── Routes ────────────────────────────────────────────────────────────────────
@router.get("", response_model=EditTemplateListResponse)
def list_templates(
page: int = Query(default=1, ge=1, description="页码"),
page_size: int = Query(default=20, ge=1, le=100, description="每页数量"),
template_type: Optional[str] = Query(default=None, description="按类型筛选"),
status_filter: Optional[str] = Query(
default=None,
alias="status",
description="按状态筛选: active / inactive",
),
db: Session = Depends(get_db_session),
current_user: AuthenticatedUser = Depends(get_current_user),
) -> EditTemplateListResponse:
"""获取模板列表(支持分页、按类型/状态筛选)"""
repo = SQLAlchemyEditTemplateRepository(db)
# 解析状态筛选
status_enum: Optional[EditTemplateStatus] = None
if status_filter:
try:
status_enum = EditTemplateStatus(status_filter)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"无效的状态值: {status_filter},可选值: active, inactive",
)
skip = (page - 1) * page_size
templates = repo.list_all(
template_type=template_type,
status=status_enum,
skip=skip,
limit=page_size,
)
total = repo.count(template_type=template_type, status=status_enum)
return EditTemplateListResponse(
items=[_to_response(t) for t in templates],
total=total,
page=page,
page_size=page_size,
)
@router.get("/{template_id}", response_model=EditTemplateResponse)
def get_template(
template_id: str,
db: Session = Depends(get_db_session),
current_user: AuthenticatedUser = Depends(get_current_user),
) -> EditTemplateResponse:
"""获取单个模板详情"""
repo = SQLAlchemyEditTemplateRepository(db)
template = repo.get(template_id)
if template is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"模板不存在: {template_id}",
)
return _to_response(template)
@router.post("", response_model=EditTemplateResponse, status_code=status.HTTP_201_CREATED)
def create_template(
body: EditTemplateCreateRequest,
db: Session = Depends(get_db_session),
current_user: AuthenticatedUser = Depends(get_current_user),
) -> EditTemplateResponse:
"""创建模板(管理员)"""
repo = SQLAlchemyEditTemplateRepository(db)
try:
template = EditTemplate.create(
name=body.name,
description=body.description,
template_type=body.template_type,
config=body.config,
preview_url=body.preview_url,
sort_weight=body.sort_weight,
)
except ValueError as exc:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=str(exc),
)
created = repo.create(template)
logger.info("创建模板: id=%s name=%s by user=%s", created.id, created.name, current_user.user.id)
return _to_response(created)
@router.put("/{template_id}", response_model=EditTemplateResponse)
def update_template(
template_id: str,
body: EditTemplateUpdateRequest,
db: Session = Depends(get_db_session),
current_user: AuthenticatedUser = Depends(get_current_user),
) -> EditTemplateResponse:
"""更新模板"""
repo = SQLAlchemyEditTemplateRepository(db)
existing = repo.get(template_id)
if existing is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"模板不存在: {template_id}",
)
# 部分更新:仅覆盖 body 中提供的字段
new_name = body.name if body.name is not None else existing.name
new_description = body.description if body.description is not None else existing.description
new_template_type = body.template_type if body.template_type is not None else existing.template_type
new_config = body.config if body.config is not None else existing.config
new_preview_url = body.preview_url if body.preview_url is not None else existing.preview_url
new_sort_weight = body.sort_weight if body.sort_weight is not None else existing.sort_weight
new_status = existing.status
if body.status is not None:
try:
new_status = EditTemplateStatus(body.status)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"无效的状态值: {body.status},可选值: active, inactive",
)
try:
updated = EditTemplate(
id=existing.id,
name=new_name.strip() if new_name else existing.name,
description=new_description.strip() if new_description is not None else existing.description,
template_type=new_template_type.strip() if new_template_type else existing.template_type,
config=new_config,
preview_url=new_preview_url.strip() if new_preview_url is not None else existing.preview_url,
sort_weight=new_sort_weight,
status=new_status,
created_at=existing.created_at,
updated_at=existing.updated_at,
)
except (ValueError, TypeError) as exc:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=str(exc),
)
result = repo.update(updated)
logger.info("更新模板: id=%s by user=%s", template_id, current_user.user.id)
return _to_response(result)
@router.delete("/{template_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_template(
template_id: str,
db: Session = Depends(get_db_session),
current_user: AuthenticatedUser = Depends(get_current_user),
) -> None:
"""删除模板(软删除 → 设为 inactive)"""
repo = SQLAlchemyEditTemplateRepository(db)
existing = repo.get(template_id)
if existing is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"模板不存在: {template_id}",
)
# 软删除:将状态设为 inactive
existing.deactivate()
repo.update(existing)
logger.info("删除模板(软删除): id=%s by user=%s", template_id, current_user.user.id)
@@ -119,9 +119,16 @@ class SQLAlchemyEditTemplateRepository:
self.session.commit()
return True
def count(self, *, status: Optional[EditTemplateStatus] = None) -> int:
def count(
self,
*,
template_type: Optional[str] = None,
status: Optional[EditTemplateStatus] = None,
) -> int:
"""统计模板数量"""
query = self.session.query(EditTemplateModel)
if template_type:
query = query.filter(EditTemplateModel.template_type == template_type)
if status:
query = query.filter(EditTemplateModel.status == status)
return query.count()
+425
View File
@@ -0,0 +1,425 @@
"""模板管理 API 单元测试 — Phase 8 任务 2.03.
覆盖 5 个端点:
GET /api/v1/edit-templates — 列表(分页 + 筛选)
GET /api/v1/edit-templates/{id} — 详情
POST /api/v1/edit-templates — 创建
PUT /api/v1/edit-templates/{id} — 更新
DELETE /api/v1/edit-templates/{id} — 软删除
使用 FastAPI TestClient + Stub Repository + dependency_overrides.
"""
from __future__ import annotations
import os
import sys
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Optional
from unittest.mock import MagicMock
os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "apps", "api"))
from packages.domain.edit_template import EditTemplate, EditTemplateStatus
# ── Stub Repository ───────────────────────────────────────────────────────────
class StubEditTemplateRepository:
"""内存中模拟 EditTemplate 仓储"""
def __init__(self) -> None:
self._store: dict[str, EditTemplate] = {}
def list_all(
self,
*,
template_type: Optional[str] = None,
status: Optional[EditTemplateStatus] = None,
skip: int = 0,
limit: int = 50,
) -> list[EditTemplate]:
items = list(self._store.values())
if template_type:
items = [t for t in items if t.template_type == template_type]
if status:
items = [t for t in items if t.status == status]
items.sort(key=lambda t: t.created_at, reverse=True)
return items[skip : skip + limit]
def list_active(
self,
*,
template_type: Optional[str] = None,
skip: int = 0,
limit: int = 50,
) -> list[EditTemplate]:
return self.list_all(
template_type=template_type, status=EditTemplateStatus.ACTIVE, skip=skip, limit=limit
)
def get(self, template_id: str) -> Optional[EditTemplate]:
return self._store.get(template_id)
def create(self, template: EditTemplate) -> EditTemplate:
self._store[template.id] = template
return template
def update(self, template: EditTemplate) -> EditTemplate:
if template.id not in self._store:
raise ValueError(f"EditTemplate {template.id} not found")
self._store[template.id] = template
return template
def delete(self, template_id: str) -> bool:
if template_id in self._store:
del self._store[template_id]
return True
return False
def count(
self,
*,
template_type: Optional[str] = None,
status: Optional[EditTemplateStatus] = None,
) -> int:
items = list(self._store.values())
if template_type:
items = [t for t in items if t.template_type == template_type]
if status:
items = [t for t in items if t.status == status]
return len(items)
# ── Fixtures ──────────────────────────────────────────────────────────────────
@dataclass
class FakeUser:
id: str = "user-001"
email: str = "test@example.com"
@dataclass
class FakeAuthenticatedUser:
user: FakeUser = field(default_factory=FakeUser)
session_id: str | None = None
token_type: str | None = None
@pytest.fixture
def stub_repo() -> StubEditTemplateRepository:
return StubEditTemplateRepository()
@pytest.fixture
def app(stub_repo: StubEditTemplateRepository) -> FastAPI:
"""构建测试 FastAPI 应用,注入 Stub Repository"""
from app.api.routes.edit_templates import router
from app.auth import get_current_user
from app.dependencies import get_db_session
import app.api.routes.edit_templates as route_module
# 替换路由模块中的 Repository 类
original_repo_cls = route_module.SQLAlchemyEditTemplateRepository
route_module.SQLAlchemyEditTemplateRepository = lambda session: stub_repo
test_app = FastAPI()
test_app.include_router(router, prefix="/api/v1/edit-templates")
# 覆盖依赖
def override_get_db_session():
yield MagicMock()
def override_get_current_user():
return FakeAuthenticatedUser()
test_app.dependency_overrides[get_db_session] = override_get_db_session
test_app.dependency_overrides[get_current_user] = override_get_current_user
yield test_app
# 恢复
route_module.SQLAlchemyEditTemplateRepository = original_repo_cls
test_app.dependency_overrides.clear()
@pytest.fixture
def client(app: FastAPI) -> TestClient:
return TestClient(app)
def _make_template(name: str = "测试模板", **kwargs: Any) -> EditTemplate:
return EditTemplate.create(name=name, **kwargs)
# ── GET /api/v1/edit-templates (列表) ─────────────────────────────────────────
class TestListTemplates:
def test_empty_list(self, client: TestClient) -> None:
resp = client.get("/api/v1/edit-templates")
assert resp.status_code == 200
data = resp.json()
assert data["items"] == []
assert data["total"] == 0
assert data["page"] == 1
def test_list_with_items(self, client: TestClient, stub_repo: StubEditTemplateRepository) -> None:
for i in range(3):
stub_repo.create(_make_template(f"模板{i}"))
resp = client.get("/api/v1/edit-templates")
assert resp.status_code == 200
data = resp.json()
assert data["total"] == 3
assert len(data["items"]) == 3
def test_pagination(self, client: TestClient, stub_repo: StubEditTemplateRepository) -> None:
for i in range(5):
stub_repo.create(_make_template(f"模板{i}"))
resp = client.get("/api/v1/edit-templates?page=1&page_size=2")
assert resp.status_code == 200
data = resp.json()
assert len(data["items"]) == 2
assert data["total"] == 5
assert data["page"] == 1
assert data["page_size"] == 2
def test_filter_by_type(self, client: TestClient, stub_repo: StubEditTemplateRepository) -> None:
stub_repo.create(_make_template("Vlog模板", template_type="vlog"))
stub_repo.create(_make_template("短视频模板", template_type="short"))
stub_repo.create(_make_template("另一个Vlog", template_type="vlog"))
resp = client.get("/api/v1/edit-templates?template_type=vlog")
assert resp.status_code == 200
data = resp.json()
assert data["total"] == 2
assert all(item["template_type"] == "vlog" for item in data["items"])
def test_filter_by_status(self, client: TestClient, stub_repo: StubEditTemplateRepository) -> None:
t1 = _make_template("活跃模板")
stub_repo.create(t1)
t2 = _make_template("停用模板", status=EditTemplateStatus.INACTIVE)
stub_repo.create(t2)
resp = client.get("/api/v1/edit-templates?status=active")
assert resp.status_code == 200
data = resp.json()
assert data["total"] == 1
assert data["items"][0]["name"] == "活跃模板"
def test_invalid_status_filter(self, client: TestClient) -> None:
resp = client.get("/api/v1/edit-templates?status=invalid")
assert resp.status_code == 400
def test_invalid_page(self, client: TestClient) -> None:
resp = client.get("/api/v1/edit-templates?page=0")
assert resp.status_code == 422
# ── GET /api/v1/edit-templates/{id} (详情) ────────────────────────────────────
class TestGetTemplate:
def test_get_existing(self, client: TestClient, stub_repo: StubEditTemplateRepository) -> None:
t = _make_template("详情模板", description="这是描述", template_type="vlog")
stub_repo.create(t)
resp = client.get(f"/api/v1/edit-templates/{t.id}")
assert resp.status_code == 200
data = resp.json()
assert data["id"] == t.id
assert data["name"] == "详情模板"
assert data["description"] == "这是描述"
assert data["template_type"] == "vlog"
assert data["status"] == "active"
def test_get_not_found(self, client: TestClient) -> None:
resp = client.get("/api/v1/edit-templates/nonexistent-id")
assert resp.status_code == 404
assert "不存在" in resp.json()["detail"]
# ── POST /api/v1/edit-templates (创建) ────────────────────────────────────────
class TestCreateTemplate:
def test_create_basic(self, client: TestClient) -> None:
resp = client.post("/api/v1/edit-templates", json={"name": "新模板"})
assert resp.status_code == 201
data = resp.json()
assert data["name"] == "新模板"
assert data["description"] == ""
assert data["template_type"] == "default"
assert data["status"] == "active"
assert data["sort_weight"] == 0
assert "id" in data
def test_create_with_all_fields(self, client: TestClient) -> None:
body = {
"name": "完整模板",
"description": "完整描述",
"template_type": "vlog",
"config": {"key": "value"},
"preview_url": "https://example.com/preview.mp4",
"sort_weight": 10,
}
resp = client.post("/api/v1/edit-templates", json=body)
assert resp.status_code == 201
data = resp.json()
assert data["name"] == "完整模板"
assert data["description"] == "完整描述"
assert data["template_type"] == "vlog"
assert data["config"] == {"key": "value"}
assert data["preview_url"] == "https://example.com/preview.mp4"
assert data["sort_weight"] == 10
def test_create_empty_name(self, client: TestClient) -> None:
resp = client.post("/api/v1/edit-templates", json={"name": ""})
assert resp.status_code == 422 # Pydantic min_length=1
def test_create_whitespace_name(self, client: TestClient) -> None:
resp = client.post("/api/v1/edit-templates", json={"name": " "})
assert resp.status_code == 400 # domain validation
def test_create_missing_name(self, client: TestClient) -> None:
resp = client.post("/api/v1/edit-templates", json={})
assert resp.status_code == 422
def test_create_negative_sort_weight(self, client: TestClient) -> None:
resp = client.post("/api/v1/edit-templates", json={"name": "模板", "sort_weight": -1})
assert resp.status_code == 422
# ── PUT /api/v1/edit-templates/{id} (更新) ────────────────────────────────────
class TestUpdateTemplate:
def test_update_name(self, client: TestClient, stub_repo: StubEditTemplateRepository) -> None:
t = _make_template("旧名称")
stub_repo.create(t)
resp = client.put(f"/api/v1/edit-templates/{t.id}", json={"name": "新名称"})
assert resp.status_code == 200
assert resp.json()["name"] == "新名称"
def test_update_multiple_fields(self, client: TestClient, stub_repo: StubEditTemplateRepository) -> None:
t = _make_template("模板")
stub_repo.create(t)
body = {"name": "更新后", "description": "新描述", "sort_weight": 5}
resp = client.put(f"/api/v1/edit-templates/{t.id}", json=body)
assert resp.status_code == 200
data = resp.json()
assert data["name"] == "更新后"
assert data["description"] == "新描述"
assert data["sort_weight"] == 5
def test_update_status(self, client: TestClient, stub_repo: StubEditTemplateRepository) -> None:
t = _make_template("模板")
stub_repo.create(t)
resp = client.put(f"/api/v1/edit-templates/{t.id}", json={"status": "inactive"})
assert resp.status_code == 200
assert resp.json()["status"] == "inactive"
def test_update_invalid_status(self, client: TestClient, stub_repo: StubEditTemplateRepository) -> None:
t = _make_template("模板")
stub_repo.create(t)
resp = client.put(f"/api/v1/edit-templates/{t.id}", json={"status": "bogus"})
assert resp.status_code == 400
def test_update_not_found(self, client: TestClient) -> None:
resp = client.put("/api/v1/edit-templates/nonexistent", json={"name": "x"})
assert resp.status_code == 404
def test_partial_update_preserves_others(
self, client: TestClient, stub_repo: StubEditTemplateRepository
) -> None:
t = _make_template("原名", description="原描述", template_type="vlog")
stub_repo.create(t)
resp = client.put(f"/api/v1/edit-templates/{t.id}", json={"name": "新名"})
assert resp.status_code == 200
data = resp.json()
assert data["name"] == "新名"
assert data["description"] == "原描述"
assert data["template_type"] == "vlog"
def test_update_empty_body(self, client: TestClient, stub_repo: StubEditTemplateRepository) -> None:
t = _make_template("模板")
stub_repo.create(t)
resp = client.put(f"/api/v1/edit-templates/{t.id}", json={})
assert resp.status_code == 200
assert resp.json()["name"] == "模板"
# ── DELETE /api/v1/edit-templates/{id} (软删除) ───────────────────────────────
class TestDeleteTemplate:
def test_soft_delete(self, client: TestClient, stub_repo: StubEditTemplateRepository) -> None:
t = _make_template("待删除")
stub_repo.create(t)
resp = client.delete(f"/api/v1/edit-templates/{t.id}")
assert resp.status_code == 204
# 软删除后仍存在,但状态为 inactive
updated = stub_repo.get(t.id)
assert updated is not None
assert updated.status == EditTemplateStatus.INACTIVE
def test_soft_delete_not_found(self, client: TestClient) -> None:
resp = client.delete("/api/v1/edit-templates/nonexistent")
assert resp.status_code == 404
def test_soft_delete_idempotent(
self, client: TestClient, stub_repo: StubEditTemplateRepository
) -> None:
t = _make_template("模板")
stub_repo.create(t)
# 第一次删除
resp1 = client.delete(f"/api/v1/edit-templates/{t.id}")
assert resp1.status_code == 204
# 第二次删除(已经是 inactive,但仍可再次设为 inactive
resp2 = client.delete(f"/api/v1/edit-templates/{t.id}")
assert resp2.status_code == 204
def test_deleted_not_in_active_list(
self, client: TestClient, stub_repo: StubEditTemplateRepository
) -> None:
t = _make_template("模板")
stub_repo.create(t)
client.delete(f"/api/v1/edit-templates/{t.id}")
resp = client.get("/api/v1/edit-templates?status=active")
data = resp.json()
assert data["total"] == 0
# ── Response Schema 验证 ──────────────────────────────────────────────────────
class TestResponseSchema:
def test_response_has_all_fields(
self, client: TestClient, stub_repo: StubEditTemplateRepository
) -> None:
t = _make_template("模板", description="描述", template_type="vlog")
stub_repo.create(t)
resp = client.get(f"/api/v1/edit-templates/{t.id}")
data = resp.json()
expected_keys = {
"id", "name", "description", "template_type",
"config", "preview_url", "sort_weight", "status",
"created_at", "updated_at",
}
assert set(data.keys()) == expected_keys
def test_list_response_structure(self, client: TestClient) -> None:
resp = client.get("/api/v1/edit-templates")
data = resp.json()
assert "items" in data
assert "total" in data
assert "page" in data
assert "page_size" in data
assert isinstance(data["items"], list)