refactor: migrate passkey host queries

app/db/models/passkey.py

@@ -8,6 +8,21 @@ from app.db.base import Base, get_id_column
 from app.db.decorators import db_query, async_db_query
 
 
+def _get_by_user_id_statement(model: type["PassKey"], user_id: int):
+    """构造按用户筛选启用 PassKey 的查询语句。"""
+    return select(model).where(model.user_id == user_id, model.is_active.is_(True))
+
+
+def _get_by_credential_id_statement(
+        model: type["PassKey"], credential_id: str,
+):
+    """构造按凭证 ID 筛选启用 PassKey 的查询语句。"""
+    return select(model).where(
+        model.credential_id == credential_id,
+        model.is_active.is_(True),
+    )
+
+
 class PassKey(Base):
     """
     用户PassKey凭证表
@@ -36,11 +51,10 @@ class PassKey(Base):
     transports: Mapped[Optional[str]] = mapped_column(String, nullable=True)
 
     @classmethod
-    @db_query
     def get_by_user_id(cls, db: Session, user_id: int):
         """获取用户的所有PassKey"""
         return list(db.execute(
-            select(cls).where(cls.user_id == user_id, cls.is_active.is_(True))
+            _get_by_user_id_statement(cls, user_id)
         ).scalars().all())
 
     @classmethod
@@ -48,16 +62,15 @@ class PassKey(Base):
     async def async_get_by_user_id(cls, db: AsyncSession, user_id: int):
         """异步获取用户的所有PassKey"""
         result = await db.execute(
-            select(cls).filter(cls.user_id == user_id, cls.is_active.is_(True))
+            _get_by_user_id_statement(cls, user_id)
         )
         return list(result.scalars().all())
 
     @classmethod
-    @db_query
     def get_by_credential_id(cls, db: Session, credential_id: str):
         """根据凭证ID获取PassKey"""
         return db.execute(
-            select(cls).where(cls.credential_id == credential_id, cls.is_active.is_(True))
+            _get_by_credential_id_statement(cls, credential_id)
         ).scalars().first()
 
     @classmethod
@@ -65,7 +78,7 @@ class PassKey(Base):
     async def async_get_by_credential_id(cls, db: AsyncSession, credential_id: str):
         """异步根据凭证ID获取PassKey"""
         result = await db.execute(
-            select(cls).filter(cls.credential_id == credential_id, cls.is_active.is_(True))
+            _get_by_credential_id_statement(cls, credential_id)
         )
         return result.scalars().first()
 

app/db/oper/passkey.py

@@ -2,8 +2,15 @@
 
 from typing import Any, Optional
 
+from sqlalchemy.orm import Session
+
 from app.db.base import DbOper
-from app.db.models.passkey import PassKey
+from app.db.models.passkey import (
+    PassKey,
+    _get_by_credential_id_statement,
+    _get_by_user_id_statement,
+)
+from app.db.uow import run_sync_transaction
 
 
 class PassKeyOper(DbOper):
@@ -11,7 +18,15 @@ class PassKeyOper(DbOper):
 
     def list_by_user_id(self, user_id: int) -> list[PassKey]:
         """读取用户启用的 PassKey。"""
-        return PassKey.get_by_user_id(self._db, user_id)
+        def query(session: Session) -> list[PassKey]:
+            """在调用方会话中读取用户启用的 PassKey。"""
+            return list(session.execute(
+                _get_by_user_id_statement(PassKey, user_id)
+            ).scalars().all())
+
+        if isinstance(self._db, Session):
+            return query(self._db)
+        return run_sync_transaction(query)
 
     def list(self) -> list[PassKey]:
         """读取全部 PassKey，用于判断系统是否已配置通行密钥。"""
@@ -19,7 +34,15 @@ class PassKeyOper(DbOper):
 
     def get_by_credential_id(self, credential_id: str) -> Optional[PassKey]:
         """按凭证 ID 读取启用的 PassKey。"""
-        return PassKey.get_by_credential_id(self._db, credential_id)
+        def query(session: Session) -> Optional[PassKey]:
+            """在调用方会话中按凭证 ID 读取启用的 PassKey。"""
+            return session.execute(
+                _get_by_credential_id_statement(PassKey, credential_id)
+            ).scalars().first()
+
+        if isinstance(self._db, Session):
+            return query(self._db)
+        return run_sync_transaction(query)
 
     def create(self, payload: dict[str, Any]) -> PassKey:
         """创建 PassKey 凭证。"""

tests/fixtures/architecture/transaction-debt-baseline.json

@@ -3,10 +3,10 @@
     "by_kind": {
       "async_db_query": 49,
       "async_db_update": 0,
-      "db_query": 72,
+      "db_query": 70,
       "db_update": 0
     },
-    "count": 121,
+    "count": 119,
     "methods": [
       {
         "decorator": "async_db_query",
@@ -188,21 +188,11 @@
         "file": "app/db/models/passkey.py",
         "method": "PassKey.async_get_by_user_id"
       },
-      {
-        "decorator": "db_query",
-        "file": "app/db/models/passkey.py",
-        "method": "PassKey.get_by_credential_id"
-      },
       {
         "decorator": "db_query",
         "file": "app/db/models/passkey.py",
         "method": "PassKey.get_by_id"
       },
-      {
-        "decorator": "db_query",
-        "file": "app/db/models/passkey.py",
-        "method": "PassKey.get_by_user_id"
-      },
       {
         "decorator": "async_db_query",
         "file": "app/db/models/plugindata.py",

tests/test_db_config_user_queries.py

@@ -212,6 +212,21 @@ def test_passkey_listing_excludes_inactive_credentials(db):
         {"cred-active-1", "cred-active-2"}
 
 
+def test_passkey_oper_queries_use_explicit_session(db, monkeypatch):
+    """PassKeyOper 的宿主查询使用调用方 Session，不创建兼容事务。"""
+    db.add(_passkey(9002, "cred-oper"), _passkey(9002, "cred-oper-inactive", is_active=False))
+    monkeypatch.setattr(
+        "app.db.oper.passkey.run_sync_transaction",
+        lambda _query: pytest.fail("显式 Session 查询不应创建兼容事务"),
+    )
+
+    oper = PassKeyOper(db.session)
+
+    assert [item.credential_id for item in oper.list_by_user_id(9002)] == ["cred-oper"]
+    assert oper.get_by_credential_id("cred-oper").user_id == 9002
+    assert oper.get_by_credential_id("cred-oper-inactive") is None
+
+
 def test_passkey_lookup_by_credential_id_skips_inactive(db):
     """
     按凭据 ID 查找同样必须忽略停用记录，否则停用的密钥仍可完成认证。