Simon
4fa2bb748e
- factor_composite 表（migration c3e9a0d1f4b5，MySQL 已应用；name 唯一） - CompositeDefinition/Component 实体 + CompositeRepository Protocol + SQLAlchemy 实现 - /api/composites：POST（注册表自动填充组件 direction；未注册因子 400）、GET 列表/详情、DELETE - tests/test_composites_api.py（repo CRUD/同名拒绝/删除、API 方向填充/404/400）；全量 pytest 通过
81 lines
2.9 KiB
Python
81 lines
2.9 KiB
Python
"""因子组合 Repository 的 SQLAlchemy 实现（M7.2b）。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import datetime
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.domain.entities.composite import CompositeComponent, CompositeDefinition
|
|
from app.infrastructure.persistence.sqlalchemy.models.composite import FactorCompositeModel
|
|
|
|
|
|
class SqlAlchemyCompositeRepository:
|
|
def __init__(self, session: Session) -> None:
|
|
self._session = session
|
|
|
|
def save(self, definition: CompositeDefinition) -> CompositeDefinition:
|
|
if not definition.id:
|
|
raise ValueError("需要 id（由调用方生成）")
|
|
exists = self._session.get(FactorCompositeModel, definition.id)
|
|
dup = self._session.scalar(
|
|
select(FactorCompositeModel)
|
|
.where(FactorCompositeModel.name == definition.name)
|
|
.limit(1)
|
|
)
|
|
if dup is not None and dup.id != definition.id:
|
|
raise ValueError(f"组合名已存在：{definition.name}")
|
|
now = definition.created_at or datetime.now()
|
|
if exists is None:
|
|
self._session.add(
|
|
FactorCompositeModel(
|
|
id=definition.id,
|
|
name=definition.name,
|
|
method=definition.method,
|
|
description=definition.description,
|
|
components_json=json.dumps(
|
|
[c.model_dump() for c in definition.components], ensure_ascii=False
|
|
),
|
|
created_at=now,
|
|
)
|
|
)
|
|
else:
|
|
exists.name = definition.name
|
|
exists.method = definition.method
|
|
exists.description = definition.description
|
|
exists.components_json = json.dumps(
|
|
[c.model_dump() for c in definition.components], ensure_ascii=False
|
|
)
|
|
self._session.flush()
|
|
return definition
|
|
|
|
def get(self, composite_id: str) -> CompositeDefinition | None:
|
|
row = self._session.get(FactorCompositeModel, composite_id)
|
|
return _to_entity(row) if row else None
|
|
|
|
def list(self) -> list[CompositeDefinition]:
|
|
rows = self._session.scalars(
|
|
select(FactorCompositeModel).order_by(FactorCompositeModel.name)
|
|
).all()
|
|
return [_to_entity(r) for r in rows]
|
|
|
|
def delete(self, composite_id: str) -> bool:
|
|
row = self._session.get(FactorCompositeModel, composite_id)
|
|
if row is None:
|
|
return False
|
|
self._session.delete(row)
|
|
return True
|
|
|
|
|
|
def _to_entity(row: FactorCompositeModel) -> CompositeDefinition:
|
|
return CompositeDefinition(
|
|
id=row.id,
|
|
name=row.name,
|
|
method=row.method,
|
|
description=row.description,
|
|
components=[CompositeComponent(**c) for c in json.loads(row.components_json)],
|
|
created_at=row.created_at,
|
|
)
|