chore(core.db): add missing type annotations

Author: Dt8333

astrbot/core/db/__init__.py

@@ -35,7 +35,7 @@ def __init__(self) -> None:
             self.engine, class_=AsyncSession, expire_on_commit=False
         )
 
-    async def initialize(self):
+    async def initialize(self) -> None:
         """初始化数据库连接"""
         pass
 
@@ -100,7 +100,7 @@ async def get_conversations(
         ...
 
     @abc.abstractmethod
-    async def get_conversation_by_id(self, cid: str) -> ConversationV2:
+    async def get_conversation_by_id(self, cid: str) -> ConversationV2 | None:
         """Get a specific conversation by its ID."""
         ...
 
@@ -118,7 +118,7 @@ async def get_filtered_conversations(
         page_size: int = 20,
         platform_ids: list[str] | None = None,
         search_query: str = "",
-        **kwargs,
+        **kwargs: T.Any,  # noqa: ANN401
     ) -> tuple[list[ConversationV2], int]:
         """Get conversations filtered by platform IDs and search query."""
         ...
@@ -145,7 +145,7 @@ async def update_conversation(
         title: str | None = None,
         persona_id: str | None = None,
         content: list[dict] | None = None,
-    ) -> None:
+    ) -> ConversationV2 | None:
         """Update a conversation's history."""
         ...
 
@@ -167,7 +167,7 @@ async def insert_platform_message_history(
         content: dict,
         sender_id: str | None = None,
         sender_name: str | None = None,
-    ) -> None:
+    ) -> PlatformMessageHistory | None:
         """Insert a new platform message history record."""
         ...
 
@@ -195,12 +195,12 @@ async def insert_attachment(
         path: str,
         type: str,
         mime_type: str,
-    ):
+    ) -> Attachment:
         """Insert a new attachment record."""
         ...
 
     @abc.abstractmethod
-    async def get_attachment_by_id(self, attachment_id: str) -> Attachment:
+    async def get_attachment_by_id(self, attachment_id: str) -> Attachment | None:
         """Get an attachment by its ID."""
         ...
 
@@ -216,7 +216,7 @@ async def insert_persona(
         ...
 
     @abc.abstractmethod
-    async def get_persona_by_id(self, persona_id: str) -> Persona:
+    async def get_persona_by_id(self, persona_id: str) -> Persona | None:
         """Get a persona by its ID."""
         ...
 
@@ -249,7 +249,9 @@ async def insert_preference_or_update(
         ...
 
     @abc.abstractmethod
-    async def get_preference(self, scope: str, scope_id: str, key: str) -> Preference:
+    async def get_preference(
+        self, scope: str, scope_id: str, key: str
+    ) -> Preference | None:
         """Get a preference by scope ID and key."""
         ...
 

astrbot/core/db/migration/helper.py

@@ -32,7 +32,7 @@ async def do_migration_v4(
     db_helper: BaseDatabase,
     platform_id_map: dict[str, dict[str, str]],
     astrbot_config: AstrBotConfig,
-):
+) -> None:
     """
     执行数据库迁移
     迁移旧的 webchat_conversation 表到新的 conversation 表。

astrbot/core/db/migration/migra_3_to_4.py

@@ -37,7 +37,7 @@ def get_platform_type(
 
 async def migration_conversation_table(
     db_helper: BaseDatabase, platform_id_map: dict[str, dict[str, str]]
-):
+) -> None:
     db_helper_v3 = SQLiteV3DatabaseV3(
         db_path=DB_PATH.replace("data_v4.db", "data_v3.db")
     )
@@ -91,7 +91,7 @@ async def migration_conversation_table(
 
 async def migration_platform_table(
     db_helper: BaseDatabase, platform_id_map: dict[str, dict[str, str]]
-):
+) -> None:
     db_helper_v3 = SQLiteV3DatabaseV3(
         db_path=DB_PATH.replace("data_v4.db", "data_v3.db")
     )
@@ -166,7 +166,7 @@ async def migration_platform_table(
 
 async def migration_webchat_data(
     db_helper: BaseDatabase, platform_id_map: dict[str, dict[str, str]]
-):
+) -> None:
     """迁移 WebChat 的历史记录到新的 PlatformMessageHistory 表中"""
     db_helper_v3 = SQLiteV3DatabaseV3(
         db_path=DB_PATH.replace("data_v4.db", "data_v3.db")
@@ -219,7 +219,7 @@ async def migration_webchat_data(
 
 async def migration_persona_data(
     db_helper: BaseDatabase, astrbot_config: AstrBotConfig
-):
+) -> None:
     """
     迁移 Persona 数据到新的表中。
     旧的 Persona 数据存储在 preference 中，新的 Persona 数据存储在 persona 表中。
@@ -261,7 +261,7 @@ async def migration_persona_data(
 
 async def migration_preferences(
     db_helper: BaseDatabase, platform_id_map: dict[str, dict[str, str]]
-):
+) -> None:
     # 1. global scope migration
     keys = [
         "inactivated_llm_tools",

astrbot/core/db/migration/migra_45_to_46.py

@@ -3,7 +3,7 @@
 from astrbot.core.umop_config_router import UmopConfigRouter
 
 
-async def migrate_45_to_46(acm: AstrBotConfigManager, ucr: UmopConfigRouter):
+async def migrate_45_to_46(acm: AstrBotConfigManager, ucr: UmopConfigRouter) -> None:
     abconf_data = acm.abconf_data
 
     if not isinstance(abconf_data, dict):

astrbot/core/db/migration/shared_preferences_v3.py

@@ -7,13 +7,13 @@
 
 
 class SharedPreferences:
-    def __init__(self, path=None):
+    def __init__(self, path: str | None = None) -> None:
         if path is None:
             path = os.path.join(get_astrbot_data_path(), "shared_preferences.json")
         self.path = path
         self._data = self._load_preferences()
 
-    def _load_preferences(self):
+    def _load_preferences(self) -> dict:
         if os.path.exists(self.path):
             try:
                 with open(self.path) as f:
@@ -22,24 +22,24 @@ def _load_preferences(self):
                 os.remove(self.path)
         return {}
 
-    def _save_preferences(self):
+    def _save_preferences(self) -> None:
         with open(self.path, "w") as f:
             json.dump(self._data, f, indent=4, ensure_ascii=False)
             f.flush()
 
-    def get(self, key, default: _VT = None) -> _VT:
+    def get(self, key: str, default: _VT = None) -> _VT:
         return self._data.get(key, default)
 
-    def put(self, key, value):
+    def put(self, key: str, value: object) -> None:
         self._data[key] = value
         self._save_preferences()
 
-    def remove(self, key):
+    def remove(self, key: str) -> None:
         if key in self._data:
             del self._data[key]
             self._save_preferences()
 
-    def clear(self):
+    def clear(self) -> None:
         self._data.clear()
         self._save_preferences()
 

astrbot/core/db/migration/sqlite_v3.py

@@ -126,7 +126,7 @@ def _get_conn(self, db_path: str) -> sqlite3.Connection:
         conn.text_factory = str
         return conn
 
-    def _exec_sql(self, sql: str, params: tuple = None):
+    def _exec_sql(self, sql: str, params: tuple | None = None) -> None:
         conn = self.conn
         try:
             c = self.conn.cursor()
@@ -143,7 +143,7 @@ def _exec_sql(self, sql: str, params: tuple = None):
 
         conn.commit()
 
-    def insert_platform_metrics(self, metrics: dict):
+    def insert_platform_metrics(self, metrics: dict) -> None:
         for k, v in metrics.items():
             self._exec_sql(
                 """
@@ -152,7 +152,7 @@ def insert_platform_metrics(self, metrics: dict):
                 (k, v, int(time.time())),
             )
 
-    def insert_llm_metrics(self, metrics: dict):
+    def insert_llm_metrics(self, metrics: dict) -> None:
         for k, v in metrics.items():
             self._exec_sql(
                 """
@@ -225,7 +225,9 @@ def get_grouped_base_stats(self, offset_sec: int = 86400) -> Stats:
 
         return Stats(platform, [], [])
 
-    def get_conversation_by_user_id(self, user_id: str, cid: str) -> Conversation:
+    def get_conversation_by_user_id(
+        self, user_id: str, cid: str
+    ) -> Conversation | None:
         try:
             c = self.conn.cursor()
         except sqlite3.ProgrammingError:
@@ -246,7 +248,7 @@ def get_conversation_by_user_id(self, user_id: str, cid: str) -> Conversation:
 
         return Conversation(*res)
 
-    def new_conversation(self, user_id: str, cid: str):
+    def new_conversation(self, user_id: str, cid: str) -> None:
         history = "[]"
         updated_at = int(time.time())
         created_at = updated_at
@@ -257,7 +259,7 @@ def new_conversation(self, user_id: str, cid: str):
             (user_id, cid, history, updated_at, created_at),
         )
 
-    def get_conversations(self, user_id: str) -> tuple:
+    def get_conversations(self, user_id: str) -> list[Conversation]:
         try:
             c = self.conn.cursor()
         except sqlite3.ProgrammingError:
@@ -284,7 +286,7 @@ def get_conversations(self, user_id: str) -> tuple:
             )
         return conversations
 
-    def update_conversation(self, user_id: str, cid: str, history: str):
+    def update_conversation(self, user_id: str, cid: str, history: str) -> None:
         """更新对话，并且同时更新时间"""
         updated_at = int(time.time())
         self._exec_sql(
@@ -294,23 +296,25 @@ def update_conversation(self, user_id: str, cid: str, history: str):
             (history, updated_at, user_id, cid),
         )
 
-    def update_conversation_title(self, user_id: str, cid: str, title: str):
+    def update_conversation_title(self, user_id: str, cid: str, title: str) -> None:
         self._exec_sql(
             """
             UPDATE webchat_conversation SET title = ? WHERE user_id = ? AND cid = ?
             """,
             (title, user_id, cid),
         )
 
-    def update_conversation_persona_id(self, user_id: str, cid: str, persona_id: str):
+    def update_conversation_persona_id(
+        self, user_id: str, cid: str, persona_id: str
+    ) -> None:
         self._exec_sql(
             """
             UPDATE webchat_conversation SET persona_id = ? WHERE user_id = ? AND cid = ?
             """,
             (persona_id, user_id, cid),
         )
 
-    def delete_conversation(self, user_id: str, cid: str):
+    def delete_conversation(self, user_id: str, cid: str) -> None:
         self._exec_sql(
             """
             DELETE FROM webchat_conversation WHERE user_id = ? AND cid = ?
@@ -381,11 +385,11 @@ def get_filtered_conversations(
         self,
         page: int = 1,
         page_size: int = 20,
-        platforms: list[str] = None,
-        message_types: list[str] = None,
-        search_query: str = None,
-        exclude_ids: list[str] = None,
-        exclude_platforms: list[str] = None,
+        platforms: list[str] | None = None,
+        message_types: list[str] | None = None,
+        search_query: str | None = None,
+        exclude_ids: list[str] | None = None,
+        exclude_platforms: list[str] | None = None,
     ) -> tuple[list[dict[str, Any]], int]:
         """获取筛选后的对话列表"""
         try: