3705945a7d
架构重构（候选 1-6）： - storage/repository.py 按域拆分为包（target/scenario/run/campaign/result） - storage/db.py 按域拆分为包（eval/campaign/file/model_config/intelligent_eval） - intelligent_eval/lifecycle.py 按状态机阶段拆分为包 - services/runs.py 编排逻辑下沉 - Campaigns.tsx 拆分为 campaigns/ 子组件 测试补全（候选 7）： 前端（+125 用例，107→232）： - utils/ 纯函数：date/campaignTime/ruleLabels/fileTree/fileFormat/colors - stores/tabStore 状态管理 - 核心组件：FormDrawer/PageWrapper/ChatBubble/GeneratedMessages/SectionHeader/StatCard/TurnList - 业务组件：CaseBlock/CaseDetail/RuleOverview/WindowTimeline/RunList/TabBar/CampaignRunTimeline - 文件管理：FileCategoryTree/FileTable - hooks：sessionReducer/useFiles/useRunSession 后端（+38 用例，916→954）： - targets API CRUD + 404 路径 - WebSocket 连接管理器 - proxy 头部重写（CSP/X-Frame-Options） - target 仓储 update 方法 - app 健康检查 + SPA 404 - scenarios 模板端点 + 404 - files API 边缘分支（404 场景 + 500 兜底） - files service update_category - 智能评估状态机迁移测试 门禁状态： - 前端：tsc 干净 + 232 passed - 后端：954 passed + ruff 全绿
59 lines
1.9 KiB
Python
59 lines
1.9 KiB
Python
"""Shared CRUD skeleton for id-keyed entity repositories."""
|
|
|
|
from typing import Generic, Optional, TypeVar
|
|
|
|
from sqlmodel import Session, select
|
|
|
|
from agenteval.storage.db import get_session
|
|
|
|
M = TypeVar("M") # domain model
|
|
DB = TypeVar("DB") # persisted table row
|
|
|
|
|
|
class BaseRepository(Generic[M, DB]):
|
|
"""Shared CRUD skeleton for id-keyed entity repositories.
|
|
|
|
Subclasses declare the table (``_table``) and the ``list_all`` ordering
|
|
column name (``_order_by``, newest-first), and implement the ``_to_db`` /
|
|
``_from_db`` converter pair. The converters are instance methods so a
|
|
subclass whose ``_from_db`` needs cross-table reads (e.g. Scenario's model
|
|
bindings) can reach ``self.session``. Entities with bespoke create/update
|
|
(binding validation, versioning) override just those methods.
|
|
"""
|
|
|
|
_table: type
|
|
_order_by: str
|
|
|
|
def __init__(self, session: Optional[Session] = None):
|
|
self.session = session or get_session()
|
|
|
|
def _to_db(self, obj: M) -> DB:
|
|
raise NotImplementedError
|
|
|
|
def _from_db(self, db: DB) -> M:
|
|
raise NotImplementedError
|
|
|
|
def list_all(self) -> list[M]:
|
|
column = getattr(self._table, self._order_by)
|
|
statement = select(self._table).order_by(column.desc())
|
|
return [self._from_db(r) for r in self.session.exec(statement).all()]
|
|
|
|
def get(self, entity_id: str) -> Optional[M]:
|
|
db = self.session.get(self._table, entity_id)
|
|
return self._from_db(db) if db else None
|
|
|
|
def create(self, obj: M) -> M:
|
|
db = self._to_db(obj)
|
|
self.session.add(db)
|
|
self.session.commit()
|
|
self.session.refresh(db)
|
|
return self._from_db(db)
|
|
|
|
def delete(self, entity_id: str) -> bool:
|
|
db = self.session.get(self._table, entity_id)
|
|
if not db:
|
|
return False
|
|
self.session.delete(db)
|
|
self.session.commit()
|
|
return True
|