qing/micro_servers
1
3 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

rewrite

Browse Source
This commit is contained in:
shufeng2012 committed 2026-07-24 06:35:46 +08:00
1 parent 0603c473fa
commit 03d7b07bf6
5 files changed
+334 -231
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

sqlite3/.gitignore
+112
View File
@@ -0,0 +1,112 @@
#prjs
*.db
*.bin
*.build
*.dist
*.onefile-build
nuitka-crash-report.xml
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
.hypothesis/
.pytest_cache/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
target/
# Jupyter Notebook
.ipynb_checkpoints
# pyenv
.python-version
# celery beat schedule file
celerybeat-schedule
# SageMath parsed files
*.sage.py
# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
sqlite3/README.md
+83 -187
View File
@@ -1,66 +1,96 @@
# 跨数据库通用查询服务（基于 SQLAlchemy Core）
# Flask SQLite 执行服务
一个轻量级的 Flask HTTP 服务，用于安全地执行带参数的 SQLite SQL 语句，支持 `?` 和命名参数占位符，杜绝 SQL 注入。
一个安全的 HTTP 服务，通过结构化 JSON 描述查询意图，后端自动转换为对应数据库的 SQL，无需手写 SQL，即可无缝切换 SQLite、PostgreSQL、MySQL 等数据库。
## 功能特点
- 通过 JSON 接口发送 SQL 和参数
- 强制使用参数绑定（`execute(sql, params)`），无 SQL 注入风险
- 统一返回格式，所有语句执行结果均包含 `data` 数组
- 自动事务管理，异常时回滚
- 单线程单进程模式，绑定 127.0.0.1，仅本地访问
- **跨数据库**：基于 SQLAlchemy Core 表达式语言，一套接口适配多种数据库
- **零 SQL 注入**：所有条件、值均通过 SQLAlchemy 参数绑定处理
- **统一 JSON 接口**：`action` + `table` + `conditions` 模式
- **支持增删改查**：select, insert, update, delete 全覆盖
- **单线程本地监听**：waitress + 127.0.0.1:8005，简单可靠
## 快速开始
### 环境要求
- Python 3.7+
- pip
### 安装依赖
```bash
pip install flask waitress
pip install flask sqlalchemy waitress
# 根据需要安装数据库驱动，例如：
# pip install psycopg2-binary # PostgreSQL
# pip install pymysql # MySQL
```
### 启动服务
### 启动服务（默认 SQLite）
```bash
python app.py
```
服务运行在 `http://127.0.0.1:8005`。
服务运行在 `http://127.0.0.1:5000`，仅允许本地访问。数据库文件 `database.db` 会在首次连接时自动创建（可提前建表或通过接口执行 `CREATE TABLE`）。
使用其他数据库：
```bash
export DATABASE_URL="postgresql://user:pass@localhost/dbname"
python app.py
```
**注意**：新接口依赖表已存在（通过反射获取结构）。首次使用时请先手动建表：
```bash
sqlite3 database.db "CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT, age INTEGER)"
```
## API 文档
### 执行 SQL 语句
### 请求格式
**端点**：`POST /execute`
**请求格式**：`Content-Type: application/json`
**端点**：`POST /execute`
**Content-Type**：`application/json`
```json
{
"sql": "SQL 语句，使用 ? 或 :key 作为占位符",
"params": [] 或 {}
"action": "操作类型", // select | insert | update | delete
"table": "表名",
"columns": ["列1", "列2"], // select/insert 时可选
"values": { // insert/update 时必填
"列名": "值"
},
"where": { // 条件，支持组合操作符
"age": {"ge": 18, "lt": 60},
"name": "Alice"
},
"order_by": [{"column": "age", "desc": true}],
"limit": 10,
"offset": 0
}
```
- `sql`：待执行的 SQL 语句（仅允许单条语句）
- `params`：参数列表 `[值1, 值2, ...]` 或参数字典 `{"key": 值}`
### where 条件支持的运算符
**成功响应**：
| 运算符 | 说明 | 示例 |
|--------|------|------|
| 直接值 | 等于 | `"name": "Alice"` |
| eq | 等于 | `"age": {"eq": 30}` |
| ne | 不等于 | `"status": {"ne": "deleted"}` |
| gt | 大于 | `"score": {"gt": 80}` |
| ge | 大于等于 | `"score": {"ge": 60}` |
| lt | 小于 | `"price": {"lt": 100}` |
| le | 小于等于 | `"price": {"le": 200}` |
| like | 模糊匹配 | `"name": {"like": "%Smith%"}` |
| in | 包含于列表 | `"id": {"in": [1,2,3]}` |
多个条件自动使用 **AND** 连接。
### 响应格式
成功：
```json
{
"success": true,
"data": [] // 查询结果数组，每条记录为字典；非查询语句返回空数组
"data": [ // 查询结果数组（insert/update/delete 时包含影响行数或 lastrowid）
{"id": 1, "name": "Alice"}
]
}
```
**错误响应**：
失败：
```json
{
"success": false,
@@ -68,184 +98,50 @@ python app.py
}
```
## 使用示例
以下示例均通过 `curl` 和 Python `requests` 库演示。
### 1. 创建数据表
## 使用示例（curl）
**插入**：
```bash
curl -X POST http://127.0.0.1:5000/execute \
curl -X POST http://127.0.0.1:8005/execute \
-H "Content-Type: application/json" \
-d '{"sql": "CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, age INTEGER)"}'
-d '{"action":"insert","table":"users","values":{"name":"Bob","age":25}}'
```
Python 版本：
```python
import requests
url = "http://127.0.0.1:5000/execute"
payload = {
"sql": "CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, age INTEGER)"
}
resp = requests.post(url, json=payload)
print(resp.json()) # {"success": true, "data": []}
```
### 2. 插入数据（使用 `?` 占位符）
**查询**：
```bash
curl -X POST http://127.0.0.1:5000/execute \
curl -X POST http://127.0.0.1:8005/execute \
-H "Content-Type: application/json" \
-d '{"sql": "INSERT INTO users (name, age) VALUES (?, ?)", "params": ["Alice", 30]}'
-d '{"action":"select","table":"users","where":{"age":{"gt":20}},"order_by":[{"column":"id","desc":true}],"limit":5}'
```
Python 版本：
```python
payload = {
"sql": "INSERT INTO users (name, age) VALUES (?, ?)",
"params": ["Bob", 25]
}
resp = requests.post(url, json=payload)
print(resp.json()) # {"success": true, "data": []}
```
### 3. 插入数据（使用命名参数 `:key`）
**更新**：
```bash
curl -X POST http://127.0.0.1:5000/execute \
curl -X POST http://127.0.0.1:8005/execute \
-H "Content-Type: application/json" \
-d '{"sql": "INSERT INTO users (name, age) VALUES (:name, :age)", "params": {"name": "Charlie", "age": 28}}'
-d '{"action":"update","table":"users","values":{"age":26},"where":{"name":"Bob"}}'
```
### 4. 查询全部数据
**删除**：
```bash
curl -X POST http://127.0.0.1:5000/execute \
curl -X POST http://127.0.0.1:8005/execute \
-H "Content-Type: application/json" \
-d '{"sql": "SELECT * FROM users"}'
```
响应示例：
```json
{
"success": true,
"data": [
{"id": 1, "name": "Alice", "age": 30},
{"id": 2, "name": "Bob", "age": 25},
{"id": 3, "name": "Charlie", "age": 28}
]
}
```
Python 版本：
```python
payload = {"sql": "SELECT * FROM users"}
resp = requests.post(url, json=payload)
users = resp.json()["data"]
for user in users:
print(user["name"], user["age"])
```
### 5. 按条件查询
```bash
curl -X POST http://127.0.0.1:5000/execute \
-H "Content-Type: application/json" \
-d '{"sql": "SELECT * FROM users WHERE age > ?", "params": [26]}'
```
Python 版本：
```python
payload = {
"sql": "SELECT * FROM users WHERE age > :min_age",
"params": {"min_age": 26}
}
resp = requests.post(url, json=payload)
print(resp.json()["data"])
```
### 6. 更新数据
```bash
curl -X POST http://127.0.0.1:5000/execute \
-H "Content-Type: application/json" \
-d '{"sql": "UPDATE users SET age = ? WHERE name = ?", "params": [31, "Alice"]}'
```
Python 版本：
```python
payload = {
"sql": "UPDATE users SET age = :new_age WHERE id = :user_id",
"params": {"new_age": 26, "user_id": 2}
}
resp = requests.post(url, json=payload)
print(resp.json()) # {"success": true, "data": []}
```
### 7. 删除数据
```bash
curl -X POST http://127.0.0.1:5000/execute \
-H "Content-Type: application/json" \
-d '{"sql": "DELETE FROM users WHERE id = ?", "params": [3]}'
```
Python 版本：
```python
payload = {
"sql": "DELETE FROM users WHERE name = :name",
"params": {"name": "Charlie"}
}
resp = requests.post(url, json=payload)
print(resp.json())
```
### 8. 错误处理示例
当 SQL 或参数错误时，返回明确的错误信息：
```bash
curl -X POST http://127.0.0.1:5000/execute \
-H "Content-Type: application/json" \
-d '{"sql": "SELECT * FROM nonexistent_table"}'
```
响应：
```json
{
"success": false,
"error": "no such table: nonexistent_table"
}
-d '{"action":"delete","table":"users","where":{"id":{"in":[1,2]}}}'
```
## 安全机制
1. **参数绑定**：使用 `sqlite3` 官方推荐的 `execute(sql, params)` 方式传参，参数值与 SQL 结构完全分离。
2. **单语句执行**：`sqlite3.execute()` 仅执行第一条语句，无法通过分号注入多条命令。
3. **错误回滚**：若执行过程中出现异常，已修改的数据会回滚，确保数据一致性。
4. **本地绑定**：服务监听 `127.0.0.1`，仅本机可访问，避免网络暴露风险。
- 强制使用 SQLAlchemy 参数绑定，杜绝 SQL 注入。
- 仅反射已有表，无法创建或删除表（保证结构安全）。
- 所有异常自动回滚事务。
- 默认监听 `127.0.0.1`，仅本机可访问。
## 项目结构
## 切换数据库
```
.
├── app.py # 主程序（使用 waitress 单线程模式）
└── README.md # 本文档
```
只需修改环境变量 `DATABASE_URL`，并安装对应驱动：
数据库文件 `database.db` 会在运行时自动生成。
- SQLite：`sqlite:///database.db`
- PostgreSQL：`postgresql://user:pass@localhost/dbname` （需 `pip install psycopg2`）
- MySQL：`mysql+pymysql://user:pass@localhost/dbname` （需 `pip install pymysql`）
## 注意事项
SQLAlchemy 会自动生成适配各数据库的 SQL 方言，无需修改接口调用。
- 服务采用 waitress 单线程模式（`threads=1`），确保请求顺序执行，适合低并发场景。
- 如需处理更高并发，可适当增加 `threads` 参数（例如 `threads=4`），但仍需注意 SQLite 的并发限制。
- 该服务无认证机制，推荐仅在本机或内网可信环境使用。如需对外服务，请添加鉴权并考虑使用 nginx 等反向代理。
sqlite3/_run.sh
+2 -2
View File
@@ -1,5 +1,5 @@
#!/bin/bash
cd "$(dirname "$0")" || exit
source ./venv/bin/activate
#exec ./main.bin
exec python3 main.py
exec ./main.bin
#exec python3 main.py
sqlite3/main.py
+135 -41
View File
@@ -1,57 +1,151 @@
import sqlite3
import os
from flask import Flask, request, jsonify, g
from sqlalchemy import (
create_engine, MetaData, Table, select, insert, update, delete, text
)
from sqlalchemy.exc import SQLAlchemyError
from waitress import serve
app = Flask(__name__)
app.config['DATABASE'] = 'database.db'
def get_db():
if 'db' not in g:
conn = sqlite3.connect(app.config['DATABASE'])
conn.row_factory = sqlite3.Row
g.db = conn
return g.db
# 数据库配置
DATABASE_URL = os.environ.get('DATABASE_URL', 'sqlite:///database.db')
engine = create_engine(DATABASE_URL)
@app.teardown_appcontext
def close_db(exception):
db = g.pop('db', None)
if db is not None:
db.close()
# 缓存反射的元数据（避免每次请求都反射）
metadata = MetaData()
def get_table(table_name):
"""反射表对象（若已缓存则直接返回）"""
if table_name not in metadata.tables:
Table(table_name, metadata, autoload_with=engine)
return metadata.tables[table_name]
def build_where(table, conditions):
"""将条件字典转换为 SQLAlchemy 条件表达式"""
if not conditions:
return None
clauses = []
for col, val in conditions.items():
column = table.columns[col]
if isinstance(val, dict):
# 支持多种比较操作
if 'eq' in val:
clauses.append(column == val['eq'])
if 'ne' in val:
clauses.append(column != val['ne'])
if 'gt' in val:
clauses.append(column > val['gt'])
if 'lt' in val:
clauses.append(column < val['lt'])
if 'ge' in val:
clauses.append(column >= val['ge'])
if 'le' in val:
clauses.append(column <= val['le'])
if 'like' in val:
clauses.append(column.like(val['like']))
if 'in' in val:
clauses.append(column.in_(val['in']))
else:
# 直接等值比较
clauses.append(column == val)
from sqlalchemy import and_
return and_(*clauses)
@app.route('/execute', methods=['POST'])
def execute_sql():
def execute_structured():
data = request.get_json()
if not data or 'sql' not in data:
return jsonify({'success': False, 'error': '缺少 sql 字段'}), 400
if not data:
return jsonify({'success': False, 'error': '请求体不能为空'}), 400
sql = data['sql'].strip()
params = data.get('params', [])
# 参数必须是列表、元组或字典
if not isinstance(params, (list, tuple, dict)):
params = []
action = data.get('action', 'select')
table_name = data.get('table')
if not table_name:
return jsonify({'success': False, 'error': '缺少 table 字段'}), 400
try:
db = get_db()
cur = db.execute(sql, params)
db.commit()
try:
rows = cur.fetchall()
result = [dict(row) for row in rows]
except Exception:
result = []
return jsonify({
'success': True,
'data': result
})
table = get_table(table_name)
except Exception as e:
db = g.get('db')
if db:
db.rollback()
return jsonify({'success': False, 'error': f'表加载失败: {str(e)}'}), 400
conn = None
try:
conn = engine.connect()
result = []
if action == 'select':
columns = data.get('columns', None)
if columns:
cols = [table.columns[c] for c in columns]
else:
cols = [table] # select *
stmt = select(*cols)
where_cond = build_where(table, data.get('where'))
if where_cond is not None:
stmt = stmt.where(where_cond)
# 排序
order_by = data.get('order_by')
if order_by:
for item in order_by:
col = table.columns[item['column']]
stmt = stmt.order_by(col.desc() if item.get('desc') else col.asc())
# 分页
limit = data.get('limit')
if limit is not None:
stmt = stmt.limit(limit)
offset = data.get('offset')
if offset is not None:
stmt = stmt.offset(offset)
res = conn.execute(stmt)
rows = res.fetchall()
result = [dict(row._mapping) for row in rows]
elif action == 'insert':
values = data.get('values', {})
stmt = insert(table).values(**values)
res = conn.execute(stmt)
conn.commit()
result = [{'lastrowid': res.lastrowid}] if hasattr(res, 'lastrowid') else []
elif action == 'update':
values = data.get('values', {})
where_cond = build_where(table, data.get('where'))
stmt = update(table).values(**values)
if where_cond is not None:
stmt = stmt.where(where_cond)
res = conn.execute(stmt)
conn.commit()
result = [{'rowcount': res.rowcount}]
elif action == 'delete':
where_cond = build_where(table, data.get('where'))
stmt = delete(table)
if where_cond is not None:
stmt = stmt.where(where_cond)
res = conn.execute(stmt)
conn.commit()
result = [{'rowcount': res.rowcount}]
else:
return jsonify({'success': False, 'error': f'不支持的操作: {action}'}), 400
return jsonify({'success': True, 'data': result})
except SQLAlchemyError as e:
if conn:
conn.rollback()
return jsonify({'success': False, 'error': str(e)}), 400
except Exception as e:
if conn:
conn.rollback()
return jsonify({'success': False, 'error': str(e)}), 400
finally:
if conn:
conn.close()
if __name__ == '__main__':
# 单线程单进程，绑定 127.0.0.1:8005
serve(app, host='127.0.0.1', port=8005, threads=1)
# 单线程单进程，绑定 127.0.0.1
serve(app, host='127.0.0.1', port=8005, threads=1)
sqlite3/requirements.txt
+2 -1
View File
@@ -1,2 +1,3 @@
nuitka
flask
waitress
waitress