Flask+Sqlite3相关问题求助

player-none

Code

1. import sqlite3
   
2. from queue import Queue
   
3. from typing import Any
   
4. from flask import Flask, request, send_file
   
5. from threading import Thread
   
6. 
   
7. class SQLiteExecutor:
   
8.     """SQLite3 仅支持在同一线程内操作，但 flask 的机制使其并不在同一线程内，故封装此类以解决此问题"""
   
9.     def __init__(self, path: str):
   
10.         """
    
11.         :param path: SQLite3 文件路径
    
12.         """
    
13.         self.path = path
    
14.         # 每次需要执行的时候都往 Queue 里扔东西就行
    
15.         self.queue = Queue()
    
16.         # 执行完之后把结果放到这里
    
17.         self.results = Queue()
    
18.         # 把所有操作都封装到 _thread 方法里，这样就不会报错了
    
19.         Thread(target=self._thread, daemon=True).start()
    
20.    
    
21.     def _thread(self) -> None:
    
22.         """线程函数。
    
23.         
    
24.         `_thread()` 一直监听 `queue`，当 `queue` 发生改变时 `_thread()` 就会自动执行里面的函数。
    
25.         
    
26.         执行完函数后，`_thread` 会立即把结果（如有）放到 `results` 里，所以理论上不存在等待时间。
    
27.         """
    
28.         while True:
    
29.             conn = sqlite3.connect(self.path)
    
30.             cur = conn.cursor()
    
31.             if (result := self.queue.get()(conn, cur)) is not None:
    
32.                 self.results.put(result)
    
33.    
    
34.     def do(self, command: str, commit: bool = True) -> None:
    
35.         """对 `cur` 执行一条命令。
    
36.         
    
37.         `do()` 的原理是将 `_execute()` 的返回值放到 `queue` 里，`_thread` 在监听到有元素进入时就会自动执行它。
    
38.         当 `_thread` 执行完毕后，它会把结果放到 `results` 里，只要监听 `results` 即可。
    
39.         
    
40.         :param command: SQLite3 命令
    
41.         :param commit: 命令执行完后是否要 commit
    
42.         :return: 无
    
43.         """
    
44.         def inner(conn: sqlite3.Connection, cur: sqlite3.Cursor):
    
45.             cur.execute(command)
    
46.             if commit:
    
47.                 conn.commit()
    
48.         self.queue.put(inner)
    
49.    
    
50.     def cur(self, name: str) -> Any:
    
51.         """让 `cur` 执行 `name`。
    
52.         
    
53.         :param name: 例如 `name == 'fetchall'`，执行 `cur.fetchall()`。
    
54.         :return: 执行的结果
    
55.         """
    
56.         def inner(_: sqlite3.Connection, cur: sqlite3.Cursor):
    
57.             return getattr(cur, name)()
    
58.         self.queue.put(inner)
    
59.         return self.results.get()
    
60. 
    
61. ext = SQLiteExecutor('data.db')
    
62. ext.do('''CREATE TABLE IF NOT EXISTS users (
    
63.     id INTEGER PRIMARY KEY AUTOINCREMENT,
    
64.     name TEXT NOT NULL,
    
65.     pwd VARCHAR(64) NOT NULL
    
66. )''')
    
67. 
    
68. def add_route(route_path: str, file_path: str = '') -> None:
    
69.     """添加一条 route。
    
70. 
    
71.     调用这个函数，可以不用单独为每个 route 写一个函数，程序使用变量 `_rid` 自动为每个 route 分配名称。
    
72.    
    
73.     :param route_path: 用户想要看到对应文件应该访问的***网址***
    
74.     :param file_path: 用户访问网住可以看到的文件的***路径***
    
75.     """
    
76.     global _rid
    
77.     file_path = (file_path or route_path).lstrip('/')
    
78.     exec(f'''@app.route({repr(route_path)})
    
79. def route{_rid}():
    
80.     return send_file({repr(file_path)})
    
81. ''')
    
82.     _rid += 1
    
83. 
    
84. def read(path: str) -> str:
    
85.     """读取一个文本文件的内容。
    
86.    
    
87.     调用此函数可以避免大量重复 with-as 语句，不仅让代码变多，还降低了可读性，敲起来也十分乏味。
    
88.    
    
89.     :param path: 文本文件的路径
    
90.     :return: 文本文件的内容
    
91.     """
    
92.     with open(path, encoding='utf-8') as f:
    
93.         return f.read()
    
94. 
    
95. _rid = 0
    
96. app = Flask(__name__)
    
97. 
    
98. add_route('/favicon.ico')
    
99. add_route('/base.css')
    
100. add_route('/daohanglan.js')
     
101. 
     
102. @app.route('/')
     
103. def main():
     
104.     return read('index.html'), 200
     
105. 
     
106. @app.route('/submit', methods=['POST'])
     
107. def handler():
     
108.     json = request.get_json()
     
109.     try:
     
110.         # todo 代表用户要做的请求
     
111.         todo = json['todo']
     
112.         if todo == 'login':
     
113.             # 用户想要登录
     
114.             # 密码 pwd 是 SHA256 加密后的样子
     
115.             name, pwd = json['name'], json['pwd']
     
116.             ext.do(f"SELECT * FROM users", False)
     
117.             print(ext.cur('fetchall'))
     
118.         elif todo == 'register':
     
119.             # 注册
     
120.             # 密码 pwd 是 SHA256 加密后的样子
     
121.             name, pwd = json['name'], json['pwd']
     
122.             ext.do(f"INSERT INTO users (name, pwd) VALUES ('{name}', '{pwd}');")
     
123.     except KeyError as ex:
     
124.         return f'400 Bad Request (Key {repr(ex.args[0])} Not Found)', 400
     
125.     return str(), 200
     
126. 
     
127. if __name__ == '__main__':
     
128.     app.run(port=22222, debug=True)
     

复制代码

输出的是空列表 []，数据库里已经有数据了

player-none

@FishC 求救

FishC

player-none 发表于 2025-05-19 18:31
@FishC 求救

Python反转字符串的两种常用方法

这里提供两种简单易懂的实现方式，根据需求选择即可：

• 切片法（推荐）
  直接通过切片操作[::-1]实现反转：
  
  1. 
     
  2. s = "FishC"
     
  3. reversed_str = s[::-1]
     
  4. print(reversed_str)  # 输出：ChsiF
     

  复制代码
  
  
• reversed函数法
  使用内置函数生成反转后的迭代器，再拼接字符串：
  
  1. 
     
  2. s = "hello"
     
  3. reversed_str = ''.join(reversed(s))
     
  4. print(reversed_str)  # 输出：olleh
     

  复制代码
  
  

注意：
1. 两种方法都不会修改原字符串（字符串不可变）
2. 切片法性能更优，适合简单场景
3. reversed函数在处理复杂对象时更灵活

以上回复来自 FishC AI 小助理，如未能正确解答您的问题，请继续追问。