From 67ae166a54bc9b88c06f2f92dd97fc0adce2d38a Mon Sep 17 00:00:00 2001 From: jxxghp Date: Fri, 18 Sep 2026 22:13:33 +0800 Subject: [PATCH] fix(servarr): restore subscription repository async contract --- app/api/dependencies/subscription.py | 5 +- app/application/servarr.py | 14 ++++- app/db/adapters/subscription.py | 69 +++++++++++++++++++++ tests/test_servarr_series_add.py | 38 +++++++++++- tests/test_subscription_repository.py | 87 +++++++++++++++++++++++++++ 5 files changed, 209 insertions(+), 4 deletions(-) diff --git a/app/api/dependencies/subscription.py b/app/api/dependencies/subscription.py index b9295e0bc..2a37cc2b0 100644 --- a/app/api/dependencies/subscription.py +++ b/app/api/dependencies/subscription.py @@ -16,6 +16,7 @@ from app.api.context import ( get_subscription_outbox, get_subscription_outbox_store, get_subscription_repository, + get_subscription_session, get_subscription_transaction, get_sync_session, resolve_background_task_registry, @@ -260,12 +261,14 @@ def get_servarr_subscription_batch_writer( def get_servarr_subscription_service( - async_db: AsyncSession = Depends(get_async_session), + async_db: AsyncSession = Depends(get_subscription_session), db: Session = Depends(get_sync_session), + unit_of_work: object = Depends(get_subscription_transaction), runtime: HostRuntime = Depends(get_host_runtime), ) -> ServarrSubscriptionService: """组装 Servarr 兼容路由的请求级订阅数据用例。""" return ServarrSubscriptionService( async_repository=runtime.subscription.repository(async_db), sync_repository=runtime.subscription.repository(db), + unit_of_work=cast(DeleteUnitOfWork, unit_of_work), ) diff --git a/app/application/servarr.py b/app/application/servarr.py index 5cd318c27..9b6c61873 100644 --- a/app/application/servarr.py +++ b/app/application/servarr.py @@ -5,6 +5,7 @@ from __future__ import annotations from dataclasses import dataclass from typing import Any, Optional, Protocol +from app.application.subscription.delete import AsyncUnitOfWork from app.schemas.types import MediaSource @@ -79,10 +80,12 @@ class ServarrSubscriptionService: *, async_repository: ServarrAsyncSubscriptionRepository, sync_repository: ServarrSyncSubscriptionRepository, + unit_of_work: Optional[AsyncUnitOfWork] = None, ) -> None: - """保存请求级同步和异步订阅仓储。""" + """保存订阅仓储，并按需保存请求级异步事务端口。""" self._async_repository = async_repository self._sync_repository = sync_repository + self._unit_of_work = unit_of_work async def list(self) -> list[ServarrSubscription]: """读取全部订阅并转换为脱离 ORM 会话的投影。""" @@ -136,7 +139,14 @@ class ServarrSubscriptionService: """删除存在的订阅并报告是否实际命中。""" if not await self._async_repository.async_get(subscribe_id): return False - await self._async_repository.async_delete(subscribe_id) + try: + await self._async_repository.async_delete(subscribe_id) + if self._unit_of_work is not None: + await self._unit_of_work.commit() + except Exception: + if self._unit_of_work is not None: + await self._unit_of_work.rollback() + raise return True @staticmethod diff --git a/app/db/adapters/subscription.py b/app/db/adapters/subscription.py index f01a5c4a4..f3957f8d5 100644 --- a/app/db/adapters/subscription.py +++ b/app/db/adapters/subscription.py @@ -305,6 +305,21 @@ class TransactionalSubscriptionRepository(_TransactionalSubscriptionWriter): async with self._async_session() as session: return await operation(SubscribeOper(session)) + async def _async_write( + self, + operation: Callable[[SubscribeOper], Awaitable[T]], + ) -> T: + """在独立异步 Session 中执行写入并提交事务。""" + async with self._async_session() as session: + unit_of_work = SqlAlchemyAsyncUnitOfWork(session) + try: + result = await operation(SubscribeOper(session)) + await unit_of_work.commit() + return result + except Exception: + await unit_of_work.rollback() + raise + def exists(self, identity: SubscriptionIdentity) -> bool: """同步判断媒体身份是否已有订阅。""" return bool( @@ -363,6 +378,29 @@ class TransactionalSubscriptionRepository(_TransactionalSubscriptionWriter): """同步按可选状态读取订阅快照。""" return self._read(lambda repository: [_project_subscription(record) for record in repository.list(state)]) + async def async_exists( + self, + media_source: MediaSource, + media_id: str, + season: Optional[int] = None, + episode_group: Optional[str] = None, + music_type: Optional[str] = None, + ) -> Optional[SubscriptionSnapshot]: + """异步按媒体身份、季号及可选剧集组读取命中的订阅快照。""" + + async def operation(repository: SubscribeOper) -> Optional[SubscriptionSnapshot]: + """在独立 Session 中执行 Servarr 查重并投影快照。""" + record = await repository.async_exists( + media_source=media_source, + media_id=media_id, + season=season, + episode_group=episode_group, + music_type=music_type, + ) + return _project_subscription(record) if record is not None else None + + return await self._async_read(operation) + async def async_get(self, subscribe_id: int) -> Optional[SubscriptionSnapshot]: """异步按主键读取订阅快照。""" @@ -373,6 +411,15 @@ class TransactionalSubscriptionRepository(_TransactionalSubscriptionWriter): return await self._async_read(operation) + async def async_delete(self, subscribe_id: int) -> None: + """在独立异步事务中删除订阅。""" + + async def operation(repository: SubscribeOper) -> None: + """在当前独立 Session 中暂存订阅删除。""" + await repository.async_delete(subscribe_id) + + await self._async_write(operation) + async def async_list( self, state: Optional[str] = None, @@ -616,6 +663,28 @@ class SessionSubscriptionRepository: record = await self._async_repository().async_get(subscribe_id) return _project_subscription(record) if record is not None else None + async def async_exists( + self, + media_source: MediaSource, + media_id: str, + season: Optional[int] = None, + episode_group: Optional[str] = None, + music_type: Optional[str] = None, + ) -> Optional[SubscriptionSnapshot]: + """异步按媒体身份、季号及可选剧集组读取命中的订阅快照。""" + record = await self._async_repository().async_exists( + media_source=media_source, + media_id=media_id, + season=season, + episode_group=episode_group, + music_type=music_type, + ) + return _project_subscription(record) if record is not None else None + + async def async_delete(self, subscribe_id: int) -> None: + """在调用方异步事务中暂存订阅删除，不提交事务。""" + await self._async_repository().stage_delete(subscribe_id) + async def async_list( self, state: Optional[str] = None, diff --git a/tests/test_servarr_series_add.py b/tests/test_servarr_series_add.py index af4227e1f..e4a7f10fe 100644 --- a/tests/test_servarr_series_add.py +++ b/tests/test_servarr_series_add.py @@ -13,8 +13,11 @@ import pytest from fastapi import HTTPException from app.api.servarr import arr_add_series, arr_series_lookup -from app.schemas import SonarrSeason, SonarrSeries +from app.application.servarr import ServarrSubscriptionService +from app.db.adapters.subscription import SessionSubscriptionRepository +from app.db.models.subscribe import Subscribe from app.schemas.types import MediaSource, MediaType +from app.schemas.servarr import SonarrSeason, SonarrSeries _TVDB_ID = 454898 _TMDB_ID = 236534 @@ -132,6 +135,39 @@ def test_add_series_already_subscribed_returns_existing(): subscribe_chain.async_add_batch.assert_not_awaited() +def test_add_series_existing_subscription_uses_async_repository(db): + """真实请求仓储查到已有季订阅时，兼容端点应幂等返回而不抛异常。""" + db.add( + Subscribe( + name="已存在的 Servarr 订阅", + type=MediaType.TV.value, + media_source=MediaSource.TMDB.value, + media_id=str(_TMDB_ID), + season=1, + ) + ) + + async def execute(session): + """通过真实请求级仓储执行新增剧集检查。""" + service = ServarrSubscriptionService( + async_repository=SessionSubscriptionRepository(session), + sync_repository=SessionSubscriptionRepository(db.session), + ) + return await arr_add_series( + tv=_series( + tmdb_id=_TMDB_ID, + seasons=[SonarrSeason(seasonNumber=1, monitored=True)], + ), + _="api-token", + subscriptions=service, + batch_writer=MagicMock(), + ) + + result = db.run_async_session(execute) + + assert result.id == 1 + + def test_add_series_identity_resolution_failure_returns_500(): """媒体身份补全失败时返回 500，避免 Seerr 误判请求已成功。""" tv = _series() diff --git a/tests/test_subscription_repository.py b/tests/test_subscription_repository.py index fc367795f..f68fa43b5 100644 --- a/tests/test_subscription_repository.py +++ b/tests/test_subscription_repository.py @@ -115,6 +115,93 @@ def test_servarr_sync_lookup_reads_session_repository(db) -> None: assert [(item.id, item.name, item.media_id) for item in result] == [(row.id, "同步 lookup 订阅", "550")] +def test_servarr_async_exists_reads_session_repository(db) -> None: + """Servarr 异步查重应从请求级订阅仓储读取命中的订阅。""" + row = db.add( + Subscribe( + name="异步查重订阅", + type=MediaType.TV.value, + media_source=MediaSource.TMDB.value, + media_id="async-servarr-exists", + season=1, + ) + ) + + async def query(session: AsyncSession) -> bool: + """通过真实请求仓储执行 Servarr 异步查重。""" + service = ServarrSubscriptionService( + async_repository=SessionSubscriptionRepository(session), + sync_repository=SessionSubscriptionRepository(db.session), + ) + return await service.exists( + media_source=MediaSource.TMDB, + media_id="async-servarr-exists", + season=1, + ) + + assert db.run_async_session(query) + assert row.id is not None + + +def test_servarr_delete_commits_session_repository_transaction(db) -> None: + """Servarr 删除应提交请求仓储暂存的删除，而不是随会话关闭回滚。""" + row = db.add( + Subscribe( + name="待删除 Servarr 订阅", + type=MediaType.TV.value, + media_source=MediaSource.TMDB.value, + media_id="servarr-delete", + season=1, + ) + ) + subscribe_id = row.id + + async def delete(session: AsyncSession) -> bool: + """通过真实请求仓储和 UoW 执行 Servarr 删除。""" + service = ServarrSubscriptionService( + async_repository=SessionSubscriptionRepository(session), + sync_repository=SessionSubscriptionRepository(db.session), + unit_of_work=SqlAlchemyAsyncUnitOfWork(session), + ) + return await service.delete(subscribe_id) + + assert db.run_async_session(delete) + db.session.expire_all() + assert db.session.get(Subscribe, subscribe_id) is None + + +def test_transactional_repository_supports_servarr_async_contract(db) -> None: + """独立订阅仓储应提供 Servarr 查重与删除所需的异步事务能力。""" + row = db.add( + Subscribe( + name="独立 Servarr 订阅", + type=MediaType.TV.value, + media_source=MediaSource.TMDB.value, + media_id="transactional-servarr", + season=1, + ) + ) + subscribe_id = row.id + repository = TransactionalSubscriptionRepository( + sync_session=SessionFactory, + async_session=async_session_scope, + ) + + async def execute(_session: AsyncSession) -> bool: + """执行独立仓储的查重与删除操作。""" + assert await repository.async_exists( + media_source=MediaSource.TMDB, + media_id="transactional-servarr", + season=1, + ) is not None + await repository.async_delete(subscribe_id) + return True + + assert db.run_async_session(execute) + db.session.expire_all() + assert db.session.get(Subscribe, subscribe_id) is None + + def test_session_repository_leaves_commit_and_rollback_to_caller(db) -> None: """请求级仓储只 flush 暂存记录，调用方回滚后数据库不保留该记录。""" db.watermark(Subscribe)