使用单独的Python进程执行SQLite查询
(self, sql: str, data: Union[Sequence, Dict[str, Any]] = ())
| 185 | pass |
| 186 | |
| 187 | async def execute(self, sql: str, data: Union[Sequence, Dict[str, Any]] = ()) -> str: |
| 188 | """使用单独的Python进程执行SQLite查询""" |
| 189 | try: |
| 190 | # 将查询参数转换为JSON字符串 |
| 191 | params_json = json.dumps(data) if isinstance(data, dict) else json.dumps(list(data)) |
| 192 | |
| 193 | # 创建内嵌的Python脚本代码，直接包含SQLite查询逻辑 |
| 194 | python_code = f''' |
| 195 | import sqlite3 |
| 196 | import json |
| 197 | import sys |
| 198 | |
| 199 | try: |
| 200 | # 解析参数 |
| 201 | params = json.loads('{params_json}') if '{params_json}' else () |
| 202 | |
| 203 | # 连接数据库 |
| 204 | conn = sqlite3.connect('{self.sqlite_path}', timeout=10.0) |
| 205 | cursor = conn.cursor() |
| 206 | |
| 207 | # 执行查询 |
| 208 | try: |
| 209 | cursor.execute({repr(sql)}, params) |
| 210 | try: |
| 211 | result = cursor.fetchall() |
| 212 | except sqlite3.OperationalError as e: |
| 213 | if "no results" in str(e).lower(): |
| 214 | conn.commit() |
| 215 | result = [] |
| 216 | else: |
| 217 | conn.rollback() |
| 218 | raise |
| 219 | result_str = str(result) |
| 220 | except Exception as e: |
| 221 | error_msg = f"SQLite execution error: {{str(e)}}" |
| 222 | print(error_msg, file=sys.stderr) |
| 223 | try: |
| 224 | conn.rollback() |
| 225 | except: |
| 226 | pass |
| 227 | result_str = error_msg |
| 228 | finally: |
| 229 | cursor.close() |
| 230 | conn.close() |
| 231 | |
| 232 | except Exception as e: |
| 233 | error_msg = f"SQLite process error: {{str(e)}}" |
| 234 | print(error_msg, file=sys.stderr) |
| 235 | result_str = error_msg |
| 236 | |
| 237 | # 截断过长的结果并输出 |
| 238 | if len(result_str) > 800: |
| 239 | result_str = result_str[:800] + "[TRUNCATED]" |
| 240 | print(result_str) |
| 241 | ''' |
| 242 | |
| 243 | # 使用子进程执行内嵌代码 |
| 244 | process = await asyncio.create_subprocess_exec( |