diff --git a/app/api/context.py b/app/api/context.py index d6510dbf5..d37363da3 100644 --- a/app/api/context.py +++ b/app/api/context.py @@ -13,6 +13,7 @@ from app.application.messaging.chat import AsyncAgentChatRepository, AsyncUnitOf from app.application.outbox import AsyncOutboxDispatchStore, AsyncOutboxStager from app.application.subscription.contract import ( SubscriptionHistoryStagingPort, + SubscriptionQueryPort, SubscriptionStagingPort, ) from app.runtime.tasks import TaskRegistry, get_task_registry @@ -136,6 +137,14 @@ def get_subscription_repository( return runtime.repository(session) +def get_sync_subscription_repository( + session: object = Depends(get_sync_session), + runtime: SubscriptionRuntime = Depends(get_subscription_runtime), +) -> SubscriptionQueryPort: + """构造绑定当前请求同步 Session 的订阅查询仓储。""" + return cast(SubscriptionQueryPort, runtime.repository(session)) + + def get_subscription_history_repository( session: object = Depends(get_subscription_session), runtime: SubscriptionRuntime = Depends(get_subscription_runtime), diff --git a/app/api/endpoints/subscribe.py b/app/api/endpoints/subscribe.py index 724cbe613..0cfa0c82b 100644 --- a/app/api/endpoints/subscribe.py +++ b/app/api/endpoints/subscribe.py @@ -11,7 +11,7 @@ from app.adapters.web.security.access import ( ) from app.api.context import ( get_background_task_registry, - get_subscription_repository, + get_sync_subscription_repository, resolve_background_task_registry, ) from app.api.dependencies.auth import ( @@ -746,10 +746,10 @@ async def user_subscribes( summary="订阅相关文件信息", response_model=_SchemaSubscrbieInfo, ) -async def subscribe_files( +def subscribe_files( subscribe_id: int, - repository: SubscriptionQueryPort = Depends(get_subscription_repository), - current_user: ApiPrincipal = Depends(get_current_active_user_async), + repository: SubscriptionQueryPort = Depends(get_sync_subscription_repository), + current_user: ApiPrincipal = Depends(get_current_active_user), ) -> Any: """ 订阅相关文件信息 @@ -758,12 +758,9 @@ async def subscribe_files( name=current_user.name, is_superuser=current_user.is_superuser, ) - subscribe = await repository.async_get(subscribe_id) + subscribe = repository.get(subscribe_id) if subscribe is not None and SubscriptionMutationService.can_access(subscribe, actor): - return await run_in_threadpool( - SubscribeChain().subscribe_files_info, - subscribe, - ) + return SubscribeChain().subscribe_files_info(subscribe) return _SchemaSubscrbieInfo() diff --git a/tests/fixtures/architecture/concurrency-baseline.json b/tests/fixtures/architecture/concurrency-baseline.json index ed69506cc..cde0ca2e0 100644 --- a/tests/fixtures/architecture/concurrency-baseline.json +++ b/tests/fixtures/architecture/concurrency-baseline.json @@ -221,8 +221,7 @@ }, "app/api/endpoints/subscribe.py:app.runtime.execution.run_in_threadpool": { "owners": { - "seerr_subscribe": 1, - "subscribe_files": 1 + "seerr_subscribe": 1 }, "target": "app.runtime.execution.run_in_threadpool" }, diff --git a/tests/test_subscribe_endpoint.py b/tests/test_subscribe_endpoint.py index 89e8a1d4b..88899ba0f 100644 --- a/tests/test_subscribe_endpoint.py +++ b/tests/test_subscribe_endpoint.py @@ -678,47 +678,33 @@ class TestSubscribeEndpoint: repository = _SubscriptionRepositoryFake(other) with patch("app.api.endpoints.subscribe.SubscribeChain") as subscribe_chain: - result = asyncio.run( - subscribe_files( - subscribe_id=19, - repository=repository, - current_user=_EndpointUser(name="alice", is_superuser=False), - ) + result = subscribe_files( + subscribe_id=19, + repository=repository, + current_user=_EndpointUser(name="alice", is_superuser=False), ) assert result.episodes == {} - repository.async_get.assert_awaited_once_with(19) subscribe_chain.return_value.subscribe_files_info.assert_not_called() - def test_subscribe_files_uses_async_query_and_offloads_file_statistics(self): - """文件统计先异步读取稳定快照，再在线程池执行同步媒体查询。""" + def test_subscribe_files_uses_sync_query_repository(self): + """文件统计使用绑定同步 Session 的查询仓储读取稳定快照。""" from app.api.endpoints.subscribe import subscribe_files own = _EndpointSubscribe(id=21, username="alice", name="自己的订阅") repository = _SubscriptionRepositoryFake(own) expected = SimpleNamespace(episodes={1: SimpleNamespace()}) - with ( - patch("app.api.endpoints.subscribe.SubscribeChain") as subscribe_chain, - patch( - "app.api.endpoints.subscribe.run_in_threadpool", - new=AsyncMock(return_value=expected), - ) as run_worker, - ): - result = asyncio.run( - subscribe_files( - subscribe_id=21, - repository=repository, - current_user=_EndpointUser(name="alice", is_superuser=False), - ) + with patch("app.api.endpoints.subscribe.SubscribeChain") as subscribe_chain: + subscribe_chain.return_value.subscribe_files_info.return_value = expected + result = subscribe_files( + subscribe_id=21, + repository=repository, + current_user=_EndpointUser(name="alice", is_superuser=False), ) assert result is expected - repository.async_get.assert_awaited_once_with(21) - run_worker.assert_awaited_once_with( - subscribe_chain.return_value.subscribe_files_info, - own, - ) + subscribe_chain.return_value.subscribe_files_info.assert_called_once_with(own) def test_user_subscribes_hides_other_user_list(self): """ diff --git a/tests/test_subscription_repository.py b/tests/test_subscription_repository.py index 10dd98547..0fccb2431 100644 --- a/tests/test_subscription_repository.py +++ b/tests/test_subscription_repository.py @@ -1,9 +1,12 @@ """订阅仓储的短 Session 投影与请求事务所有权测试。""" +from types import SimpleNamespace + import pytest from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession +from app.api.context import get_sync_subscription_repository from app.application.subscription.complete import CompleteSubscriptionCommand from app.application.subscription.contract import ( SubscriptionIdentity, @@ -72,6 +75,15 @@ def test_transactional_repository_returns_frozen_detached_snapshot(db) -> None: snapshot.episode_priority["1"] = 100 # type: ignore[index] +def test_sync_api_repository_dependency_binds_sync_session(db) -> None: + """文件统计依赖必须把同步请求 Session 交给订阅仓储。""" + runtime = SimpleNamespace(repository=SessionSubscriptionRepository) + + repository = get_sync_subscription_repository(db.session, runtime) + + assert repository.get(-1) is None + + def test_session_repository_leaves_commit_and_rollback_to_caller(db) -> None: """请求级仓储只 flush 暂存记录，调用方回滚后数据库不保留该记录。""" db.watermark(Subscribe)