From d8b8583d499552ff77755f27916264a822acccc2 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sun, 23 Aug 2026 14:15:37 +0800 Subject: [PATCH] refactor: isolate media server and site userdata queries --- app/db/models/mediaserver.py | 16 ++--- app/db/models/siteuserdata.py | 12 ++-- app/db/oper/mediaserver.py | 66 ++++++++++++++----- app/db/oper/site.py | 40 ++++++++--- docs/architecture-overview.md | 4 +- .../backend-architecture-next-stage.md | 10 ++- docs/rules/10-data-and-persistent.md | 2 +- .../transaction-debt-baseline.json | 66 +------------------ tests/test_architecture_contract_baseline.py | 4 +- tests/test_db_mediaserver_queries.py | 49 ++++++++++++++ tests/test_db_site_queries.py | 43 ++++++++++++ 11 files changed, 202 insertions(+), 110 deletions(-) diff --git a/app/db/models/mediaserver.py b/app/db/models/mediaserver.py index 6e85b47a5..26e8965aa 100644 --- a/app/db/models/mediaserver.py +++ b/app/db/models/mediaserver.py @@ -7,7 +7,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import Mapped, Session, mapped_column from app.db.base import Base, execute_dml, get_id_column -from app.db.decorators import async_db_query, db_query +from app.db.decorators import legacy_async_db_query, legacy_db_query from app.db.models._constraints import media_identity_constraint from app.schemas.types import MediaSource @@ -53,12 +53,12 @@ class MediaServerItem(Base): ) @classmethod - @db_query + @legacy_db_query def get_by_itemid(cls, db: Session, item_id: str): return db.execute(select(cls).where(cls.item_id == item_id)).scalars().first() @classmethod - @db_query + @legacy_db_query def get_by_server_itemid(cls, db: Session, server: str, item_id: str): return db.execute( select(cls).where(cls.server == server, cls.item_id == item_id) @@ -97,7 +97,7 @@ class MediaServerItem(Base): ) @classmethod - @db_query + @legacy_db_query def exist_by_media_identity( cls, db: Session, media_source: MediaSource, media_id: str, mtype: str, ): @@ -109,7 +109,7 @@ class MediaServerItem(Base): )).scalars().first() @classmethod - @db_query + @legacy_db_query def exists_by_title(cls, db: Session, title: str, mtype: str, year: str): statement = select(cls).where(cls.title == title) if mtype: @@ -119,13 +119,13 @@ class MediaServerItem(Base): return db.execute(statement).scalars().first() @classmethod - @async_db_query + @legacy_async_db_query async def async_get_by_itemid(cls, db: AsyncSession, item_id: str): result = await db.execute(select(cls).filter(cls.item_id == item_id)) return result.scalars().first() @classmethod - @async_db_query + @legacy_async_db_query async def async_exist_by_media_identity( cls, db: AsyncSession, media_source: MediaSource, media_id: str, mtype: str, ): @@ -138,7 +138,7 @@ class MediaServerItem(Base): return result.scalars().first() @classmethod - @async_db_query + @legacy_async_db_query async def async_exists_by_title(cls, db: AsyncSession, title: str, mtype: str, year: str): if not mtype and not year: result = await db.execute(select(cls).filter(cls.title == title)) diff --git a/app/db/models/siteuserdata.py b/app/db/models/siteuserdata.py index 0823b594c..452e8d4c3 100644 --- a/app/db/models/siteuserdata.py +++ b/app/db/models/siteuserdata.py @@ -6,7 +6,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import Mapped, Session, mapped_column from app.db.base import Base, execute_dml, get_id_column -from app.db.decorators import async_db_query, db_query +from app.db.decorators import legacy_async_db_query, legacy_db_query class SiteUserData(Base): @@ -61,7 +61,7 @@ class SiteUserData(Base): ) @classmethod - @db_query + @legacy_db_query def get_by_domain(cls, db: Session, domain: str, workdate: Optional[str] = None, worktime: Optional[str] = None): statement = select(cls).where(cls.domain == domain) if workdate and worktime: @@ -72,7 +72,7 @@ class SiteUserData(Base): return list(db.execute(statement).scalars().all()) @classmethod - @async_db_query + @legacy_async_db_query async def async_get_by_domain(cls, db: AsyncSession, domain: str, workdate: Optional[str] = None, worktime: Optional[str] = None): query = select(cls).filter(cls.domain == domain) if workdate and worktime: @@ -83,12 +83,12 @@ class SiteUserData(Base): return list(result.scalars().all()) @classmethod - @db_query + @legacy_db_query def get_by_date(cls, db: Session, date: str): return list(db.execute(select(cls).where(cls.updated_day == date)).scalars().all()) @classmethod - @db_query + @legacy_db_query def get_latest(cls, db: Session): """ 获取各站点最新一天的数据 @@ -113,7 +113,7 @@ class SiteUserData(Base): ).scalars().all()) @classmethod - @async_db_query + @legacy_async_db_query async def async_get_latest(cls, db: AsyncSession): """ 异步获取各站点最新一天的数据 diff --git a/app/db/oper/mediaserver.py b/app/db/oper/mediaserver.py index a33bfe6d4..72ab46c6a 100644 --- a/app/db/oper/mediaserver.py +++ b/app/db/oper/mediaserver.py @@ -1,5 +1,6 @@ -from typing import Optional +from typing import Optional, Union +from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import Session from app.db.base import DbOper @@ -11,7 +12,11 @@ class MediaServerOper(DbOper): 媒体服务器数据管理 """ - def __init__(self, db: Optional[Session] = None): + def __init__( + self, + db: Optional[Union[Session, AsyncSession]] = None, + ) -> None: + """保存调用方提供的同步或异步查询会话。""" super().__init__(db) @staticmethod @@ -34,7 +39,12 @@ class MediaServerOper(DbOper): if not server or not item_id: return False item = MediaServerItem(**kwargs) - if not item.get_by_server_itemid(self._db, server, item_id): + existing = self._execute_sync_query( + lambda session: MediaServerItem.get_by_server_itemid( + session, server, item_id + ) + ) + if not existing: self._stage_create(item) return True return False @@ -49,7 +59,11 @@ class MediaServerOper(DbOper): if not server or not item_id: return False - item = MediaServerItem.get_by_server_itemid(self._db, server, item_id) + item = self._execute_sync_query( + lambda session: MediaServerItem.get_by_server_itemid( + session, server, item_id + ) + ) if item: self._stage_update(item, kwargs) return False @@ -93,16 +107,24 @@ class MediaServerOper(DbOper): 判断媒体服务器数据是否存在 """ if kwargs.get("media_source") and kwargs.get("media_id"): - item = MediaServerItem.exist_by_media_identity( - self._db, - media_source=kwargs.get("media_source"), - media_id=kwargs.get("media_id"), - mtype=kwargs.get("mtype"), + item = self._execute_sync_query( + lambda session: MediaServerItem.exist_by_media_identity( + session, + media_source=kwargs.get("media_source"), + media_id=kwargs.get("media_id"), + mtype=kwargs.get("mtype"), + ) ) elif kwargs.get("title"): # 按标题、类型、年份查 - item = MediaServerItem.exists_by_title(self._db, title=kwargs.get("title"), - mtype=kwargs.get("mtype"), year=kwargs.get("year")) + item = self._execute_sync_query( + lambda session: MediaServerItem.exists_by_title( + session, + title=kwargs.get("title"), + mtype=kwargs.get("mtype"), + year=kwargs.get("year"), + ) + ) else: return None if not item: @@ -122,16 +144,24 @@ class MediaServerOper(DbOper): 异步判断媒体服务器数据是否存在 """ if kwargs.get("media_source") and kwargs.get("media_id"): - item = await MediaServerItem.async_exist_by_media_identity( - self._db, - media_source=kwargs.get("media_source"), - media_id=kwargs.get("media_id"), - mtype=kwargs.get("mtype"), + item = await self._execute_async_query( + lambda session: MediaServerItem.async_exist_by_media_identity( + session, + media_source=kwargs.get("media_source"), + media_id=kwargs.get("media_id"), + mtype=kwargs.get("mtype"), + ) ) elif kwargs.get("title"): # 按标题、类型、年份查 - item = await MediaServerItem.async_exists_by_title(self._db, title=kwargs.get("title"), - mtype=kwargs.get("mtype"), year=kwargs.get("year")) + item = await self._execute_async_query( + lambda session: MediaServerItem.async_exists_by_title( + session, + title=kwargs.get("title"), + mtype=kwargs.get("mtype"), + year=kwargs.get("year"), + ) + ) else: return None if not item: diff --git a/app/db/oper/site.py b/app/db/oper/site.py index 2790b79aa..e1be9263b 100644 --- a/app/db/oper/site.py +++ b/app/db/oper/site.py @@ -280,7 +280,13 @@ class SiteOper(DbOper): "err_msg": payload.get("err_msg") or "" }) # 按站点+天判断是否存在数据 - siteuserdatas = SiteUserData.get_by_domain(self._db, domain=domain, workdate=current_day) + siteuserdatas = self._execute_sync_query( + lambda session: SiteUserData.get_by_domain( + session, + domain=domain, + workdate=current_day, + ) + ) if siteuserdatas: # 存在则更新 if not payload.get("err_msg"): @@ -294,13 +300,21 @@ class SiteOper(DbOper): """ 获取站点用户数据 """ - return SiteUserData.list(self._db) + return self._execute_sync_query( + lambda session: SiteUserData.list(session) + ) def get_userdata_by_domain(self, domain: str, workdate: Optional[str] = None) -> List[SiteUserData]: """ 获取站点用户数据 """ - return SiteUserData.get_by_domain(self._db, domain=domain, workdate=workdate) + return self._execute_sync_query( + lambda session: SiteUserData.get_by_domain( + session, + domain=domain, + workdate=workdate, + ) + ) async def async_get_userdata_by_domain( self, domain: str, workdate: Optional[str] = None @@ -308,13 +322,19 @@ class SiteOper(DbOper): """ 异步获取站点用户数据。 """ - return await SiteUserData.async_get_by_domain( - self._db, domain=domain, workdate=workdate + return await self._execute_async_query( + lambda session: SiteUserData.async_get_by_domain( + session, + domain=domain, + workdate=workdate, + ) ) async def async_get_userdata_latest(self) -> List[SiteUserData]: """异步获取各站点最新用户数据。""" - return await SiteUserData.async_get_latest(self._db) + return await self._execute_async_query( + lambda session: SiteUserData.async_get_latest(session) + ) async def async_get_icon_by_domain(self, domain: str) -> Optional[SiteIcon]: """异步按域名获取站点图标。""" @@ -335,13 +355,17 @@ class SiteOper(DbOper): """ 获取站点用户数据 """ - return SiteUserData.get_by_date(self._db, date) + return self._execute_sync_query( + lambda session: SiteUserData.get_by_date(session, date) + ) def get_userdata_latest(self) -> List[SiteUserData]: """ 获取站点最新数据 """ - return SiteUserData.get_latest(self._db) + return self._execute_sync_query( + lambda session: SiteUserData.get_latest(session) + ) def get_icon_by_domain(self, domain: str) -> Optional[SiteIcon]: """ diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index 7baa7f930..72df34a93 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -378,9 +378,9 @@ flowchart LR 成功后执行。订阅新增样板由 `startup/subscription.py` 创建独占 Session， `application/subscription/write.py` 决定事务与 post-commit 边界，`SubscribeOper.stage_add()` 只查重、`add` 和 `flush`。旧 SDK 显式构造的无会话 Oper 暂留兼容自动短会话，不得被新代码复用。 - `transaction-debt-baseline.json` 当前冻结 30 个正式只读查询装饰器；原有同步/异步写装饰器 + `transaction-debt-baseline.json` 当前冻结 18 个正式只读查询装饰器；原有同步/异步写装饰器 已全部移除，`db_update` 与 `async_db_update` 必须持续保持为 0。下载/整理历史的旧插件 Model - 与工作流旧插件 Model 调用由 `legacy_*` 兼容外壳承接，宿主 Oper 必须显式传递 Session。宿主 Oper 也不得调用 Base 保留的 + 与工作流、媒体服务器、站点用户数据旧插件 Model 调用由 `legacy_*` 兼容外壳承接，宿主 Oper 必须显式传递 Session。宿主 Oper 也不得调用 Base 保留的 `create/update/delete/truncate` 兼容包装器；AST 门禁保证显式 Session 的提交权不会被底层抢走。 - 站点、历史、工作流、Agent 会话删除和插件数据重置已经形成同构事务切片；对应 Application Command/Service 持有 UoW，Oper 的 `stage_*` 方法只修改当前会话。插件数据重置从 diff --git a/docs/refactor/backend-architecture-next-stage.md b/docs/refactor/backend-architecture-next-stage.md index e1cd0fc52..636fd226c 100644 --- a/docs/refactor/backend-architecture-next-stage.md +++ b/docs/refactor/backend-architecture-next-stage.md @@ -28,7 +28,7 @@ 1. **后台任务的统一所有权已覆盖 API 入口，但仍有更深层任务机制待分级。** `app/runtime/tasks.py` 已建立 lifespan 级 TaskRegistry，启动收尾、插件 Release 刷新、Webhook E0 广播、CookieCloud E1 手工调度、消息入口、Seerr 订阅和 WebAgent 断线后执行/快照保存均不再维护端点模块级任务集合或 Starlette 回调，shutdown 会停止接收、取消并有限等待，且生命周期清单明确登记其顺序。主仓 `app/` 已无裸 FastAPI `BackgroundTasks`；当前仍有约 `50` 个更底层 `create_task`/等价任务创建点，与线程池和 APScheduler 并存，后续需逐项确认 owner、取消、等待、重试、幂等和是否 durable，关键业务副作用优先接入已有 Outbox/恢复表。 2. **动态模块契约仍以 legacy 聚合语义为主。** 当前登记 `212` 个模块方法，其中 `194` 个仍使用 `legacy` aggregation，只有 `14` 个 `first_non_empty`、`4` 个 `ordered_list_merge`。`app/runtime/extensions/module/contracts.py:422-455` 已能登记 family、输入/结果标签和基础签名诊断，但 `193` 个方法没有 required parameters，调度器 `app/runtime/extensions/module/dispatcher.py:109-260` 仍主要依赖运行时反射、返回值形状和短路规则。未知第三方方法保留 legacy fallback 是兼容要求，不应删除；宿主高频能力则应逐族补齐可执行的输入校验、结果校验、超时和错误语义。 -3. **查询侧数据库兼容 ABI 仍未完全收口。** 写事务装饰器已降为 `0`，正式 `db_query/async_db_query` 已降至 `30` 个（`16` 个同步、`14` 个异步）。站点、消息、用户、订阅、下载/整理历史和工作流的宿主 Oper 已迁到显式 Session 路径；对应旧插件 Model 调用由独立 `legacy_*` 外壳保留，可同时接受显式 Session 与无 Session 的位置/关键字参数。剩余正式装饰器仍会隐式创建会话，查询返回的 ORM 对象也可能跨层流转，后续继续按 MediaServer、SiteUserData 等风险切片迁移。 +3. **查询侧数据库兼容 ABI 仍未完全收口。** 写事务装饰器已降为 `0`，正式 `db_query/async_db_query` 已降至 `18` 个（`9` 个同步、`9` 个异步）。站点、消息、用户、订阅、下载/整理历史、工作流、MediaServer 和 SiteUserData 的宿主 Oper 已迁到显式 Session 路径；对应旧插件 Model 调用由独立 `legacy_*` 外壳保留，可同时接受显式 Session 与无 Session 的位置/关键字参数。剩余正式装饰器仍会隐式创建会话，查询返回的 ORM 对象也可能跨层流转，后续继续按 AgentChat、AgentTaskRun、TransferPending、SystemConfig、PassKey 和 SubscribeHistory 等风险切片迁移。 4. **组合根和全局状态仍形成复杂的隐式运行时图。** Singleton 实例、模块级 provider、`configure_*` 注册函数和兼容 Facade 同时存在；它们解决了旧 ABI 和启动顺序问题，但增加测试污染、重复装配、实例身份和初始化顺序风险。`app/startup/lifecycle/__init__.py:161-376` 已有声明式生命周期，`app/startup/modules_initializer.py:505-530` 也有分阶段关闭，但尚未做到所有进程级资源都只通过 typed HostRuntime 访问。后续应以“新代码禁止新增 Service Locator/Singleton 依赖、旧入口有命中观测”为 ratchet。 ### P2：中长期可演进性债务 @@ -1188,6 +1188,12 @@ host/plugin 架构基线均通过。 30 个且写装饰器保持 0。旧插件仍可直接调用 Workflow Model 方法，显式 Session 与无 Session 的 关键字调用均有回归覆盖；Workflow、架构基线专项共 `76 passed`，host/plugin 架构基线通过。 +2026-08-23 完成 MediaServer 与 SiteUserData 查询切片：`MediaServerOper`、`SiteOper` 的同步/异步 +查询入口统一通过 `_execute_sync_query` / `_execute_async_query` 复用调用方 Session，正式查询装饰器 +由 30 降至 18 个（同步 9、异步 9），写装饰器保持 0。旧插件仍可直接调用对应 Model 方法，显式 +Session 不创建额外会话，无 Session 的位置参数和关键字参数继续由 `legacy_*` 外壳兼容；专项测试 +`158 passed`，四分片全量测试 `5539 passed, 3 skipped`，host/plugin 架构基线和 Pylint 均通过。 + #### ARCH-272：异步阻塞检测 **目标**：对新 API/Agent/Application async 路径检测 `open`、文件遍历、同步 HTTP、阻塞 sleep 和重 CPU 解析。 @@ -1365,7 +1371,7 @@ rollback: | 基线写入行为 | 默认命令可能覆盖 fixture | 所有默认/check 命令保证工作树不变；write 必须显式 scope | | 全功能 worker | 配置允许 >1，控制面会复制 | 启动期明确拒绝 >1；文档与配置一致 | | 健康接口 | 认证 `/system/ping` 为主 | 分离公开 live 与受限/安全 ready；失败原因可诊断 | -| Model 事务装饰器 | 当前 30 个且全部只读；写装饰器 0 | 查询债务只降不增；写事务不回退到 Model/Base 隐式提交 | +| Model 事务装饰器 | 当前 18 个且全部只读；写装饰器 0 | 查询债务只降不增；写事务不回退到 Model/Base 隐式提交 | | 新写用例事务 | 宿主写 Oper 已脱离 Base 隐式提交 | 100% 由入口/Application 边界拥有 Session/UoW | | 高频 Module 契约 | 212 个宿主能力显式登记 | 新观察到的宿主方法必须同步登记完整契约 | | Event payload | 53 类型全部登记 typed payload 与可靠性 | 新事件必须同步登记，不回退裸 dict | diff --git a/docs/rules/10-data-and-persistent.md b/docs/rules/10-data-and-persistent.md index d27d416a8..aa63b39ba 100644 --- a/docs/rules/10-data-and-persistent.md +++ b/docs/rules/10-data-and-persistent.md @@ -84,7 +84,7 @@ Oper classes accept and return persistence values. Turning a `MediaInfo` or ### Transaction ownership ratchet - `tests/fixtures/architecture/transaction-debt-baseline.json` records the - existing Model transaction decorators. The current 30 decorators are query-only + existing Model transaction decorators. The current 18 decorators are query-only migration debt: they may decrease but must never increase or move to a new Model method. Both `db_update` and `async_db_update` must remain at zero. - `legacy_db_query` / `legacy_async_db_query` are compatibility-only shells for diff --git a/tests/fixtures/architecture/transaction-debt-baseline.json b/tests/fixtures/architecture/transaction-debt-baseline.json index 40f9661b1..1353d6aa4 100644 --- a/tests/fixtures/architecture/transaction-debt-baseline.json +++ b/tests/fixtures/architecture/transaction-debt-baseline.json @@ -1,12 +1,12 @@ { "model_decorators": { "by_kind": { - "async_db_query": 14, + "async_db_query": 9, "async_db_update": 0, - "db_query": 16, + "db_query": 9, "db_update": 0 }, - "count": 30, + "count": 18, "methods": [ { "decorator": "async_db_query", @@ -38,41 +38,6 @@ "file": "app/db/models/agenttaskrun.py", "method": "AgentTaskRun.list_for_task" }, - { - "decorator": "async_db_query", - "file": "app/db/models/mediaserver.py", - "method": "MediaServerItem.async_exist_by_media_identity" - }, - { - "decorator": "async_db_query", - "file": "app/db/models/mediaserver.py", - "method": "MediaServerItem.async_exists_by_title" - }, - { - "decorator": "async_db_query", - "file": "app/db/models/mediaserver.py", - "method": "MediaServerItem.async_get_by_itemid" - }, - { - "decorator": "db_query", - "file": "app/db/models/mediaserver.py", - "method": "MediaServerItem.exist_by_media_identity" - }, - { - "decorator": "db_query", - "file": "app/db/models/mediaserver.py", - "method": "MediaServerItem.exists_by_title" - }, - { - "decorator": "db_query", - "file": "app/db/models/mediaserver.py", - "method": "MediaServerItem.get_by_itemid" - }, - { - "decorator": "db_query", - "file": "app/db/models/mediaserver.py", - "method": "MediaServerItem.get_by_server_itemid" - }, { "decorator": "async_db_query", "file": "app/db/models/passkey.py", @@ -93,31 +58,6 @@ "file": "app/db/models/passkey.py", "method": "PassKey.get_by_id" }, - { - "decorator": "async_db_query", - "file": "app/db/models/siteuserdata.py", - "method": "SiteUserData.async_get_by_domain" - }, - { - "decorator": "async_db_query", - "file": "app/db/models/siteuserdata.py", - "method": "SiteUserData.async_get_latest" - }, - { - "decorator": "db_query", - "file": "app/db/models/siteuserdata.py", - "method": "SiteUserData.get_by_date" - }, - { - "decorator": "db_query", - "file": "app/db/models/siteuserdata.py", - "method": "SiteUserData.get_by_domain" - }, - { - "decorator": "db_query", - "file": "app/db/models/siteuserdata.py", - "method": "SiteUserData.get_latest" - }, { "decorator": "async_db_query", "file": "app/db/models/subscribehistory.py", diff --git a/tests/test_architecture_contract_baseline.py b/tests/test_architecture_contract_baseline.py index 7e0c204a4..388322288 100644 --- a/tests/test_architecture_contract_baseline.py +++ b/tests/test_architecture_contract_baseline.py @@ -126,8 +126,8 @@ def test_transaction_debt_baseline_is_a_model_and_oper_ratchet() -> None: baseline = json.loads(baseline_path.read_text(encoding="utf-8")) assert baseline["schema_version"] == 1 - assert baseline["model_decorators"]["count"] == 30 - assert sum(baseline["model_decorators"]["by_kind"].values()) == 30 + assert baseline["model_decorators"]["count"] == 18 + assert sum(baseline["model_decorators"]["by_kind"].values()) == 18 assert baseline["model_decorators"]["by_kind"]["db_update"] == 0 assert baseline["model_decorators"]["by_kind"]["async_db_update"] == 0 assert baseline["model_transaction_calls"] == {"count": 0, "calls": []} diff --git a/tests/test_db_mediaserver_queries.py b/tests/test_db_mediaserver_queries.py index bb2d84c00..22731f20e 100644 --- a/tests/test_db_mediaserver_queries.py +++ b/tests/test_db_mediaserver_queries.py @@ -9,7 +9,10 @@ import asyncio import pytest +from app.db import decorators from app.db.models.mediaserver import MediaServerItem +from app.db.oper.mediaserver import MediaServerOper +from app.db.session import SessionFactory, async_session_scope from app.schemas.types import MediaSource @@ -40,6 +43,52 @@ def test_get_by_itemid_matches_async_twin(db): assert MediaServerItem.get_by_itemid(db.session, "it-missing") is None +def test_mediaserver_oper_reuses_explicit_query_sessions(db, monkeypatch): + """媒体服务器 Oper 绑定调用方会话后不得再创建兼容查询会话。""" + db.add(_item("emby", "explicit-ms", media_id="explicit-1001")) + monkeypatch.setattr( + decorators, + "ScopedSession", + lambda: (_ for _ in ()).throw(AssertionError("不应创建额外同步会话")), + ) + + assert MediaServerOper(db.session).exists( + media_source=MediaSource.TMDB, + media_id="explicit-1001", + mtype="电影", + ) is not None + + async def check() -> None: + """验证异步存在性查询复用显式 AsyncSession。""" + async with async_session_scope() as session: + monkeypatch.setattr( + decorators, + "async_session_scope", + lambda: (_ for _ in ()).throw(AssertionError("不应创建额外异步会话")), + ) + assert await MediaServerOper(session).async_exists( + media_source=MediaSource.TMDB, + media_id="explicit-1001", + mtype="电影", + ) is not None + + asyncio.run(check()) + + +def test_mediaserver_model_legacy_query_keeps_keyword_abi(db, monkeypatch): + """旧插件以关键字直调媒体服务器 Model 时仍自动补入短会话。""" + db.add(_item("emby", "legacy-ms")) + opened = [] + monkeypatch.setattr( + decorators, + "ScopedSession", + lambda: (opened.append(True) or SessionFactory()), + ) + + assert MediaServerItem.get_by_itemid(item_id="legacy-ms") is not None + assert opened == [True] + + def test_get_by_server_itemid_scopes_by_server(db): """ 条目 ID 只在单个服务器内唯一，查找必须同时限定服务器。 diff --git a/tests/test_db_site_queries.py b/tests/test_db_site_queries.py index c5d263fc2..c3ab7fbda 100644 --- a/tests/test_db_site_queries.py +++ b/tests/test_db_site_queries.py @@ -9,10 +9,13 @@ import asyncio import pytest +from app.db import decorators from app.db.models.site import Site from app.db.models.siteicon import SiteIcon from app.db.models.sitestatistic import SiteStatistic from app.db.models.siteuserdata import SiteUserData +from app.db.oper.site import SiteOper +from app.db.session import SessionFactory, async_session_scope @pytest.fixture(autouse=True) @@ -187,6 +190,46 @@ def test_userdata_get_by_domain_matches_async_twin(db): assert len(sync_rows) == len(async_rows) +def test_site_oper_reuses_explicit_userdata_query_sessions(db, monkeypatch): + """站点用户数据 Oper 必须复用调用方同步与异步会话。""" + db.add(_userdata("explicit-site.test", "2026-08-12", "10:00:00")) + monkeypatch.setattr( + decorators, + "ScopedSession", + lambda: (_ for _ in ()).throw(AssertionError("不应创建额外同步会话")), + ) + + assert SiteOper(db.session).get_userdata_by_domain("explicit-site.test") + + async def check() -> None: + """验证异步站点用户数据查询复用显式 AsyncSession。""" + async with async_session_scope() as session: + monkeypatch.setattr( + decorators, + "async_session_scope", + lambda: (_ for _ in ()).throw(AssertionError("不应创建额外异步会话")), + ) + assert await SiteOper(session).async_get_userdata_by_domain( + "explicit-site.test" + ) + + asyncio.run(check()) + + +def test_site_userdata_model_legacy_query_keeps_keyword_abi(db, monkeypatch): + """旧插件以关键字直调 SiteUserData 时仍自动补入短会话。""" + db.add(_userdata("legacy-site.test", "2026-08-12", "10:00:00")) + opened = [] + monkeypatch.setattr( + decorators, + "ScopedSession", + lambda: (opened.append(True) or SessionFactory()), + ) + + assert SiteUserData.get_by_domain(domain="legacy-site.test") + assert opened == [True] + + def test_userdata_get_by_date_returns_all_domains_of_that_day(db): """ 按日期查询应跨站点返回当天全部快照。