feat(templates): F3b 模板库 repo + 端点(GET/POST/DELETE /templates)

- TemplateRepo/SqlTemplateRepo(list/create/delete,owner stub 过滤,只 flush 端点 commit)
- schemas/templates.py(title/body min_length=1→422)+ routers/templates.py(删不存在→404)+ main 注册
- 单测:repo CRUD/owner 隔离/frozen + 端点 201/列出/204/422×2/404
This commit is contained in:
Yaojia Wang
2026-06-23 20:17:32 +02:00
parent 9bb453acc8
commit cfe6a535b0
8 changed files with 480 additions and 0 deletions

View File

@@ -60,6 +60,12 @@ from ww_core.domain.style_repo import (
StyleFingerprintView,
StyleFingerprintWriteRepo,
)
from ww_core.domain.template_repo import (
SqlTemplateRepo,
TemplateCreate,
TemplateRepo,
TemplateView,
)
from ww_core.domain.world_entity_repo import (
SqlWorldEntityWriteRepo,
WorldEntityWriteRepo,
@@ -116,4 +122,8 @@ __all__ = [
"StyleFingerprintWriteRepo",
"StyleFingerprintView",
"SqlStyleFingerprintWriteRepo",
"TemplateCreate",
"TemplateRepo",
"TemplateView",
"SqlTemplateRepo",
]

View File

@@ -0,0 +1,105 @@
"""提示词/模板库 RepositoryF3 / 契约 §F3
单用户本地版:作者保存/复用提示词模板,可一键填入生成器的 brief/text。**不做分享/市场**
(需多租户)。统一按 `owner_id` 过滤(单用户原型 stub多租户化时由认证主体替换
提交边界:`create`/`delete` 只 `flush()` 不 `commit()`——提交交端点事务(与项目其它写侧
repo 一致,见 memory/gotchas。`delete` 返回是否删到行(端点据此映射 404
视图是与 ORM 解耦的只读 Pydantic 快照frozen——路由不碰 SQLAlchemy 行。
"""
from __future__ import annotations
import uuid
from typing import Protocol
from pydantic import BaseModel
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from ww_db.models import PromptTemplate
class TemplateView(BaseModel):
"""模板只读快照snake_casefrozen"""
model_config = {"frozen": True}
id: uuid.UUID
title: str
body: str
category: str | None = None
tool_key: str | None = None
class TemplateCreate(BaseModel):
"""新建模板写入字段owner_id 由服务层补 stub"""
title: str
body: str
category: str | None = None
tool_key: str | None = None
class TemplateRepo(Protocol):
"""模板读写接口(按 owner_id 隔离create/delete 只 flush"""
async def create(self, owner_id: uuid.UUID, data: TemplateCreate) -> TemplateView: ...
async def list_for_owner(self, owner_id: uuid.UUID) -> list[TemplateView]: ...
async def delete(self, owner_id: uuid.UUID, template_id: uuid.UUID) -> bool: ...
def _to_view(row: PromptTemplate) -> TemplateView:
return TemplateView(
id=row.id,
title=row.title,
body=row.body,
category=row.category,
tool_key=row.tool_key,
)
class SqlTemplateRepo:
"""SQLAlchemy 实现:写/读/删 `prompt_templates`,按 owner_id 过滤(只 flush 不 commit"""
def __init__(self, session: AsyncSession) -> None:
self._s = session
async def create(self, owner_id: uuid.UUID, data: TemplateCreate) -> TemplateView:
row = PromptTemplate(
owner_id=owner_id,
title=data.title,
body=data.body,
category=data.category,
tool_key=data.tool_key,
)
self._s.add(row)
await self._s.flush()
await self._s.refresh(row)
return _to_view(row)
async def list_for_owner(self, owner_id: uuid.UUID) -> list[TemplateView]:
rows = (
await self._s.execute(
select(PromptTemplate)
.where(PromptTemplate.owner_id == owner_id)
.order_by(PromptTemplate.created_at)
)
).scalars()
return [_to_view(r) for r in rows]
async def delete(self, owner_id: uuid.UUID, template_id: uuid.UUID) -> bool:
row = (
await self._s.execute(
select(PromptTemplate).where(
PromptTemplate.owner_id == owner_id,
PromptTemplate.id == template_id,
)
)
).scalar_one_or_none()
if row is None:
return False
await self._s.delete(row)
await self._s.flush()
return True