shangfangjian
53e602b4f6
- unit_of_work.py: 简化为纯 session 上下文管理器,消除 double-close 风险 - validation.py: 移除未使用的 import,简化 _score_to_1x2 逻辑 - repositories.py: 将 func 导入移到模块顶层 - 更新所有 get_uow() 调用点使用新接口(yield session 而非 uow 对象)
34 lines
853 B
Python
34 lines
853 B
Python
"""工作单元(Unit of Work):统一事务边界。
|
|
|
|
使用方式:
|
|
async with get_uow() as uow:
|
|
await uow.session.get(Match, 1)
|
|
await uow.commit()
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import AsyncIterator
|
|
from contextlib import asynccontextmanager
|
|
|
|
from src.db.base import AsyncSessionLocal
|
|
|
|
|
|
@asynccontextmanager
|
|
async def get_uow() -> AsyncIterator[AsyncSessionLocal]:
|
|
"""创建新的工作单元(用于非路由上下文)。
|
|
|
|
用法:
|
|
async with get_uow() as session:
|
|
await session.get(...)
|
|
# 退出时自动 commit(无异常) 或 rollback(有异常)
|
|
"""
|
|
session = AsyncSessionLocal()
|
|
try:
|
|
yield session
|
|
await session.commit()
|
|
except Exception:
|
|
await session.rollback()
|
|
raise
|
|
finally:
|
|
await session.close()
|