From 5395bbff546c8115ff75bd55bb1f6ea13310b25e Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sun, 23 Aug 2026 01:08:01 +0800 Subject: [PATCH] fix: preserve legacy model query ABI --- app/db/decorators.py | 12 +++++ app/db/models/agenttask.py | 40 ++++++++++---- app/db/models/passkey.py | 52 +++++++++++++++---- .../backend-architecture-next-stage.md | 4 +- .../architecture/dependency-baseline.json | 5 +- tests/test_db_config_user_queries.py | 14 +++++ tests/test_db_plugin_message_agent_queries.py | 15 ++++++ 7 files changed, 117 insertions(+), 25 deletions(-) diff --git a/app/db/decorators.py b/app/db/decorators.py index 83b8a5941..15305805c 100644 --- a/app/db/decorators.py +++ b/app/db/decorators.py @@ -31,6 +31,18 @@ _R = TypeVar("_R") # 接管、返回值原样透传」。否则调用方传 None 或传异步会话都会被判成类型不符，而这恰恰是 # 装饰器存在的理由（各 Oper 的 self._db 常态就是 None）。 + +def run_legacy_sync_query(operation: Callable[[Session], _R]) -> _R: + """为已移除查询装饰器的旧 Model ABI 提供一次性同步会话。""" + db = ScopedSession() + try: + return operation(db) + finally: + try: + db.close() + except Exception as close_err: # noqa: BLE001 兼容查询释放失败不改变返回语义 + logger.error(f"释放数据库会话失败：{close_err}") + def _get_args_db( args: tuple[Any, ...], kwargs: dict[str, Any], diff --git a/app/db/models/agenttask.py b/app/db/models/agenttask.py index 920f1a1d2..947c91cf9 100644 --- a/app/db/models/agenttask.py +++ b/app/db/models/agenttask.py @@ -4,6 +4,7 @@ from sqlalchemy import Boolean, Index, Integer, String, Text, select, update from sqlalchemy.orm import Mapped, Session, mapped_column from app.db.base import Base, execute_dml, get_id_column +from app.db.decorators import run_legacy_sync_query def _get_for_user_statement( @@ -86,30 +87,47 @@ class AgentTask(Base): @classmethod def get_for_user( cls, - db: Session, - task_id: int, + db: Session | int | None = None, + task_id: int | None = None, user_id: Optional[str] = None, ) -> Optional["AgentTask"]: """ - 按任务 ID 和可选用户 ID 查询 Agent 定时任务。 + 按任务 ID 和可选用户 ID 查询，并保留无 Session 的旧插件调用方式。 """ - return db.execute( - _get_for_user_statement(cls, task_id=task_id, user_id=user_id) - ).scalars().first() + if task_id is None and isinstance(db, int): + task_id, db = db, None + if task_id is None: + raise TypeError("task_id is required") + + def query(session: Session) -> Optional["AgentTask"]: + """在给定会话中读取单个 Agent 任务。""" + return session.execute( + _get_for_user_statement(cls, task_id=task_id, user_id=user_id) + ).scalars().first() + + if isinstance(db, Session): + return query(db) + return run_legacy_sync_query(query) @classmethod def list_for_user( cls, - db: Session, + db: Session | None = None, user_id: Optional[str] = None, enabled: Optional[bool] = None, ) -> list["AgentTask"]: """ - 按用户和启用状态查询 Agent 定时任务。 + 按用户和启用状态查询，并保留无 Session 的旧插件调用方式。 """ - return list(db.execute( - _list_for_user_statement(cls, user_id=user_id, enabled=enabled) - ).scalars().all()) + def query(session: Session) -> list["AgentTask"]: + """在给定会话中读取 Agent 任务列表。""" + return list(session.execute( + _list_for_user_statement(cls, user_id=user_id, enabled=enabled) + ).scalars().all()) + + if isinstance(db, Session): + return query(db) + return run_legacy_sync_query(query) @classmethod def update_task( diff --git a/app/db/models/passkey.py b/app/db/models/passkey.py index 52aac0b6f..3df67158b 100644 --- a/app/db/models/passkey.py +++ b/app/db/models/passkey.py @@ -5,7 +5,7 @@ from sqlalchemy.orm import Mapped, Session, mapped_column from datetime import datetime from app.db.base import Base, get_id_column -from app.db.decorators import db_query, async_db_query +from app.db.decorators import async_db_query, db_query, run_legacy_sync_query def _get_by_user_id_statement(model: type["PassKey"], user_id: int): @@ -51,11 +51,26 @@ class PassKey(Base): transports: Mapped[Optional[str]] = mapped_column(String, nullable=True) @classmethod - def get_by_user_id(cls, db: Session, user_id: int): - """获取用户的所有PassKey""" - return list(db.execute( - _get_by_user_id_statement(cls, user_id) - ).scalars().all()) + def get_by_user_id( + cls, + db: Session | int | None = None, + user_id: int | None = None, + ): + """获取用户的所有 PassKey，并保留无 Session 的旧插件调用方式。""" + if user_id is None and isinstance(db, int): + user_id, db = db, None + if user_id is None: + raise TypeError("user_id is required") + + def query(session: Session): + """在给定会话中执行启用凭证查询。""" + return list(session.execute( + _get_by_user_id_statement(cls, user_id) + ).scalars().all()) + + if isinstance(db, Session): + return query(db) + return run_legacy_sync_query(query) @classmethod @async_db_query @@ -67,11 +82,26 @@ class PassKey(Base): return list(result.scalars().all()) @classmethod - def get_by_credential_id(cls, db: Session, credential_id: str): - """根据凭证ID获取PassKey""" - return db.execute( - _get_by_credential_id_statement(cls, credential_id) - ).scalars().first() + def get_by_credential_id( + cls, + db: Session | str | None = None, + credential_id: str | None = None, + ): + """按凭证 ID 获取 PassKey，并保留无 Session 的旧插件调用方式。""" + if credential_id is None and isinstance(db, str): + credential_id, db = db, None + if credential_id is None: + raise TypeError("credential_id is required") + + def query(session: Session): + """在给定会话中执行启用凭证查询。""" + return session.execute( + _get_by_credential_id_statement(cls, credential_id) + ).scalars().first() + + if isinstance(db, Session): + return query(db) + return run_legacy_sync_query(query) @classmethod @async_db_query diff --git a/docs/refactor/backend-architecture-next-stage.md b/docs/refactor/backend-architecture-next-stage.md index 2a6c22fc0..4b3159718 100644 --- a/docs/refactor/backend-architecture-next-stage.md +++ b/docs/refactor/backend-architecture-next-stage.md @@ -433,7 +433,9 @@ flowchart TB 无 Session 的旧 Oper 入口继续由组合根兼容事务执行器承接。随后 PassKey 的宿主同步查询迁移到 `PassKeyOper`，其按用户/凭证的启用状态过滤由显式 Session 测试覆盖；异步 Model 查询保留旧 ABI。 查询装饰器低水位由 123 降至 119，归属过滤、启用状态过滤和创建时间/主键稳定排序由 canonical - Oper 测试覆盖。 + Oper 测试覆盖。`PassKey.get_by_user_id/get_by_credential_id` 与 + `AgentTask.get_for_user/list_for_user` 同时保留旧插件省略 Session 的同步调用方式；该路径显式委托 + 一次性兼容查询会话，不重新增加 Model 查询装饰器，也不影响宿主显式 Session 的事务所有权。 #### ARCH-222：按风险迁移其余写用例 diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 9bae55c03..4d5dbf7f3 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -13,8 +13,8 @@ "runtime_to_db": [], "workflow_to_db": [] }, - "edge_count": 6413, - "edge_sha256": "02b7d1b32347dbd9be15ab2634c196dd36806fa9db89b0f8989ac6cc152c2b73", + "edge_count": 6414, + "edge_sha256": "aea2c4a5f65a9800bc69a83990efd368ba0ec7a9b19e40f38f8d0df6c222ae26", "edges": [ "app -> app.runtime", "app -> app.runtime.compat", @@ -3473,6 +3473,7 @@ "app.db.models.agentchat -> app.db.decorators", "app.db.models.agenttask -> app.db", "app.db.models.agenttask -> app.db.base", + "app.db.models.agenttask -> app.db.decorators", "app.db.models.agenttaskrun -> app.db", "app.db.models.agenttaskrun -> app.db.base", "app.db.models.agenttaskrun -> app.db.decorators", diff --git a/tests/test_db_config_user_queries.py b/tests/test_db_config_user_queries.py index 7c6118c5c..a7d4cc194 100644 --- a/tests/test_db_config_user_queries.py +++ b/tests/test_db_config_user_queries.py @@ -238,6 +238,20 @@ def test_passkey_lookup_by_credential_id_skips_inactive(db): assert asyncio.run(PassKey.async_get_by_credential_id(credential_id="cred-dead")) is None +def test_passkey_model_sync_queries_keep_no_session_plugin_abi(db, monkeypatch): + """旧插件不传 Session 时仍应获得短会话查询，而不恢复 Model 装饰器。""" + db.add(_passkey(9004, "cred-legacy")) + monkeypatch.setattr( + "app.db.models.passkey.run_legacy_sync_query", + lambda operation: operation(db.session), + ) + + assert [item.credential_id for item in PassKey.get_by_user_id(user_id=9004)] == [ + "cred-legacy" + ] + assert PassKey.get_by_credential_id("cred-legacy").user_id == 9004 + + def test_passkey_get_by_id_ignores_active_flag(db): """ 按主键取记录是管理用途，不应过滤停用状态——否则管理端看不到自己刚停用的凭据。 diff --git a/tests/test_db_plugin_message_agent_queries.py b/tests/test_db_plugin_message_agent_queries.py index 43e399cdc..fd8ade0a1 100644 --- a/tests/test_db_plugin_message_agent_queries.py +++ b/tests/test_db_plugin_message_agent_queries.py @@ -308,6 +308,21 @@ def test_agenttask_get_for_user_enforces_ownership(db): assert AgentTask.get_for_user(db.session, task_id, user_id="bob") is None +def test_agenttask_model_queries_keep_no_session_plugin_abi(db, monkeypatch): + """旧插件省略 Session 时仍可按原关键字参数查询 Agent 任务。""" + task_id = AgentTask.add_task(db.session, **_task("legacy", user_id="legacy-user")) + monkeypatch.setattr( + "app.db.models.agenttask.run_legacy_sync_query", + lambda operation: operation(db.session), + ) + + assert AgentTask.get_for_user( + task_id=task_id, + user_id="legacy-user", + ).id == task_id + assert [task.id for task in AgentTask.list_for_user(user_id="legacy-user")] == [task_id] + + def test_agenttask_oper_reads_with_explicit_session(db, monkeypatch): """AgentTaskOper 的宿主查询使用调用方 Session，不再经过旧事务兼容执行器。""" task_id = AgentTask.add_task(db.session, **_task("canonical", user_id="alice"))