From 7de72e0e83b74dcd7d611bd9d620f10f01460b61 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Fri, 21 Aug 2026 20:37:31 +0800 Subject: [PATCH] refactor: own subscription create transactions --- app/application/subscription/write.py | 174 +++++++++++- app/chain/subscribe.py | 258 +++++++++++------- app/db/oper/subscribe.py | 142 +++++++++- app/startup/modules_initializer.py | 16 +- app/startup/subscription.py | 70 +++++ docs/architecture-overview.md | 7 +- .../backend-architecture-next-stage.md | 16 +- docs/rules/10-data-and-persistent.md | 7 +- tests/conftest.py | 15 +- .../architecture/dependency-baseline.json | 15 +- tests/test_subscribe_create_command.py | 240 ++++++++++++++++ 11 files changed, 840 insertions(+), 120 deletions(-) create mode 100644 app/startup/subscription.py create mode 100644 tests/test_subscribe_create_command.py diff --git a/app/application/subscription/write.py b/app/application/subscription/write.py index 581fbba71..f7bf5fd20 100644 --- a/app/application/subscription/write.py +++ b/app/application/subscription/write.py @@ -15,7 +15,7 @@ app/application/history.py 里整理历史的写入路径同构。 下方 _translate 单点承担，两条链路只在「怎么查、怎么写」上分叉。 """ -from collections.abc import Callable +from collections.abc import Awaitable, Callable from typing import Optional, Protocol, Tuple from app.domain.context import MediaInfo, MusicInfo @@ -26,6 +26,9 @@ from app.schemas.types import MUSIC_ENTITY_ALBUM, MediaType # 而后续按身份去重也会失效，所以必须在查询与建模之前短路 INCOMPLETE_IDENTITY = (0, "媒体身份不完整") +AfterCommitEffect = Callable[[int], None] +AsyncAfterCommitEffect = Callable[[int], Awaitable[None]] + class SubscribeWriter(Protocol): """订阅写入应用服务使用的数据端口。""" @@ -35,16 +38,151 @@ class SubscribeWriter(Protocol): identity: dict, payload: dict, username: Optional[str] = None, + after_commit: Optional[AfterCommitEffect] = None, ) -> Tuple[int, str]: - """同步新增订阅。""" + """同步新增订阅，并在事务成功后执行外部副作用。""" async def async_add( self, identity: dict, payload: dict, username: Optional[str] = None, + after_commit: Optional[AsyncAfterCommitEffect] = None, ) -> Tuple[int, str]: - """异步新增订阅。""" + """异步新增订阅，并在事务成功后执行外部副作用。""" + + +class StagedSubscription(Protocol): + """订阅仓储暂存结果的结构化端口，避免 Application 反向约束适配器类型。""" + + @property + def subscribe_id(self) -> int: + """返回已创建或已存在的订阅 ID。""" + ... + + @property + def message(self) -> str: + """返回兼容旧入口的结果说明。""" + ... + + @property + def created(self) -> bool: + """标识本次是否暂存了一条新记录。""" + ... + + +class SubscriptionStagingRepository(Protocol): + """新增订阅命令需要的无提交仓储端口。""" + + def stage_add( + self, + identity: dict, + payload: dict, + username: Optional[str] = None, + ) -> StagedSubscription: + """暂存同步新增，命中重复订阅时不写入。""" + ... + + async def async_stage_add( + self, + identity: dict, + payload: dict, + username: Optional[str] = None, + ) -> StagedSubscription: + """暂存异步新增，命中重复订阅时不写入。""" + ... + + +class UnitOfWork(Protocol): + """同步订阅新增命令使用的最小事务端口。""" + + def commit(self) -> None: + """提交当前逻辑操作。""" + ... + + def rollback(self) -> None: + """回滚当前逻辑操作。""" + ... + + +class AsyncUnitOfWork(Protocol): + """异步订阅新增命令使用的最小事务端口。""" + + async def commit(self) -> None: + """提交当前逻辑操作。""" + ... + + async def rollback(self) -> None: + """回滚当前逻辑操作。""" + ... + + +class CreateSubscriptionCommand: + """暂存并提交一条同步订阅，重复请求保持历史返回且不产生提交。""" + + def __init__( + self, + repository: SubscriptionStagingRepository, + unit_of_work: UnitOfWork, + ) -> None: + """注入无提交仓储和事务所有者。""" + self._repository = repository + self._unit_of_work = unit_of_work + + def execute( + self, + identity: dict, + payload: dict, + username: Optional[str] = None, + after_commit: Optional[AfterCommitEffect] = None, + ) -> Tuple[int, str]: + """执行同步新增；事务失败回滚，提交后副作用失败不反向回滚。""" + try: + staged = self._repository.stage_add(identity, payload, username) + if staged.created: + self._unit_of_work.commit() + except Exception: + self._unit_of_work.rollback() + raise + if staged.subscribe_id and after_commit: + after_commit(staged.subscribe_id) + return staged.subscribe_id, staged.message + + +class AsyncCreateSubscriptionCommand: + """暂存并提交一条异步订阅，事务成功后才把结果交给副作用调用方。""" + + def __init__( + self, + repository: SubscriptionStagingRepository, + unit_of_work: AsyncUnitOfWork, + ) -> None: + """注入无提交异步仓储和事务所有者。""" + self._repository = repository + self._unit_of_work = unit_of_work + + async def execute( + self, + identity: dict, + payload: dict, + username: Optional[str] = None, + after_commit: Optional[AsyncAfterCommitEffect] = None, + ) -> Tuple[int, str]: + """执行异步新增；事务失败回滚，提交后副作用失败不反向回滚。""" + try: + staged = await self._repository.async_stage_add( + identity, + payload, + username, + ) + if staged.created: + await self._unit_of_work.commit() + except Exception: + await self._unit_of_work.rollback() + raise + if staged.subscribe_id and after_commit: + await after_commit(staged.subscribe_id) + return staged.subscribe_id, staged.message _configured_subscribe_writer: Callable[[], SubscribeWriter] | None = None @@ -130,6 +268,7 @@ def _translate( def add_subscribe( mediainfo: MediaInfo | MusicInfo, subscribe_oper: Optional[SubscribeWriter] = None, + after_commit: Optional[AfterCommitEffect] = None, **kwargs, ) -> Tuple[int, str]: """ @@ -137,6 +276,7 @@ def add_subscribe( :param mediainfo: 识别结果 :param subscribe_oper: 复用的订阅操作对象，未传时由启动组合根提供 + :param after_commit: 数据提交后执行的消息、事件或上报编排 :param kwargs: 订阅设置；owner_scope 为真时按用户名限定查重范围 :return: (订阅 ID, 结果说明)；ID 为 0 表示未新增 """ @@ -145,12 +285,20 @@ def add_subscribe( return INCOMPLETE_IDENTITY identity, payload, username = translated oper = _get_subscribe_writer(subscribe_oper) - return oper.add(identity=identity, payload=payload, username=username) + if after_commit is None: + return oper.add(identity=identity, payload=payload, username=username) + return oper.add( + identity=identity, + payload=payload, + username=username, + after_commit=after_commit, + ) async def async_add_subscribe( mediainfo: MediaInfo | MusicInfo, subscribe_oper: Optional[SubscribeWriter] = None, + after_commit: Optional[AsyncAfterCommitEffect] = None, **kwargs, ) -> Tuple[int, str]: """ @@ -158,6 +306,7 @@ async def async_add_subscribe( :param mediainfo: 识别结果 :param subscribe_oper: 复用的订阅操作对象，未传时由启动组合根提供 + :param after_commit: 数据提交后执行的异步消息、事件或上报编排 :param kwargs: 订阅设置；owner_scope 为真时按用户名限定查重范围 :return: (订阅 ID, 结果说明)；ID 为 0 表示未新增 """ @@ -166,12 +315,27 @@ async def async_add_subscribe( return INCOMPLETE_IDENTITY identity, payload, username = translated oper = _get_subscribe_writer(subscribe_oper) - return await oper.async_add(identity=identity, payload=payload, username=username) + if after_commit is None: + return await oper.async_add(identity=identity, payload=payload, username=username) + return await oper.async_add( + identity=identity, + payload=payload, + username=username, + after_commit=after_commit, + ) __all__ = [ + "AfterCommitEffect", + "AsyncCreateSubscriptionCommand", + "AsyncAfterCommitEffect", + "AsyncUnitOfWork", + "CreateSubscriptionCommand", "INCOMPLETE_IDENTITY", + "StagedSubscription", + "SubscriptionStagingRepository", "SubscribeWriter", + "UnitOfWork", "add_subscribe", "async_add_subscribe", "configure_subscribe_writer", diff --git a/app/chain/subscribe.py b/app/chain/subscribe.py index 5150ff1f3..55fedb116 100644 --- a/app/chain/subscribe.py +++ b/app/chain/subscribe.py @@ -3,6 +3,7 @@ import json import random import threading import time +from dataclasses import dataclass from datetime import datetime from typing import Any, Callable, Dict, List, Optional, Union, Tuple @@ -88,6 +89,24 @@ SystemConfigOper = get_configured_system_config _DEFAULT_SYSTEM_CONFIG_PROVIDER = get_configured_system_config +@dataclass(frozen=True, slots=True) +class _SubscribePostCommitContext: + """订阅提交后副作用所需的不可变业务快照。""" + + title: str + year: str + metainfo: MetaBase + mediainfo: MediaInfo + media_source: Optional[MediaSource] + media_id: Optional[str] + season: Optional[int] + channel: Optional[NotificationChannel] + source: Optional[str] + userid: Optional[str] + username: Optional[str] + message: bool + + def _system_config(): """返回配置端口，并兼容旧测试对本地别名的替换。""" if SystemConfigOper is not _DEFAULT_SYSTEM_CONFIG_PROVIDER: @@ -851,6 +870,98 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): }) return defaults + @staticmethod + def __subscribe_added_link(mtype: MediaType) -> str: + """返回订阅类型对应的前端详情入口。""" + if mtype == MediaType.TV: + return settings.MP_DOMAIN('#/subscribe/tv?tab=mysub') + if mtype == MediaType.MUSIC: + return settings.MP_DOMAIN('#/subscribe/music?tab=mysub') + return settings.MP_DOMAIN('#/subscribe/movie?tab=mysub') + + @staticmethod + def __subscribe_report_payload(context: _SubscribePostCommitContext) -> dict: + """构造保持旧字段和值语义的订阅统计上报。""" + mediainfo = context.mediainfo + music_type = getattr(mediainfo, "music_type", None) + return { + "name": context.title, + "year": context.year, + "type": context.metainfo.type.value, + "media_source": context.media_source, + "media_id": context.media_id, + "music_type": music_type, + "total_tracks": getattr(mediainfo, "total_tracks", None) + if music_type == MUSIC_ENTITY_ALBUM else None, + "season": context.season, + "poster": mediainfo.get_poster_image(), + "backdrop": mediainfo.get_backdrop_image(), + "vote": mediainfo.vote_average, + "description": mediainfo.overview, + } + + def __post_subscribe_added( + self, + subscribe_id: int, + context: _SubscribePostCommitContext, + ) -> None: + """同步执行提交后消息、事件和统计，异常不再触碰数据库事务。""" + if context.message: + self.post_message( + _SchemaMessage( + channel=context.channel, + source=context.source, + mtype=MessageType.Subscribe, + ctype=ContentType.SubscribeAdded, + image=context.mediainfo.get_message_image(), + link=self.__subscribe_added_link(context.mediainfo.type), + userid=context.userid, + username=context.username, + ), + meta=context.metainfo, + mediainfo=context.mediainfo, + username=context.username, + ) + eventmanager.send_event(EventType.SubscribeAdded, { + "subscribe_id": subscribe_id, + "username": context.username, + "mediainfo": context.mediainfo.to_dict(), + }) + MoviePilotServerHelper.sub_reg_async( + self.__subscribe_report_payload(context) + ) + + async def __async_post_subscribe_added( + self, + subscribe_id: int, + context: _SubscribePostCommitContext, + ) -> None: + """异步执行提交后消息、事件和统计，保持与同步入口相同顺序。""" + if context.message: + await self.async_post_message( + _SchemaMessage( + channel=context.channel, + source=context.source, + mtype=MessageType.Subscribe, + ctype=ContentType.SubscribeAdded, + image=context.mediainfo.get_message_image(), + link=self.__subscribe_added_link(context.mediainfo.type), + userid=context.userid, + username=context.username, + ), + meta=context.metainfo, + mediainfo=context.mediainfo, + username=context.username, + ) + await eventmanager.async_send_event(EventType.SubscribeAdded, { + "subscribe_id": subscribe_id, + "username": context.username, + "mediainfo": context.mediainfo.to_dict(), + }) + await MoviePilotServerHelper.async_sub_reg( + self.__subscribe_report_payload(context) + ) + def add(self, title: str, year: str, mtype: MediaType = None, episode_group: Optional[str] = None, @@ -992,8 +1103,33 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): # 添加订阅 kwargs.update(self.__get_default_kwargs(mediainfo.type, **kwargs)) + post_commit_context = _SubscribePostCommitContext( + title=title, + year=year, + metainfo=metainfo, + mediainfo=mediainfo, + media_source=media_source, + media_id=media_id, + season=season, + channel=channel, + source=source, + userid=userid, + username=username, + message=bool(message), + ) + + def _after_commit(subscribe_id: int) -> None: + """把同步提交后的副作用委托给单一顺序实现。""" + self.__post_subscribe_added(subscribe_id, post_commit_context) + # 操作数据库 - sid, err_msg = add_subscribe(mediainfo=mediainfo, season=season, username=username, **kwargs) + sid, err_msg = add_subscribe( + mediainfo=mediainfo, + season=season, + username=username, + after_commit=_after_commit, + **kwargs, + ) if not sid: logger.error(f'{mediainfo.title_year} {err_msg}') if not exist_ok and message: @@ -1007,51 +1143,6 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): image=mediainfo.get_message_image(), userid=userid)) return None, err_msg - elif message: - if mediainfo.type == MediaType.TV: - link = settings.MP_DOMAIN('#/subscribe/tv?tab=mysub') - elif mediainfo.type == MediaType.MUSIC: - link = settings.MP_DOMAIN('#/subscribe/music?tab=mysub') - else: - link = settings.MP_DOMAIN('#/subscribe/movie?tab=mysub') - # 订阅成功按规则发送消息 - self.post_message( - _SchemaMessage( - channel=channel, - source=source, - mtype=MessageType.Subscribe, - ctype=ContentType.SubscribeAdded, - image=mediainfo.get_message_image(), - link=link, - userid=userid, - username=username - ), - meta=metainfo, - mediainfo=mediainfo, - username=username - ) - # 发送事件 - eventmanager.send_event(EventType.SubscribeAdded, { - "subscribe_id": sid, - "username": username, - "mediainfo": mediainfo.to_dict(), - }) - # 统计订阅 - MoviePilotServerHelper.sub_reg_async({ - "name": title, - "year": year, - "type": metainfo.type.value, - "media_source": media_source, - "media_id": media_id, - "music_type": getattr(mediainfo, "music_type", None), - "total_tracks": getattr(mediainfo, "total_tracks", None) - if getattr(mediainfo, "music_type", None) == MUSIC_ENTITY_ALBUM else None, - "season": season, - "poster": mediainfo.get_poster_image(), - "backdrop": mediainfo.get_backdrop_image(), - "vote": mediainfo.vote_average, - "description": mediainfo.overview - }) # 返回结果 return sid, err_msg @@ -1196,8 +1287,36 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): # 列新默认参数 kwargs.update(self.__get_default_kwargs(mediainfo.type, **kwargs)) + post_commit_context = _SubscribePostCommitContext( + title=title, + year=year, + metainfo=metainfo, + mediainfo=mediainfo, + media_source=media_source, + media_id=media_id, + season=season, + channel=channel, + source=source, + userid=userid, + username=username, + message=bool(message), + ) + + async def _after_commit(subscribe_id: int) -> None: + """把异步提交后的副作用委托给单一顺序实现。""" + await self.__async_post_subscribe_added( + subscribe_id, + post_commit_context, + ) + # 操作数据库 - sid, err_msg = await async_add_subscribe(mediainfo=mediainfo, season=season, username=username, **kwargs) + sid, err_msg = await async_add_subscribe( + mediainfo=mediainfo, + season=season, + username=username, + after_commit=_after_commit, + **kwargs, + ) if not sid: logger.error(f'{mediainfo.title_year} {err_msg}') if not exist_ok and message: @@ -1211,51 +1330,6 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): image=mediainfo.get_message_image(), userid=userid)) return None, err_msg - elif message: - if mediainfo.type == MediaType.TV: - link = settings.MP_DOMAIN('#/subscribe/tv?tab=mysub') - elif mediainfo.type == MediaType.MUSIC: - link = settings.MP_DOMAIN('#/subscribe/music?tab=mysub') - else: - link = settings.MP_DOMAIN('#/subscribe/movie?tab=mysub') - # 订阅成功按规则发送消息 - await self.async_post_message( - _SchemaMessage( - channel=channel, - source=source, - mtype=MessageType.Subscribe, - ctype=ContentType.SubscribeAdded, - image=mediainfo.get_message_image(), - link=link, - userid=userid, - username=username - ), - meta=metainfo, - mediainfo=mediainfo, - username=username - ) - # 发送事件 - await eventmanager.async_send_event(EventType.SubscribeAdded, { - "subscribe_id": sid, - "username": username, - "mediainfo": mediainfo.to_dict(), - }) - # 统计订阅 - await MoviePilotServerHelper.async_sub_reg({ - "name": title, - "year": year, - "type": metainfo.type.value, - "media_source": media_source, - "media_id": media_id, - "music_type": getattr(mediainfo, "music_type", None), - "total_tracks": getattr(mediainfo, "total_tracks", None) - if getattr(mediainfo, "music_type", None) == MUSIC_ENTITY_ALBUM else None, - "season": season, - "poster": mediainfo.get_poster_image(), - "backdrop": mediainfo.get_backdrop_image(), - "vote": mediainfo.vote_average, - "description": mediainfo.overview - }) # 返回结果 return sid, err_msg diff --git a/app/db/oper/subscribe.py b/app/db/oper/subscribe.py index 5b5863044..ce41c698a 100644 --- a/app/db/oper/subscribe.py +++ b/app/db/oper/subscribe.py @@ -8,9 +8,13 @@ 留在这一层的只有列类型强转与建库时间戳——它们跟着订阅表的列走，换谁来调都一样。 """ import time +from collections.abc import Awaitable, Callable +from dataclasses import dataclass from typing import Any, Tuple, List, Optional -from sqlalchemy import delete as sqlalchemy_delete +from sqlalchemy import delete as sqlalchemy_delete, select +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm import Session from app.application.subscription.delete import SubscribeDeletionCandidate from app.db.base import DbOper @@ -20,6 +24,18 @@ from app.schemas.types import MediaSource INTEGER_FLAG_FIELDS = ("best_version", "best_version_full", "search_imdbid", "manual_total_episode") +AfterCommitEffect = Callable[[int], None] +AsyncAfterCommitEffect = Callable[[int], Awaitable[None]] + + +@dataclass(frozen=True, slots=True) +class SubscribeStageResult: + """Oper 暂存结果，按 Application 端口需要暴露最小只读状态。""" + + subscribe_id: int + message: str + created: bool + def _normalize_integer_flags(payload: dict, fields: Tuple[str, ...] = INTEGER_FLAG_FIELDS) -> dict: """ @@ -67,6 +83,25 @@ class SubscribeOper(DbOper): 订阅管理 """ + @staticmethod + def _identity_statement(identity: dict, username: Optional[str] = None): + """构造订阅查重语句，SQL 所有权收口在 Oper。""" + condition = Subscribe._identity_condition( # pylint: disable=protected-access + identity.get("media_source"), + identity.get("media_id"), + identity.get("music_type"), + ) + if condition is None or username == "": + return None + statement = select(Subscribe).where(condition) + if username: + statement = statement.where(Subscribe.username == username) + if identity.get("season") is not None: + statement = statement.where(Subscribe.season == identity["season"]) + return statement.where( + Subscribe.episode_group == identity.get("episode_group") + ) + def _exists(self, identity: dict, username: Optional[str]) -> Optional[Any]: """ 按身份查重。 @@ -74,8 +109,18 @@ class SubscribeOper(DbOper): :param username: 非空时只在该用户的订阅内查 :return: 命中的订阅行，未命中为 None """ + if isinstance(self._db, Session): + statement = self._identity_statement(identity, username) + if statement is None: + return None + return self._db.execute(statement).scalars().first() + # 旧 SDK 允许无会话构造 Oper；保留其自动短会话行为，但规范入口不得走这里。 if username: - return Subscribe.exists_by_username(self._db, username=username, **identity) + return Subscribe.exists_by_username( + self._db, + username=username, + **identity, + ) return Subscribe.exists(self._db, **identity) async def _async_exists(self, identity: dict, username: Optional[str]) -> Optional[Any]: @@ -85,12 +130,70 @@ class SubscribeOper(DbOper): :param username: 非空时只在该用户的订阅内查 :return: 命中的订阅行，未命中为 None """ + if isinstance(self._db, AsyncSession): + statement = self._identity_statement(identity, username) + if statement is None: + return None + result = await self._db.execute(statement) + return result.scalars().first() + # 同步路径一样只为无会话旧入口保留 Model 的自动短会话兼容。 if username: - return await Subscribe.async_exists_by_username(self._db, username=username, **identity) + return await Subscribe.async_exists_by_username( + self._db, + username=username, + **identity, + ) return await Subscribe.async_exists(self._db, **identity) + def stage_add( + self, + identity: dict, + payload: dict, + username: Optional[str] = None, + ) -> SubscribeStageResult: + """暂存同步新增并 flush 主键，不提交调用方拥有的事务。""" + if not isinstance(self._db, Session): + raise RuntimeError("同步订阅新增需要调用方提供 Session") + subscribe = self._exists(identity, username) + if subscribe: + return SubscribeStageResult( + subscribe_id=subscribe.id, + message="订阅已存在", + created=False, + ) + subscribe = Subscribe(**_persistable(payload)) + self._db.add(subscribe) + self._db.flush() + if not subscribe.id: + return SubscribeStageResult(0, "新增订阅失败", True) + return SubscribeStageResult(subscribe.id, "新增订阅成功", True) + + async def async_stage_add( + self, + identity: dict, + payload: dict, + username: Optional[str] = None, + ) -> SubscribeStageResult: + """暂存异步新增并 flush 主键，不提交调用方拥有的事务。""" + if not isinstance(self._db, AsyncSession): + raise RuntimeError("异步订阅新增需要调用方提供 AsyncSession") + subscribe = await self._async_exists(identity, username) + if subscribe: + return SubscribeStageResult( + subscribe_id=subscribe.id, + message="订阅已存在", + created=False, + ) + subscribe = Subscribe(**_persistable(payload)) + self._db.add(subscribe) + await self._db.flush() + if not subscribe.id: + return SubscribeStageResult(0, "新增订阅失败", True) + return SubscribeStageResult(subscribe.id, "新增订阅成功", True) + def add(self, identity: dict, payload: dict, - username: Optional[str] = None) -> Tuple[int, str]: + username: Optional[str] = None, + after_commit: Optional[AfterCommitEffect] = None) -> Tuple[int, str]: """ 新增订阅：命中既有订阅则原样返回，否则落库后回读。 @@ -99,33 +202,44 @@ class SubscribeOper(DbOper): :param identity: 查重身份（media_source/media_id/music_type/season/episode_group） :param payload: 订阅表的写入字段，媒体翻译由 application/subscription/write.py 完成 :param username: 非空时把查重限定在该用户的订阅内 + :param after_commit: 兼容旧调用方的提交后副作用；新入口由 Application Command 调用 :return: (订阅 ID, 结果说明)；ID 为 0 表示未新增 """ subscribe = self._exists(identity, username) if subscribe: + if after_commit: + after_commit(subscribe.id) return subscribe.id, "订阅已存在" Subscribe(**_persistable(payload)).create(self._db) subscribe = self._exists(identity, username) if not subscribe: return 0, "新增订阅失败" + if after_commit: + after_commit(subscribe.id) return subscribe.id, "新增订阅成功" async def async_add(self, identity: dict, payload: dict, - username: Optional[str] = None) -> Tuple[int, str]: + username: Optional[str] = None, + after_commit: Optional[AsyncAfterCommitEffect] = None) -> Tuple[int, str]: """ 异步新增订阅，语义与 add 完全一致。 :param identity: 查重身份（media_source/media_id/music_type/season/episode_group） :param payload: 订阅表的写入字段，媒体翻译由 application/subscription/write.py 完成 :param username: 非空时把查重限定在该用户的订阅内 + :param after_commit: 兼容旧调用方的异步提交后副作用 :return: (订阅 ID, 结果说明)；ID 为 0 表示未新增 """ subscribe = await self._async_exists(identity, username) if subscribe: + if after_commit: + await after_commit(subscribe.id) return subscribe.id, "订阅已存在" await Subscribe(**_persistable(payload)).async_create(self._db) subscribe = await self._async_exists(identity, username) if not subscribe: return 0, "新增订阅失败" + if after_commit: + await after_commit(subscribe.id) return subscribe.id, "新增订阅成功" def exists( @@ -143,7 +257,7 @@ class SubscribeOper(DbOper): "season": season, "episode_group": episode_group, } - return bool(Subscribe.exists(self._db, **identity_params)) + return bool(self._exists(identity_params, username=None)) async def async_exists( self, media_source: MediaSource, media_id: str, @@ -151,13 +265,15 @@ class SubscribeOper(DbOper): music_type: Optional[str] = None, ) -> Optional[Subscribe]: """异步按媒体身份、季号及可选剧集组读取命中的订阅。""" - return await Subscribe.async_exists( - self._db, - media_source=media_source, - media_id=media_id, - music_type=music_type, - season=season, - episode_group=episode_group, + return await self._async_exists( + { + "media_source": media_source, + "media_id": media_id, + "music_type": music_type, + "season": season, + "episode_group": episode_group, + }, + username=None, ) def get(self, sid: int) -> Optional[Subscribe]: diff --git a/app/startup/modules_initializer.py b/app/startup/modules_initializer.py index 6bba1b837..7076226b2 100644 --- a/app/startup/modules_initializer.py +++ b/app/startup/modules_initializer.py @@ -61,7 +61,13 @@ from app.adapters.external.server import ( ) from app.application.server.report import ServerReportService from app.application.server.share import ServerSharingService -from app.db.session import close_database, get_async_db, get_db +from app.db.session import ( + SessionFactory, + async_session_scope, + close_database, + get_async_db, + get_db, +) from app.db.uow import SqlAlchemyAsyncUnitOfWork, SqlAlchemyUnitOfWork from app.db.oper.subscribe import SubscribeOper from app.db.oper.agentchat import AgentChatOper @@ -90,6 +96,7 @@ from app.startup.managed_resources_initializer import ( init_managed_resources, stop_managed_resources, ) +from app.startup.subscription import TransactionalSubscribeWriter from app.adapters.web.security.access import set_superuser_token_payload_provider from app.application.security.auth import build_superuser_token_payload from app.application.image import configure_wallpaper_providers @@ -465,7 +472,12 @@ async def init_modules(): workflow=lambda: WorkflowOper(), plugin_data=lambda: PluginDataOper(), ) - configure_subscribe_writer(lambda: SubscribeOper()) + configure_subscribe_writer( + lambda: TransactionalSubscribeWriter( + sync_session=SessionFactory, + async_session=async_session_scope, + ) + ) # 托管资源只在这里装配声明与 adapter，具体资源仍由首个消费者显式激活。 init_managed_resources() # 应用服务不反向依赖 Chain，由启动组合层注入壁纸来源。 diff --git a/app/startup/subscription.py b/app/startup/subscription.py new file mode 100644 index 000000000..f1ca37f09 --- /dev/null +++ b/app/startup/subscription.py @@ -0,0 +1,70 @@ +"""订阅写入事务适配器的启动装配。""" + +from collections.abc import Callable +from contextlib import AbstractAsyncContextManager + +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm import Session + +from app.application.subscription.write import ( + AfterCommitEffect, + AsyncAfterCommitEffect, + AsyncCreateSubscriptionCommand, + CreateSubscriptionCommand, +) +from app.db.oper.subscribe import SubscribeOper +from app.db.uow import SqlAlchemyAsyncUnitOfWork, SqlAlchemyUnitOfWork + + +class TransactionalSubscribeWriter: + """为每次订阅新增创建独占会话，并把提交权交给 Application Command。""" + + def __init__( + self, + sync_session: Callable[[], Session], + async_session: Callable[ + [], + AbstractAsyncContextManager[AsyncSession], + ], + ) -> None: + """注入同步会话工厂和异步会话作用域。""" + self._sync_session = sync_session + self._async_session = async_session + + def add( + self, + identity: dict, + payload: dict, + username: str | None = None, + after_commit: AfterCommitEffect | None = None, + ) -> tuple[int, str]: + """在独占同步会话内执行一次完整订阅新增事务。""" + session = self._sync_session() + try: + command = CreateSubscriptionCommand( + repository=SubscribeOper(session), + unit_of_work=SqlAlchemyUnitOfWork(session), + ) + return command.execute(identity, payload, username, after_commit) + finally: + session.close() + + async def async_add( + self, + identity: dict, + payload: dict, + username: str | None = None, + after_commit: AsyncAfterCommitEffect | None = None, + ) -> tuple[int, str]: + """在独占异步会话作用域内执行一次完整订阅新增事务。""" + async with self._async_session() as session: + command = AsyncCreateSubscriptionCommand( + repository=SubscribeOper(session), + unit_of_work=SqlAlchemyAsyncUnitOfWork(session), + ) + return await command.execute( + identity, + payload, + username, + after_commit, + ) diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index 47b169491..801119102 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -361,8 +361,11 @@ flowchart LR 归 `app/application/`（见 `application/subscription/write.py`、`application/history.py`）。 订阅新增、查询、变更、删除、身份和搜索契约已经统一收口在 `application/subscription/`， 不再保留主题包之外的第二个写入入口。 -- Oper 只 stage mutation，不创建独立 Session、不提交；Application Command 通过请求或任务 - 入口注入的 UnitOfWork 统一 `commit/rollback`，事件、刷新和上报只在 commit 成功后执行。 +- 规范写入口中的 Oper 只 stage mutation，不创建独立 Session、不提交；Application Command + 通过请求或任务入口注入的 UnitOfWork 统一 `commit/rollback`，事件、刷新和上报只在 commit + 成功后执行。订阅新增样板由 `startup/subscription.py` 创建独占 Session， + `application/subscription/write.py` 决定事务与 post-commit 边界，`SubscribeOper.stage_add()` + 只查重、`add` 和 `flush`。旧 SDK 显式构造的无会话 Oper 暂留兼容自动短会话，不得被新代码复用。 `transaction-debt-baseline.json` 将存量 178 个 Model 事务装饰器冻结为只降不增低水位。 - 每次表结构变更必须新增 `database/versions/` 下的 Alembic 迁移。 - 运行期业务配置使用 `SystemConfigKey` 枚举 + `SystemConfigOper`，禁止裸字符串键； diff --git a/docs/refactor/backend-architecture-next-stage.md b/docs/refactor/backend-architecture-next-stage.md index 57baa542f..66498573e 100644 --- a/docs/refactor/backend-architecture-next-stage.md +++ b/docs/refactor/backend-architecture-next-stage.md @@ -6,7 +6,7 @@ > 审计范围：宿主后端；排除 `app/plugins/**` 运行时插件副本 > 规范优先级：`AGENTS.md` 与 `docs/rules/` 高于本文 > 相关文档：`docs/architecture-overview.md`、`docs/refactor/backend-architecture-governance.md`、`docs/refactor/backend-module-refactor-compatibility.md` -> 实施进度：阶段 0（ARCH-201～203）、阶段 1（ARCH-210～212）与 ARCH-220 已完成，后续任务按 ID 独立提交和回滚 +> 实施进度：阶段 0（ARCH-201～203）、阶段 1（ARCH-210～212）与 ARCH-220～221 已完成，后续任务按 ID 独立提交和回滚 ## 1. 结论先行 @@ -373,6 +373,20 @@ flowchart TB **完成标准**：一个业务动作只有一个事务所有者；任意入口都不会因内部 Model 方法提前 commit 而产生部分写入。 +**实施记录（2026-08-21）**： + +- `app/startup/subscription.py` 为每次规范新增创建独占同步/异步 Session； + `CreateSubscriptionCommand` / `AsyncCreateSubscriptionCommand` 持有 UoW，Oper 只执行 + 查重、`add` 与 `flush`。 +- `SubscribeOper.stage_add()` 的查重 SQL 已收口到 Oper，不再调用 Model 自动会话装饰器； + 无会话构造 `SubscribeOper()` 的旧 SDK 路径保留原自动短会话和返回值，未扩散为规范入口。 +- Chain 把原有“成功消息 → `SubscribeAdded` 事件 → Server 统计”作为显式 post-commit + 回调交给 Command；commit/flush 失败回滚，事件或上报失败只传播原异常，不回滚已提交记录。 +- 同步/异步 `SubscribeChain.add` 方法长度从各 203 行降至 183/186 行；新增 9 个事务边界测试， + 覆盖成功顺序、commit/flush 失败、重复请求、Oper 不提交、事件失败、上报失败与真实落库。 +- Model 装饰器总数仍为 178：本切片绕开了继承自 `Base.create/async_create` 的自动提交， + 但为保留既有 Model/旧 SDK 查询兼容未机械删除查询装饰器；ratchet 保持不增，后续切片继续下降。 + #### ARCH-222：按风险迁移其余写用例 推荐顺序： diff --git a/docs/rules/10-data-and-persistent.md b/docs/rules/10-data-and-persistent.md index 2c49cb2f8..0d71fe03b 100644 --- a/docs/rules/10-data-and-persistent.md +++ b/docs/rules/10-data-and-persistent.md @@ -99,6 +99,11 @@ Oper classes accept and return persistence values. Turning a `MediaInfo` or - A synchronous Session is private to one worker thread. An AsyncSession is private to one asyncio task/operation; neither may be stored in a process singleton or reused by concurrent work. +- Subscription creation is the reference slice: `app/startup/subscription.py` + creates an exclusive Session, `app/application/subscription/write.py` owns the + UoW and post-commit callback, and `SubscribeOper.stage_add()` only queries, + adds, and flushes. Preserve `SubscribeOper.add()` only for legacy SDK callers; + new host code must not use that auto-commit compatibility path. Run `./.venv/bin/python scripts/architecture/baseline.py --check-host` after persistence changes. A deliberate debt reduction may refresh the low-water mark @@ -214,4 +219,4 @@ When `REDIS_HOST` is configured, `app/modules/redis/` provides a distributed cac - `settings.API_TOKEN` and other secret fields must not be included in log output or API responses. - The `config list --show-secrets` flag exists specifically to gate secret visibility in the CLI. -*Last Updated: 2026-08-14* +*Last Updated: 2026-08-21* diff --git a/tests/conftest.py b/tests/conftest.py index f4d4fbdce..f0e5f9dcd 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -30,13 +30,19 @@ def configure_plugin_system_services(): from app.api.data import configure_api_data_ports from app.application.configuration import SystemConfigService, configure_system_config from app.application.service import configure_service_directory - from app.db.session import get_async_db, get_db + from app.db.session import ( + SessionFactory, + async_session_scope, + get_async_db, + get_db, + ) from app.db.uow import SqlAlchemyAsyncUnitOfWork, SqlAlchemyUnitOfWork from app.db.oper.systemconfig import SystemConfigOper configure_token_codec(create_access_token, decode_access_token) configure_system_config(SystemConfigService(repository=SystemConfigOper())) from app.application.chain.data import configure_chain_data_ports + from app.application.subscription.write import configure_subscribe_writer from app.application.plugin.runtime import configure_plugin_runtime from app.application.module import configure_module_runtime from app.application.chain.context import ( @@ -73,6 +79,7 @@ def configure_plugin_system_services(): from app.db.oper.workflow import WorkflowOper from app.db.oper.message import MessageOper from app.db.oper.passkey import PassKeyOper + from app.startup.subscription import TransactionalSubscribeWriter configure_api_data_ports( sync_session=get_db, @@ -100,6 +107,12 @@ def configure_plugin_system_services(): "sync": SqlAlchemyUnitOfWork, }, ) + configure_subscribe_writer( + lambda: TransactionalSubscribeWriter( + sync_session=SessionFactory, + async_session=async_session_scope, + ) + ) configure_chain_data_ports( site=lambda: SiteOper(), diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 5d3c9d5d9..6d7660350 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -13,8 +13,8 @@ "runtime_to_db": [], "workflow_to_db": [] }, - "edge_count": 6084, - "edge_sha256": "0dbce22b56284b845591fd38771a96c937ec898db7e9d1aece1ec47ceff8de04", + "edge_count": 6092, + "edge_sha256": "1935d43d8d3c0c3e3687109f0b81a4e56cbc4bc2ed4763a706e2b0d7c119f378", "edges": [ "app -> app.runtime", "app -> app.runtime.compat", @@ -5873,6 +5873,7 @@ "app.startup.modules_initializer -> app.startup.agent_initializer", "app.startup.modules_initializer -> app.startup.database", "app.startup.modules_initializer -> app.startup.managed_resources_initializer", + "app.startup.modules_initializer -> app.startup.subscription", "app.startup.monitor_initializer -> app.monitor", "app.startup.plugins_initializer -> app.adapters", "app.startup.plugins_initializer -> app.adapters.external", @@ -5922,6 +5923,13 @@ "app.startup.scheduler_initializer -> app.application", "app.startup.scheduler_initializer -> app.application.scheduling", "app.startup.scheduler_initializer -> app.scheduler", + "app.startup.subscription -> app.application", + "app.startup.subscription -> app.application.subscription", + "app.startup.subscription -> app.application.subscription.write", + "app.startup.subscription -> app.db", + "app.startup.subscription -> app.db.oper", + "app.startup.subscription -> app.db.oper.subscribe", + "app.startup.subscription -> app.db.uow", "app.startup.transfer_initializer -> app.chain", "app.startup.transfer_initializer -> app.chain.transfer", "app.startup.workflow_initializer -> app.workflow", @@ -6101,7 +6109,7 @@ "app.workflow.actions.transfer_file -> app.workflow", "app.workflow.actions.transfer_file -> app.workflow.actions" ], - "module_count": 756, + "module_count": 757, "modules": [ "app", "app.adapters", @@ -6836,6 +6844,7 @@ "app.startup.plugins_initializer", "app.startup.routers_initializer", "app.startup.scheduler_initializer", + "app.startup.subscription", "app.startup.transfer_initializer", "app.startup.workflow_initializer", "app.testing", diff --git a/tests/test_subscribe_create_command.py b/tests/test_subscribe_create_command.py new file mode 100644 index 000000000..286866316 --- /dev/null +++ b/tests/test_subscribe_create_command.py @@ -0,0 +1,240 @@ +"""订阅新增事务所有权与默认入口集成测试。""" + +import asyncio +from unittest.mock import AsyncMock, Mock + +import pytest + +from app.application.subscription.write import ( + AsyncCreateSubscriptionCommand, + CreateSubscriptionCommand, + add_subscribe, + async_add_subscribe, +) +from app.db.models.subscribe import Subscribe +from app.db.oper.subscribe import SubscribeOper, SubscribeStageResult +from app.domain.context import MediaInfo +from app.schemas.types import MediaSource, MediaType + + +def _media(media_id: str) -> MediaInfo: + """构造默认事务写入路径所需的最小媒体信息。""" + media = MediaInfo() + media.type = MediaType.MOVIE + media.title = "事务测试电影" + media.year = "2026" + media.media_source = MediaSource.TMDB + media.media_id = media_id + media.vote_average = 8.0 + media.overview = "事务切片" + return media + + +def test_sync_command_orders_stage_commit_before_caller_effect() -> None: + """同步新增只有在仓储暂存和提交成功后才把结果交给外部副作用。""" + calls: list[str] = [] + repository = Mock() + repository.stage_add.side_effect = lambda *_: ( + calls.append("stage") + or SubscribeStageResult(10, "新增订阅成功", True) + ) + unit_of_work = Mock() + unit_of_work.commit.side_effect = lambda: calls.append("commit") + command = CreateSubscriptionCommand(repository, unit_of_work) + + result = command.execute({"media_id": "10"}, {"name": "demo"}) + calls.append("effect") + + assert result == (10, "新增订阅成功") + assert calls == ["stage", "commit", "effect"] + unit_of_work.rollback.assert_not_called() + + +def test_sync_command_rolls_back_commit_failure_and_skips_effect() -> None: + """提交失败必须回滚并传播原异常，调用方不能误执行提交后副作用。""" + commit_error = RuntimeError("commit failed") + repository = Mock() + repository.stage_add.return_value = SubscribeStageResult( + 11, + "新增订阅成功", + True, + ) + unit_of_work = Mock() + unit_of_work.commit.side_effect = commit_error + command = CreateSubscriptionCommand(repository, unit_of_work) + effects: list[str] = [] + + with pytest.raises(RuntimeError) as raised: + command.execute({"media_id": "11"}, {"name": "demo"}) + effects.append("effect") + + assert raised.value is commit_error + assert effects == [] + unit_of_work.rollback.assert_called_once_with() + + +def test_sync_command_does_not_commit_duplicate_request() -> None: + """查重命中沿用旧 ID 和消息，不开启无意义写事务。""" + repository = Mock() + repository.stage_add.return_value = SubscribeStageResult( + 12, + "订阅已存在", + False, + ) + unit_of_work = Mock() + command = CreateSubscriptionCommand(repository, unit_of_work) + + assert command.execute({}, {}) == (12, "订阅已存在") + unit_of_work.commit.assert_not_called() + unit_of_work.rollback.assert_not_called() + + +def test_sync_event_failure_does_not_roll_back_committed_subscription() -> None: + """事件属于提交后副作用，失败只向上传播且不能伪装成数据库回滚。""" + calls: list[str] = [] + repository = Mock() + repository.stage_add.return_value = SubscribeStageResult( + 13, + "新增订阅成功", + True, + ) + unit_of_work = Mock() + unit_of_work.commit.side_effect = lambda: calls.append("commit") + event_error = RuntimeError("event failed") + + def send_event(_subscribe_id: int) -> None: + """模拟 Chain 在提交后发送订阅事件失败。""" + calls.append("event") + raise event_error + + command = CreateSubscriptionCommand(repository, unit_of_work) + + with pytest.raises(RuntimeError) as raised: + command.execute({}, {}, after_commit=send_event) + + assert raised.value is event_error + assert calls == ["commit", "event"] + unit_of_work.rollback.assert_not_called() + + +@pytest.mark.asyncio +async def test_async_command_rolls_back_staging_failure() -> None: + """异步 flush 或唯一约束失败同样由命令回滚，不留部分写入。""" + staging_error = RuntimeError("flush failed") + repository = Mock() + repository.async_stage_add = AsyncMock(side_effect=staging_error) + unit_of_work = Mock() + unit_of_work.commit = AsyncMock() + unit_of_work.rollback = AsyncMock() + command = AsyncCreateSubscriptionCommand(repository, unit_of_work) + + with pytest.raises(RuntimeError) as raised: + await command.execute({}, {}) + + assert raised.value is staging_error + unit_of_work.commit.assert_not_awaited() + unit_of_work.rollback.assert_awaited_once_with() + + +@pytest.mark.asyncio +async def test_async_report_failure_happens_after_event_without_rollback() -> None: + """异步上报失败保留事件先行顺序，也不回滚已经提交的订阅。""" + calls: list[str] = [] + repository = Mock() + repository.async_stage_add = AsyncMock( + return_value=SubscribeStageResult(14, "新增订阅成功", True) + ) + unit_of_work = Mock() + unit_of_work.commit = AsyncMock(side_effect=lambda: calls.append("commit")) + unit_of_work.rollback = AsyncMock() + report_error = RuntimeError("report failed") + + async def send_event_and_report(_subscribe_id: int) -> None: + """模拟 Chain 先发事件再执行统计上报。""" + calls.append("event") + calls.append("report") + raise report_error + + command = AsyncCreateSubscriptionCommand(repository, unit_of_work) + + with pytest.raises(RuntimeError) as raised: + await command.execute({}, {}, after_commit=send_event_and_report) + + assert raised.value is report_error + assert calls == ["commit", "event", "report"] + unit_of_work.rollback.assert_not_awaited() + + +def test_default_sync_writer_persists_once_and_reuses_duplicate(db) -> None: + """Chain 默认入口使用独立事务写入，重复媒体身份返回同一订阅。""" + db.watermark(Subscribe) + media = _media("arch-221-sync") + after_commit = Mock() + + first = add_subscribe(mediainfo=media, after_commit=after_commit) + second = add_subscribe(mediainfo=media, after_commit=after_commit) + + assert first[0] > 0 + assert first[1] == "新增订阅成功" + assert second == (first[0], "订阅已存在") + db.session.expire_all() + rows = Subscribe.list_by_media_identity( + db.session, + media_source=MediaSource.TMDB, + media_id="arch-221-sync", + ) + assert [row.id for row in rows] == [first[0]] + assert after_commit.call_args_list == [ + ((first[0],), {}), + ((first[0],), {}), + ] + + +def test_stage_add_executes_identity_sql_in_oper(db, monkeypatch) -> None: + """规范新增路径直接由 Oper 查询，不能退回 Model 自动会话装饰器。""" + db.watermark(Subscribe) + commit = Mock(wraps=db.session.commit) + monkeypatch.setattr(db.session, "commit", commit) + monkeypatch.setattr( + Subscribe, + "exists", + Mock(side_effect=AssertionError("model query must not run")), + ) + oper = SubscribeOper(db.session) + identity = { + "media_source": str(MediaSource.TMDB), + "media_id": "arch-221-stage", + "music_type": None, + "season": None, + "episode_group": None, + } + + staged = oper.stage_add( + identity, + { + "name": "Oper SQL", + "type": MediaType.MOVIE.value, + "state": "N", + **identity, + }, + ) + + assert staged.created is True + assert staged.subscribe_id > 0 + commit.assert_not_called() + db.session.rollback() + + +def test_default_async_writer_persists_committed_row(db) -> None: + """Agent/API 使用的异步 Chain 入口在返回前已完成请求级提交。""" + db.watermark(Subscribe) + media = _media("arch-221-async") + + subscribe_id, message = asyncio.run(async_add_subscribe(mediainfo=media)) + + assert subscribe_id > 0 + assert message == "新增订阅成功" + db.session.expire_all() + persisted = Subscribe.get(db.session, subscribe_id) + assert persisted is not None + assert persisted.media_id == "arch-221-async"