fix(servarr): restore subscription repository async contract

2026-11-02 05:17:23 +08:00 · 2026-09-18 22:13:33 +08:00
parent 2e2a037eb8
commit 67ae166a54
5 changed files with 209 additions and 4 deletions
--- 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),
    )
--- 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
--- 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,
--- 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()
--- 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)