fix(subscription): bind file statistics to sync session

This commit is contained in:
jxxghp
2026-09-02 12:02:59 +08:00
parent d94e74c241
commit 2767bb2e4f
5 changed files with 41 additions and 38 deletions

View File

@@ -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),

View File

@@ -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()

View File

@@ -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"
},

View File

@@ -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):
"""

View File

@@ -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)