fix: migrate database operations to async

Update BangumiDatabase, RSSDatabase, TorrentDatabase, and
UserDatabase to use AsyncSession.execute() instead of sync
Session.exec() to match the async engine migration.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: EstrellaXD

backend/src/module/database/bangumi.py

@@ -1,38 +1,54 @@
 import logging
+import time
 from typing import Optional
 
+from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.sql import func
-from sqlmodel import Session, and_, delete, false, or_, select
+from sqlmodel import and_, delete, false, or_, select
 
 from module.models import Bangumi, BangumiUpdate
 
 logger = logging.getLogger(__name__)
 
+# Module-level TTL cache for search_all results
+_bangumi_cache: list[Bangumi] | None = None
+_bangumi_cache_time: float = 0
+_BANGUMI_CACHE_TTL: float = 60.0  # seconds
+
+
+def _invalidate_bangumi_cache():
+    global _bangumi_cache, _bangumi_cache_time
+    _bangumi_cache = None
+    _bangumi_cache_time = 0
+
 
 class BangumiDatabase:
-    def __init__(self, session: Session):
+    def __init__(self, session: AsyncSession):
         self.session = session
 
-    def add(self, data: Bangumi):
+    async def add(self, data: Bangumi) -> bool:
         statement = select(Bangumi).where(Bangumi.title_raw == data.title_raw)
-        bangumi = self.session.exec(statement).first()
+        result = await self.session.execute(statement)
+        bangumi = result.scalar_one_or_none()
         if bangumi:
             return False
         self.session.add(data)
-        self.session.commit()
+        await self.session.commit()
+        _invalidate_bangumi_cache()
         logger.debug(f"[Database] Insert {data.official_title} into database.")
         return True
 
-    def add_all(self, datas: list[Bangumi]):
+    async def add_all(self, datas: list[Bangumi]):
         self.session.add_all(datas)
-        self.session.commit()
+        await self.session.commit()
+        _invalidate_bangumi_cache()
         logger.debug(f"[Database] Insert {len(datas)} bangumi into database.")
 
-    def update(self, data: Bangumi | BangumiUpdate, _id: int = None) -> bool:
+    async def update(self, data: Bangumi | BangumiUpdate, _id: int = None) -> bool:
         if _id and isinstance(data, BangumiUpdate):
-            db_data = self.session.get(Bangumi, _id)
+            db_data = await self.session.get(Bangumi, _id)
         elif isinstance(data, Bangumi):
-            db_data = self.session.get(Bangumi, data.id)
+            db_data = await self.session.get(Bangumi, data.id)
         else:
             return False
         if not db_data:
@@ -41,133 +57,155 @@ def update(self, data: Bangumi | BangumiUpdate, _id: int = None) -> bool:
         for key, value in bangumi_data.items():
             setattr(db_data, key, value)
         self.session.add(db_data)
-        self.session.commit()
-        self.session.refresh(db_data)
+        await self.session.commit()
+        _invalidate_bangumi_cache()
         logger.debug(f"[Database] Update {data.official_title}")
         return True
 
-    def update_all(self, datas: list[Bangumi]):
+    async def update_all(self, datas: list[Bangumi]):
         self.session.add_all(datas)
-        self.session.commit()
+        await self.session.commit()
+        _invalidate_bangumi_cache()
         logger.debug(f"[Database] Update {len(datas)} bangumi.")
 
-    def update_rss(self, title_raw, rss_set: str):
-        # Update rss and added
+    async def update_rss(self, title_raw: str, rss_set: str):
         statement = select(Bangumi).where(Bangumi.title_raw == title_raw)
-        bangumi = self.session.exec(statement).first()
-        bangumi.rss_link = rss_set
-        bangumi.added = False
-        self.session.add(bangumi)
-        self.session.commit()
-        self.session.refresh(bangumi)
-        logger.debug(f"[Database] Update {title_raw} rss_link to {rss_set}.")
-
-    def update_poster(self, title_raw, poster_link: str):
+        result = await self.session.execute(statement)
+        bangumi = result.scalar_one_or_none()
+        if bangumi:
+            bangumi.rss_link = rss_set
+            bangumi.added = False
+            self.session.add(bangumi)
+            await self.session.commit()
+            _invalidate_bangumi_cache()
+            logger.debug(f"[Database] Update {title_raw} rss_link to {rss_set}.")
+
+    async def update_poster(self, title_raw: str, poster_link: str):
         statement = select(Bangumi).where(Bangumi.title_raw == title_raw)
-        bangumi = self.session.exec(statement).first()
-        bangumi.poster_link = poster_link
-        self.session.add(bangumi)
-        self.session.commit()
-        self.session.refresh(bangumi)
-        logger.debug(f"[Database] Update {title_raw} poster_link to {poster_link}.")
-
-    def delete_one(self, _id: int):
+        result = await self.session.execute(statement)
+        bangumi = result.scalar_one_or_none()
+        if bangumi:
+            bangumi.poster_link = poster_link
+            self.session.add(bangumi)
+            await self.session.commit()
+            _invalidate_bangumi_cache()
+            logger.debug(f"[Database] Update {title_raw} poster_link to {poster_link}.")
+
+    async def delete_one(self, _id: int):
         statement = select(Bangumi).where(Bangumi.id == _id)
-        bangumi = self.session.exec(statement).first()
-        self.session.delete(bangumi)
-        self.session.commit()
-        logger.debug(f"[Database] Delete bangumi id: {_id}.")
+        result = await self.session.execute(statement)
+        bangumi = result.scalar_one_or_none()
+        if bangumi:
+            await self.session.delete(bangumi)
+            await self.session.commit()
+            _invalidate_bangumi_cache()
+            logger.debug(f"[Database] Delete bangumi id: {_id}.")
 
-    def delete_all(self):
+    async def delete_all(self):
         statement = delete(Bangumi)
-        self.session.exec(statement)
-        self.session.commit()
-
-    def search_all(self) -> list[Bangumi]:
+        await self.session.execute(statement)
+        await self.session.commit()
+        _invalidate_bangumi_cache()
+
+    async def search_all(self) -> list[Bangumi]:
+        global _bangumi_cache, _bangumi_cache_time
+        now = time.time()
+        if _bangumi_cache is not None and (now - _bangumi_cache_time) < _BANGUMI_CACHE_TTL:
+            return _bangumi_cache
         statement = select(Bangumi)
-        return self.session.exec(statement).all()
+        result = await self.session.execute(statement)
+        _bangumi_cache = list(result.scalars().all())
+        _bangumi_cache_time = now
+        return _bangumi_cache
 
-    def search_id(self, _id: int) -> Optional[Bangumi]:
+    async def search_id(self, _id: int) -> Optional[Bangumi]:
         statement = select(Bangumi).where(Bangumi.id == _id)
-        bangumi = self.session.exec(statement).first()
+        result = await self.session.execute(statement)
+        bangumi = result.scalar_one_or_none()
         if bangumi is None:
             logger.warning(f"[Database] Cannot find bangumi id: {_id}.")
             return None
         else:
             logger.debug(f"[Database] Find bangumi id: {_id}.")
-            return self.session.exec(statement).first()
+            return bangumi
 
-    def match_poster(self, bangumi_name: str) -> str:
-        # Use like to match
+    async def match_poster(self, bangumi_name: str) -> str:
         statement = select(Bangumi).where(
             func.instr(bangumi_name, Bangumi.official_title) > 0
         )
-        data = self.session.exec(statement).first()
+        result = await self.session.execute(statement)
+        data = result.scalar_one_or_none()
         if data:
             return data.poster_link
         else:
             return ""
 
-    def match_list(self, torrent_list: list, rss_link: str) -> list:
-        match_datas = self.search_all()
+    async def match_list(self, torrent_list: list, rss_link: str) -> list:
+        match_datas = await self.search_all()
         if not match_datas:
             return torrent_list
-        # Match title
-        i = 0
-        while i < len(torrent_list):
-            torrent = torrent_list[i]
-            for match_data in match_datas:
-                if match_data.title_raw in torrent.name:
-                    if rss_link not in match_data.rss_link:
+        # Build index for faster lookup
+        title_index = {m.title_raw: m for m in match_datas}
+        unmatched = []
+        rss_updated = set()
+        for torrent in torrent_list:
+            matched = False
+            for title_raw, match_data in title_index.items():
+                if title_raw in torrent.name:
+                    if rss_link not in match_data.rss_link and title_raw not in rss_updated:
                         match_data.rss_link += f",{rss_link}"
-                        self.update_rss(match_data.title_raw, match_data.rss_link)
-                    # if not match_data.poster_link:
-                    #     self.update_poster(match_data.title_raw, torrent.poster_link)
-                    torrent_list.pop(i)
+                        match_data.added = False
+                        rss_updated.add(title_raw)
+                    matched = True
                     break
-            else:
-                i += 1
-        return torrent_list
-
-    def match_torrent(self, torrent_name: str) -> Optional[Bangumi]:
+            if not matched:
+                unmatched.append(torrent)
+        # Batch commit all rss_link updates
+        if rss_updated:
+            await self.session.commit()
+            _invalidate_bangumi_cache()
+            logger.debug(f"[Database] Batch updated rss_link for {len(rss_updated)} bangumi.")
+        return unmatched
+
+    async def match_torrent(self, torrent_name: str) -> Optional[Bangumi]:
         statement = select(Bangumi).where(
             and_(
                 func.instr(torrent_name, Bangumi.title_raw) > 0,
-                # use `false()` to avoid E712 checking
-                # see: https://docs.astral.sh/ruff/rules/true-false-comparison/
                 Bangumi.deleted == false(),
             )
         )
-        return self.session.exec(statement).first()
+        result = await self.session.execute(statement)
+        return result.scalar_one_or_none()
 
-    def not_complete(self) -> list[Bangumi]:
-        # Find eps_complete = False
-        # use `false()` to avoid E712 checking
-        # see: https://docs.astral.sh/ruff/rules/true-false-comparison/
+    async def not_complete(self) -> list[Bangumi]:
         condition = select(Bangumi).where(
             and_(Bangumi.eps_collect == false(), Bangumi.deleted == false())
         )
-        datas = self.session.exec(condition).all()
-        return datas
+        result = await self.session.execute(condition)
+        return list(result.scalars().all())
 
-    def not_added(self) -> list[Bangumi]:
+    async def not_added(self) -> list[Bangumi]:
         conditions = select(Bangumi).where(
             or_(
-                Bangumi.added == 0, Bangumi.rule_name is None, Bangumi.save_path is None
+                Bangumi.added == 0,
+                Bangumi.rule_name is None,
+                Bangumi.save_path is None,
             )
         )
-        datas = self.session.exec(conditions).all()
-        return datas
+        result = await self.session.execute(conditions)
+        return list(result.scalars().all())
 
-    def disable_rule(self, _id: int):
+    async def disable_rule(self, _id: int):
         statement = select(Bangumi).where(Bangumi.id == _id)
-        bangumi = self.session.exec(statement).first()
-        bangumi.deleted = True
-        self.session.add(bangumi)
-        self.session.commit()
-        self.session.refresh(bangumi)
-        logger.debug(f"[Database] Disable rule {bangumi.title_raw}.")
-
-    def search_rss(self, rss_link: str) -> list[Bangumi]:
+        result = await self.session.execute(statement)
+        bangumi = result.scalar_one_or_none()
+        if bangumi:
+            bangumi.deleted = True
+            self.session.add(bangumi)
+            await self.session.commit()
+            logger.debug(f"[Database] Disable rule {bangumi.title_raw}.")
+
+    async def search_rss(self, rss_link: str) -> list[Bangumi]:
         statement = select(Bangumi).where(func.instr(rss_link, Bangumi.rss_link) > 0)
-        return self.session.exec(statement).all()
+        result = await self.session.execute(statement)
+        return list(result.scalars().all())