database 接入

konabot/common/database/__init__.py

@@ -0,0 +1,64 @@
+import os
+import sqlite3
+from typing import List, Dict, Any, Optional
+
+class DatabaseManager:
+    """超级无敌神奇的数据库！"""
+    
+    @classmethod
+    def query(cls, query: str, params: Optional[tuple] = None) -> List[Dict[str, Any]]:
+        """执行查询语句并返回结果"""
+        conn = sqlite3.connect(os.environ.get('DATABASE_PATH', './data/database.db'))
+        cursor = conn.cursor()
+        cursor.execute(query, params or ())
+        columns = [description[0] for description in cursor.description]
+        results = [dict(zip(columns, row)) for row in cursor.fetchall()]
+        cursor.close()
+        conn.close()
+        return results
+    
+    @classmethod
+    def query_by_sql_file(cls, file_path: str, params: Optional[tuple] = None) -> List[Dict[str, Any]]:
+        """从 SQL 文件中读取查询语句并执行"""
+        with open(file_path, 'r', encoding='utf-8') as f:
+            query = f.read()
+        return cls.query(query, params)
+
+    @classmethod
+    def execute(cls, command: str, params: Optional[tuple] = None) -> None:
+        """执行非查询语句"""
+        conn = sqlite3.connect(os.environ.get('DATABASE_PATH', './data/database.db'))
+        cursor = conn.cursor()
+        cursor.execute(command, params or ())
+        conn.commit()
+        cursor.close()
+        conn.close()
+
+    @classmethod
+    def execute_by_sql_file(cls, file_path: str, params: Optional[tuple] = None) -> None:
+        """从 SQL 文件中读取非查询语句并执行"""
+        with open(file_path, 'r', encoding='utf-8') as f:
+            command = f.read()
+        # 按照需要执行多条语句
+        commands = command.split(';')
+        for cmd in commands:
+            cmd = cmd.strip()
+            if cmd:
+                cls.execute(cmd, params)
+    
+    @classmethod
+    def execute_many(cls, command: str, seq_of_params: List[tuple]) -> None:
+        """执行多条非查询语句"""
+        conn = sqlite3.connect(os.environ.get('DATABASE_PATH', './data/database.db'))
+        cursor = conn.cursor()
+        cursor.executemany(command, seq_of_params)
+        conn.commit()
+        cursor.close()
+        conn.close()
+
+    @classmethod
+    def execute_many_values_by_sql_file(cls, file_path: str, seq_of_params: List[tuple]) -> None:
+        """从 SQL 文件中读取一条语句，但是被不同值同时执行"""
+        with open(file_path, 'r', encoding='utf-8') as f:
+            command = f.read()
+        cls.execute_many(command, seq_of_params)