refactor: migrate site read queries to explicit sessions

app/db/decorators.py

@@ -43,6 +43,14 @@ def run_legacy_sync_query(operation: Callable[[Session], _R]) -> _R:
         except Exception as close_err:  # noqa: BLE001 兼容查询释放失败不改变返回语义
             logger.error(f"释放数据库会话失败：{close_err}")
 
+
+async def run_legacy_async_query(
+    operation: Callable[[AsyncSession], Awaitable[_R]],
+) -> _R:
+    """为移除异步查询装饰器的旧 Model ABI 提供一次性异步会话。"""
+    async with async_session_scope() as db:
+        return await operation(db)
+
 def _get_args_db(
     args: tuple[Any, ...],
     kwargs: dict[str, Any],

app/db/models/siteicon.py

@@ -4,7 +4,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import Mapped, Session, mapped_column
 
 from app.db.base import Base, get_id_column
-from app.db.decorators import db_query, async_db_query
+from app.db.decorators import run_legacy_async_query
 
 
 class SiteIcon(Base):
@@ -22,12 +22,25 @@ class SiteIcon(Base):
     base64: Mapped[Optional[str]] = mapped_column(String)
 
     @classmethod
-    @db_query
     def get_by_domain(cls, db: Session, domain: str):
+        """在调用方 Session 中查询站点图标。"""
         return db.execute(select(cls).where(cls.domain == domain)).scalars().first()
 
     @classmethod
-    @async_db_query
-    async def async_get_by_domain(cls, db: AsyncSession, domain: str):
-        result = await db.execute(select(cls).where(cls.domain == domain))
-        return result.scalar_one_or_none()
+    async def async_get_by_domain(
+        cls,
+        db: AsyncSession | None = None,
+        domain: str | None = None,
+    ):
+        """在调用方 AsyncSession 中查询站点图标。"""
+        if domain is None:
+            raise TypeError("domain is required")
+
+        async def query(session: AsyncSession):
+            """在给定异步会话中执行站点图标查询。"""
+            result = await session.execute(select(cls).where(cls.domain == domain))
+            return result.scalar_one_or_none()
+
+        if isinstance(db, AsyncSession):
+            return await query(db)
+        return await run_legacy_async_query(query)

app/db/models/sitestatistic.py

@@ -6,7 +6,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import Mapped, Session, mapped_column
 
 from app.db.base import get_id_column, Base
-from app.db.decorators import db_query, async_db_query
+from app.db.decorators import run_legacy_async_query
 
 
 class SiteStatistic(Base):
@@ -30,15 +30,28 @@ class SiteStatistic(Base):
     note: Mapped[Optional[Any]] = mapped_column(JSON)
 
     @classmethod
-    @db_query
     def get_by_domain(cls, db: Session, domain: str):
+        """在调用方 Session 中查询站点统计。"""
         return db.execute(select(cls).where(cls.domain == domain)).scalars().first()
 
     @classmethod
-    @async_db_query
-    async def async_get_by_domain(cls, db: AsyncSession, domain: str):
-        result = await db.execute(select(cls).where(cls.domain == domain))
-        return result.scalar_one_or_none()
+    async def async_get_by_domain(
+        cls,
+        db: AsyncSession | None = None,
+        domain: str | None = None,
+    ):
+        """在调用方 AsyncSession 中查询站点统计，并兼容旧无会话调用。"""
+        if domain is None:
+            raise TypeError("domain is required")
+
+        async def query(session: AsyncSession):
+            """在给定异步会话中执行站点统计查询。"""
+            result = await session.execute(select(cls).where(cls.domain == domain))
+            return result.scalar_one_or_none()
+
+        if isinstance(db, AsyncSession):
+            return await query(db)
+        return await run_legacy_async_query(query)
 
     @classmethod
     def reset(cls, db: Session):

app/db/models/userconfig.py

@@ -3,7 +3,6 @@ from sqlalchemy import String, UniqueConstraint, JSON, select
 from sqlalchemy.orm import Mapped, Session, mapped_column
 
 from app.db.base import get_id_column, Base
-from app.db.decorators import db_query
 
 
 class UserConfig(Base):
@@ -24,8 +23,8 @@ class UserConfig(Base):
     )
 
     @classmethod
-    @db_query
     def get_by_key(cls, db: Session, username: str, key: str):
+        """在调用方 Session 中查询用户配置。"""
         return db.execute(
             select(cls).where(cls.username == username, cls.key == key)
         ).scalars().first()