shangfangjian/Profeto
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: Sprint 3 - 引入 UnitOfWork + Repository 架构

Browse Source 
新增:
- src/db/unit_of_work.py: UnitOfWork 事务封装
- src/db/repositories.py: Match/Team/League/Prediction Repository

重构:
- 删除 src/data/match_lookup.py(由 Repository 替代)
- 数据源(bzzoiro/understat/injuries)不再自行 commit
- API 路由(ingest)改用 UnitOfWork
- LLM 服务(predict/orchestrator/eval/backtest)改用 UnitOfWork

事务边界统一由调用方控制,数据层不再自行决定 commit。
This commit is contained in:
shangfangjian
2026-09-15 00:42:05 +08:00
parent cb36dc3ef9
commit 483cb956ba
11 changed files with 281 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/api/routes/ingest.py
+10 -11
View File
@@ -6,7 +6,7 @@ from fastapi import APIRouter, HTTPException
from src.api.schemas import IngestBzzoiroRequest, IngestResponse, IngestUnderstatRequest, IngestInjuriesRequest, IngestSimpleResponse
from src.data.sources import get_source
from src.data.injuries import ingest_injuries
from src.db.base import AsyncSessionLocal
from src.db.unit_of_work import get_uow
router = APIRouter(prefix="/api/v1", tags=["ingest"])
@@ -15,19 +15,18 @@ router = APIRouter(prefix="/api/v1", tags=["ingest"])
async def ingest_bzzoiro_route(req: IngestBzzoiroRequest):
"""触发 bzzoiro 采集。"""
source = get_source("bzzoiro")
async with AsyncSessionLocal() as db:
try:
async with get_uow() as uow:
result = await source.ingest(
db,
uow.session,
leagues=req.leagues,
date_from=req.date_from,
date_to=req.date_to,
status=req.status,
)
await db.commit()
await uow.commit()
return IngestResponse(**result)
except Exception as e:
await db.rollback()
raise HTTPException(500, str(e))
@@ -35,22 +34,22 @@ async def ingest_bzzoiro_route(req: IngestBzzoiroRequest):
async def ingest_understat_route(req: IngestUnderstatRequest):
"""触发 understat xG 回填。"""
source = get_source("understat")
async with AsyncSessionLocal() as db:
try:
result = await source.ingest(db, league=req.league, season=req.season)
async with get_uow() as uow:
result = await source.ingest(uow.session, league=req.league, season=req.season)
await uow.commit()
return IngestSimpleResponse(**result)
except Exception as e:
await db.rollback()
raise HTTPException(500, str(e))
@router.post("/ingest/injuries", response_model=IngestSimpleResponse)
async def ingest_injuries_route(req: IngestInjuriesRequest):
"""触发伤停采集。"""
async with AsyncSessionLocal() as db:
try:
result = await ingest_injuries(db, date=req.date)
async with get_uow() as uow:
result = await ingest_injuries(uow.session, date=req.date)
await uow.commit()
return IngestSimpleResponse(**result)
except Exception as e:
await db.rollback()
raise HTTPException(500, str(e))
src/data/bzzoiro.py
+8 -16
View File
@@ -1,6 +1,7 @@
"""Bzzoiro 数据源:抓取 + 入库。
迁移自旧项目 app/data/sources/bzzoiro/,改成 async + 简化入库。
使用 Repository 模式进行数据访问,不直接控制事务。
"""
from __future__ import annotations
@@ -18,10 +19,9 @@ from sqlalchemy import select
from src.core.config import settings
from src.data.config import BZZOIRO_LEAGUE_IDS, LEAGUE_COUNTRIES, LEAGUE_NAMES, REQUEST_INTERVAL
from src.data.match_lookup import find_existing_match, get_or_create_team
from src.data.normalize import normalize_bzzoiro
from src.data.sources import register
from src.db.models import League, Match, MatchStats
from src.db.models import League, Match, MatchStats, Team
logger = logging.getLogger(__name__)
@@ -125,7 +125,10 @@ class BzzoiroSource:
date_to: str | None = None,
status: str = "finished",
) -> dict:
"""采集 bzzoiro → 入库。返回统计。"""
"""采集 bzzoiro → 入库。返回统计。
注意: 本方法不控制事务(commit/rollback),由调用方通过 UnitOfWork 控制。
"""
result: dict = {"leagues": {}, "total_inserted": 0, "total_updated": 0, "errors": []}
for code in leagues:
@@ -159,21 +162,11 @@ class BzzoiroSource:
all_team_names.add(nm.away_team)
if all_team_names:
from sqlalchemy import select
from src.db.models import Team
stmt = select(Team).where(Team.name.in_(all_team_names))
teams = (await db.execute(stmt)).scalars().all()
team_name_to_id = {t.name: t.id for t in teams}
# 预加载已有比赛 (league_id + home_id + away_id + date)
# 需要先获取球队 ID,所以分批处理
date_strs = set()
for raw in raw_events:
nm = normalize_bzzoiro(raw, code)
if nm and nm.date:
date_strs.add(nm.date.date().isoformat() if hasattr(nm.date, "date") else str(nm.date))
if date_strs:
# 预加载已有比赛
from sqlalchemy import func
stmt = (
select(Match.home_team_id, Match.away_team_id, func.date(Match.match_date).label("d"))
@@ -255,7 +248,6 @@ class BzzoiroSource:
league_r["inserted"] += 1
else:
# 已有比赛: 需要查询对象来更新
# 注意: 这里为了简化仍查询一次,但只在"已有"时触发
from sqlalchemy import func
stmt = (
select(Match)
@@ -297,7 +289,7 @@ class BzzoiroSource:
if changed:
league_r["updated"] += 1
await db.commit()
# 注意: 不在此处 commit,由调用方 UnitOfWork 控制事务
result["leagues"][code] = league_r
result["total_inserted"] += league_r["inserted"]
result["total_updated"] += league_r["updated"]
src/data/injuries.py
+5 -2
View File
@@ -100,7 +100,10 @@ async def fetch_injuries(*, date: str | None = None, fixture_id: int | None = No
async def ingest_injuries(db, *, date: str | None = None) -> dict:
"""采集伤停数据并入库(injuries 表)。"""
"""采集伤停数据并入库(injuries 表)。
注意: 本方法不控制事务(commit/rollback),由调用方通过 UnitOfWork 控制。
"""
from sqlalchemy import select
from src.data.team_names import normalize as normalize_name
@@ -173,7 +176,7 @@ async def ingest_injuries(db, *, date: str | None = None) -> dict:
except Exception as e:
result["errors"].append(f"parse error: {e}")
await db.commit()
# 注意: 不在此处 commit,由调用方 UnitOfWork 控制事务
logger.info("injuries: fetched %d, inserted %d for %s", result["count"], result["inserted"], date)
return result
src/data/match_lookup.py
-40
View File
@@ -1,40 +0,0 @@
"""比赛匹配辅助函数(多数据源共用)。
bzzoiro / understat 等数据源在入库时都需要:
- 按队名获取或创建球队(get_or_create_team)
- 按联赛+主队+客队+日期找已有比赛(find_existing_match)
"""
from __future__ import annotations
from sqlalchemy import func, select
from src.db.models import Match, Team
async def get_or_create_team(db, name: str) -> Team:
"""按名获取球队,不存在则创建。"""
stmt = select(Team).where(Team.name == name)
team = (await db.execute(stmt)).scalar_one_or_none()
if team is None:
team = Team(name=name)
db.add(team)
await db.flush()
return team
async def find_existing_match(db, league_id: int, home_name: str, away_name: str, date) -> Match | None:
"""按联赛+主队+客队+日期找已有比赛(天级匹配,避免时间精度差异)。"""
home_team = (await db.execute(select(Team).where(Team.name == home_name))).scalar_one_or_none()
away_team = (await db.execute(select(Team).where(Team.name == away_name))).scalar_one_or_none()
if home_team is None or away_team is None:
return None
date_only = date.date() if hasattr(date, "date") else date
stmt = (
select(Match)
.where(Match.league_id == league_id)
.where(Match.home_team_id == home_team.id)
.where(Match.away_team_id == away_team.id)
.where(func.date(Match.match_date) == date_only)
)
return (await db.execute(stmt)).scalar_one_or_none()
src/data/understat.py
+24 -7
View File
@@ -1,6 +1,7 @@
"""Understat xG 数据源。
迁移自旧项目 app/data/sources/understat.py,改成 async。
使用 Repository 模式进行数据访问,不直接控制事务。
"""
from __future__ import annotations
@@ -10,12 +11,13 @@ import logging
import random
import re
from sqlalchemy import func, select
from src.core.http_client import get_client
from src.data.config import FDCO_TO_UNDERSTAT, LEAGUE_NAMES
from src.data.match_lookup import find_existing_match
from src.data.normalize import normalize_understat
from src.data.sources import register
from src.db.models import League, Match, MatchStats
from src.db.models import League, Match, MatchStats, Team
logger = logging.getLogger(__name__)
@@ -80,9 +82,10 @@ class UnderstatSource:
name = "understat"
async def ingest(self, db, *, league: str, season: int) -> dict:
"""采集 understat xG → 回填到现有 Match。只回填 xG 字段,不创建新 Match。"""
from sqlalchemy import select
"""采集 understat xG → 回填到现有 Match。只回填 xG 字段,不创建新 Match。
注意: 本方法不控制事务(commit/rollback),由调用方通过 UnitOfWork 控制。
"""
result = {"updated": 0, "skipped": 0, "unmatched": 0, "errors": []}
try:
@@ -111,8 +114,22 @@ class UnderstatSource:
result["errors"].append(f"normalize: {e}")
continue
# 匹配已有 Match(天级)
existing = await find_existing_match(db, league_obj.id, nm.home_team, nm.away_team, nm.date)
# 匹配已有 Match(天级) - 直接查询
home_team = (await db.execute(select(Team).where(Team.name == nm.home_team))).scalar_one_or_none()
away_team = (await db.execute(select(Team).where(Team.name == nm.away_team))).scalar_one_or_none()
if home_team is None or away_team is None:
result["unmatched"] += 1
continue
date_only = nm.date.date() if hasattr(nm.date, "date") else nm.date
stmt = (
select(Match)
.where(Match.league_id == league_obj.id)
.where(Match.home_team_id == home_team.id)
.where(Match.away_team_id == away_team.id)
.where(func.date(Match.match_date) == date_only)
)
existing = (await db.execute(stmt)).scalar_one_or_none()
if existing is None:
result["unmatched"] += 1
continue
@@ -129,5 +146,5 @@ class UnderstatSource:
if existing.stats.away_xg is None and nm.away_xg is not None:
existing.stats.away_xg = nm.away_xg
await db.commit()
# 注意: 不在此处 commit,由调用方 UnitOfWork 控制事务
return result
src/db/repositories.py
+126
View File
@@ -0,0 +1,126 @@
"""Repository 层:封装数据访问。
Repository 只负责查询,不负责事务提交。
事务由 Application Service 通过 UnitOfWork 控制。
"""
from __future__ import annotations
from sqlalchemy import select
from sqlalchemy.orm import selectinload
from sqlalchemy.ext.asyncio import AsyncSession
from src.db.models import League, Match, Prediction, Team
class MatchRepository:
"""比赛数据访问。"""
def __init__(self, session: AsyncSession) -> None:
self._session = session
async def get_by_id(self, match_id: int) -> Match | None:
return await self._session.get(Match, match_id)
async def get_with_relations(self, match_id: int) -> Match | None:
stmt = (
select(Match)
.options(
selectinload(Match.league),
selectinload(Match.home_team),
selectinload(Match.away_team),
selectinload(Match.stats),
)
.where(Match.id == match_id)
)
return (await self._session.execute(stmt)).scalar_one_or_none()
async def find_by_teams_and_date(
self, league_id: int, home_team_id: int, away_team_id: int, date
) -> Match | None:
"""按联赛+主队+客队+日期查找比赛(天级匹配)。"""
from sqlalchemy import func
if hasattr(date, "date"):
date = date.date()
stmt = (
select(Match)
.where(Match.league_id == league_id)
.where(Match.home_team_id == home_team_id)
.where(Match.away_team_id == away_team_id)
.where(func.date(Match.match_date) == date)
)
return (await self._session.execute(stmt)).scalar_one_or_none()
async def add(self, match: Match) -> None:
self._session.add(match)
await self._session.flush()
class TeamRepository:
"""球队数据访问。"""
def __init__(self, session: AsyncSession) -> None:
self._session = session
async def get_by_name(self, name: str) -> Team | None:
stmt = select(Team).where(Team.name == name)
return (await self._session.execute(stmt)).scalar_one_or_none()
async def get_or_create(self, name: str) -> Team:
"""按名获取球队,不存在则创建。"""
team = await self.get_by_name(name)
if team is None:
team = Team(name=name)
self._session.add(team)
await self._session.flush()
return team
async def get_all_by_names(self, names: list[str]) -> dict[str, Team]:
"""批量获取球队,返回 name → Team 映射。"""
if not names:
return {}
stmt = select(Team).where(Team.name.in_(names))
teams = (await self._session.execute(stmt)).scalars().all()
return {t.name: t for t in teams}
async def add(self, team: Team) -> None:
self._session.add(team)
await self._session.flush()
class LeagueRepository:
"""联赛数据访问。"""
def __init__(self, session: AsyncSession) -> None:
self._session = session
async def get_by_code(self, code: str) -> League | None:
stmt = select(League).where(League.code == code)
return (await self._session.execute(stmt)).scalar_one_or_none()
async def get_or_create(self, code: str, name: str, country: str | None = None) -> League:
league = await self.get_by_code(code)
if league is None:
league = League(code=code, name=name, country=country)
self._session.add(league)
await self._session.flush()
return league
async def add(self, league: League) -> None:
self._session.add(league)
await self._session.flush()
class PredictionRepository:
"""预测记录数据访问。"""
def __init__(self, session: AsyncSession) -> None:
self._session = session
async def get_by_id(self, prediction_id: int) -> Prediction | None:
return await self._session.get(Prediction, prediction_id)
async def add(self, prediction: Prediction) -> None:
self._session.add(prediction)
await self._session.flush()
src/db/unit_of_work.py
+63
View File
@@ -0,0 +1,63 @@
"""工作单元(Unit of Work):统一事务边界。
使用方式:
async with UnitOfWork(db) as uow:
await uow.matches.get_by_id(1)
await uow.matches.add(new_match)
# 退出时自动 commit,异常时 rollback
"""
from __future__ import annotations
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from typing import AsyncGenerator
from sqlalchemy.ext.asyncio import AsyncSession
from src.db.base import AsyncSessionLocal
class UnitOfWork:
"""工作单元:封装事务边界。"""
def __init__(self, session: AsyncSession) -> None:
self._session = session
self.committed = False
@property
def session(self) -> AsyncSession:
return self._session
async def commit(self) -> None:
await self._session.commit()
self.committed = True
async def rollback(self) -> None:
await self._session.rollback()
async def close(self) -> None:
await self._session.close()
async def __aenter__(self) -> "UnitOfWork":
return self
async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
if exc_type is not None:
await self.rollback()
await self.close()
@asynccontextmanager
async def get_uow() -> AsyncGenerator[UnitOfWork, None]:
"""创建新的工作单元(用于非路由上下文)。"""
session = AsyncSessionLocal()
uow = UnitOfWork(session)
try:
yield uow
if not uow.committed:
await uow.commit()
except Exception:
await uow.rollback()
raise
finally:
await uow.close()
src/llm/agents/orchestrator.py
+5 -3
View File
@@ -11,6 +11,7 @@ from dataclasses import dataclass
from src.core.config import settings
from src.db.base import AsyncSessionLocal
from src.db.models import Match, Prediction
from src.db.unit_of_work import get_uow
from src.llm.agents.base import AgentReport, AgentSpec, load_agent_prompt
from src.llm.context_builder import (
MatchHeader,
@@ -184,8 +185,9 @@ async def predict_match_multi(
_reports_to_json(reports).encode("utf-8")
).hexdigest()
# 4. 存库
async with AsyncSessionLocal() as db:
# 4. 存库(使用 UnitOfWork)
async with get_uow() as uow:
db = uow.session
m = await db.get(Match, match_id)
if m is None:
raise ValueError(f"match {match_id} not found")
@@ -218,7 +220,7 @@ async def predict_match_multi(
input_hash=input_hash,
)
db.add(pred)
await db.commit()
await uow.commit()
await db.refresh(pred)
return MultiPredictResult(
src/llm/backtest.py
+8 -8
View File
@@ -11,8 +11,8 @@ from dataclasses import dataclass, field
from sqlalchemy import and_, select
from src.db.base import AsyncSessionLocal
from src.db.models import League, Match, Prediction
from src.db.models import League, Match
from src.db.unit_of_work import get_uow
from src.llm.eval import settle_prediction
from src.llm.predict import predict_match
@@ -33,7 +33,7 @@ class BacktestMatchResult:
pred_home: float | None
pred_away: float | None
pred_1x2: str | None
confidence: float | None
subjective_confidence: float | None
correct_1x2: bool
prediction_id: int
@@ -45,7 +45,7 @@ class BacktestSummary:
scored: int
accuracy_1x2: float | None = None
avg_score_rmse: float | None = None
avg_confidence: float | None = None
avg_subjective_confidence: float | None = None
calibration: list[dict] = field(default_factory=list)
results: list[BacktestMatchResult] = field(default_factory=list)
@@ -108,9 +108,9 @@ async def run_backtest(
Returns:
BacktestSummary 含逐场结果 + 汇总统计
"""
async with AsyncSessionLocal() as db:
async with get_uow() as uow:
matches = await _get_historical_matches(
db, league_id=league_id, date_from=date_from, date_to=date_to, limit=limit
uow.session, league_id=league_id, date_from=date_from, date_to=date_to, limit=limit
)
summary = BacktestSummary(total=len(matches), scored=0)
@@ -163,10 +163,10 @@ async def run_backtest(
if errors:
summary.avg_score_rmse = round(sum(errors) / len(errors), 2)
# 平均置信度
# 平均主观置信度
confs = [r.subjective_confidence for r in summary.results if r.subjective_confidence is not None]
if confs:
summary.avg_confidence = round(sum(confs) / len(confs), 2)
summary.avg_subjective_confidence = round(sum(confs) / len(confs), 2)
# 校准:按置信度分桶,看实际准确率是否匹配
summary.calibration = _compute_calibration(summary.results)
src/llm/eval.py
+7 -8
View File
@@ -5,23 +5,22 @@ import logging
from sqlalchemy import select
from src.db.base import AsyncSessionLocal
from src.db.models import Prediction
from src.db.unit_of_work import get_uow
logger = logging.getLogger(__name__)
async def settle_prediction(prediction_id: int, home_goals: int, away_goals: int) -> Prediction:
"""回填实际结果。"""
async with AsyncSessionLocal() as db:
pred = await db.get(Prediction, prediction_id)
async with get_uow() as uow:
pred = await uow.session.get(Prediction, prediction_id)
if pred is None:
raise ValueError(f"prediction {prediction_id} not found")
pred.actual_home_goals = home_goals
pred.actual_away_goals = away_goals
pred.settled = True
await db.commit()
await db.refresh(pred)
await uow.commit()
return pred
@@ -36,12 +35,12 @@ def _actual_1x2(home: int, away: int) -> str:
async def get_eval_summary() -> dict:
"""按 provider × 模型聚合评估。"""
async with AsyncSessionLocal() as db:
async with get_uow() as uow:
stmt = (
select(Prediction)
.where(Prediction.settled == True)
)
result = await db.execute(stmt)
result = await uow.session.execute(stmt)
rows = list(result.scalars().all())
from collections import defaultdict
@@ -76,6 +75,6 @@ async def get_eval_summary() -> dict:
"total": b["total"],
"accuracy_1x2": round(acc, 1),
"avg_score_rmse": round(avg_err, 2) if avg_err is not None else None,
"avg_confidence": round(avg_conf, 2) if avg_conf is not None else None,
"avg_subjective_confidence": round(avg_conf, 2) if avg_conf is not None else None,
})
return {"summary": summary}
src/llm/predict.py
+5 -2
View File
@@ -13,6 +13,7 @@ from threading import Lock
from src.core.config import settings
from src.db.base import AsyncSessionLocal
from src.db.models import Match, Prediction
from src.db.unit_of_work import get_uow
from src.llm.context_builder import build_context
from src.llm.provider import LLMProvider, get_default_provider
@@ -143,8 +144,9 @@ async def _predict_single(
except Exception as e:
raise RuntimeError(f"LLM 输出校验失败: {e}")
# 4. 存预测(独立 session,因为 context 用的是自己的 session)
async with AsyncSessionLocal() as db:
# 4. 存预测(使用 UnitOfWork 统一事务)
async with get_uow() as uow:
db = uow.session
# 验证 match 存在
m = await db.get(Match, match_id)
if m is None:
@@ -188,4 +190,5 @@ async def _predict_single(
# 5. 写入缓存
_set_cached(match_id, settings.LLM_PROVIDER, provider.model, version, result)
await uow.commit()
return result