qing
101aed48be
pymsg 客户端: - MsgClient.call() 新增 retry_on_refused/retry_interval 参数 - retry_on_refused=True 时，ConnectionRefusedError 不抛出，按 retry_interval(默认1s)阻塞重试 - 重试期间每5秒向 stderr 打印等待提示，避免日志刷屏 - 仅对 ConnectionRefusedError 重试；TimeoutError/ConnectionError/RpcError 仍立即抛出 - 默认 retry_on_refused=False，完全向后兼容 - RpcClient 构造函数新增同名关键字参数，call() 支持单次调用覆盖 - 新增 time 模块导入 - 更新 README 文档（新增"启动等待模式"章节和参数表） 各服务（db/dnfs/static）: - _fetch_from_confs() 改用 retry_on_refused=True，移除手动 ConnectionRefusedError 捕获和 sys.exit - 保留 RpcError（如配置文件不存在）的友好错误处理和 sys.exit(1) - 启动时不再因 confs 未就绪而退出：pymsg 自动阻塞等待 confs 启动后自动连接成功 - 启动顺序从"刚性要求 confs 先启动"变为"不严格要求启动顺序" - log_client 保持默认 retry_on_refused=False，后台线程不阻塞（fail-safe 丢弃日志） - 更新各服务 README 和 confs README 的启动顺序描述 测试验证: - 基本 call() 正常工作 - retry_on_refused=False 连接拒绝立即抛 ConnectionRefusedError - retry_on_refused=True 阻塞等待（延迟1.5s启动服务后成功连接，重试日志正常输出）
307 lines
12 KiB
Markdown
307 lines
12 KiB
Markdown
# 通用数据库服务 db（pymsg/TCP + SQLAlchemy Core）
|
||
|
||
一个安全的数据库访问服务。各业务服务通过自研 pymsg 协议（TCP 长度前缀帧 + JSON）发送结构化查询意图，db 翻译为 SQL 执行，全程参数绑定，业务侧无需直连数据库、无需手写 SQL。
|
||
|
||
## 功能特点
|
||
|
||
- **跨数据库**：基于 SQLAlchemy Core 表达式语言，一套接口适配 SQLite / PostgreSQL / MySQL
|
||
- **零 SQL 注入**：所有条件、值均通过 SQLAlchemy 参数绑定处理
|
||
- **统一 JSON 接口**：`_route` + `action` + `table` + `where` 模式
|
||
- **支持增删改查**：select / insert / update / delete 全覆盖
|
||
- **内置日志表**：启动自动建 Logs 表，供全系统集中写日志；带后台自动清理
|
||
- **仅监听本机**：默认 `127.0.0.1:8005`，不对外暴露
|
||
|
||
## 快速开始
|
||
|
||
### 安装依赖
|
||
|
||
```bash
|
||
./_set.sh # 建 venv 并安装 requirements.txt（SQLAlchemy + 本地 ../pymsg）
|
||
```
|
||
|
||
### 启动服务
|
||
|
||
```bash
|
||
./_run.sh # venv 中执行 python3 dbv2.py
|
||
```
|
||
|
||
服务监听 `tcp://0.0.0.0:8005`（默认）。生产环境由 core（Nidus）按 init.txt 统一拉起，守护靠 systemd。
|
||
|
||
**注意**：`execute` 路由依赖表已存在（通过反射获取结构）。建表见 `create_sqlite3.sql`（Users / Notes；Logs 表由服务启动时自建）。
|
||
|
||
## 配置管理
|
||
|
||
db 已接入 [confs](../confs/) 统一配置服务，confs 是 db 启动的**强依赖**：
|
||
|
||
1. **confs 统一配置服务** — 必须成功从 confs 拉取 `db.ini`，否则 db 启动失败直接退出（exit code 1）
|
||
2. **环境变量覆盖**（最高优先级）— Docker/部署环境通过环境变量覆盖 confs 中的值
|
||
|
||
### confs 配置文件（db.ini）
|
||
|
||
配置文件托管于 [confs/configs/db.ini](../confs/configs/db.ini)：
|
||
|
||
```ini
|
||
[server]
|
||
host = 0.0.0.0
|
||
port = 8005
|
||
workers = 4
|
||
|
||
[database]
|
||
url = sqlite:///database.db
|
||
|
||
[logs]
|
||
retention_days = 30
|
||
max_records = 1000000
|
||
cleanup_interval = 3600
|
||
```
|
||
|
||
### 引导配置（confs 连接地址）
|
||
|
||
confs 自身的连接地址不通过 confs 获取（避免循环依赖），通过以下方式确定（优先级从高到低）：
|
||
- 环境变量 `CONFS_HOST` / `CONFS_PORT`
|
||
- 默认值 `127.0.0.1:8010`
|
||
|
||
### 环境变量（最高优先级，覆盖 confs 配置值）
|
||
|
||
| 环境变量 | 说明 | 对应 confs 配置项 |
|
||
|---------|------|------------------|
|
||
| `LISTEN_ADDR` | 监听地址 | `[server] host` |
|
||
| `LISTEN_PORT` | 监听端口 | `[server] port` |
|
||
| `WORKERS` | 回调并发线程数 | `[server] workers` |
|
||
| `DATABASE_URL` | 数据库连接 URL | `[database] url` |
|
||
| `CONFS_HOST` | confs 服务地址（引导配置） | — |
|
||
| `CONFS_PORT` | confs 服务端口（引导配置） | — |
|
||
|
||
### 启动顺序
|
||
|
||
db 通过 pymsg `retry_on_refused=True` 自动阻塞等待 confs 就绪，**不严格要求启动顺序**：
|
||
- confs 先启动 → db 立即拉取配置，正常启动
|
||
- db 先启动 → pymsg 自动重试（每秒一次，每 5 秒打印等待提示），confs 启动后自动连接成功
|
||
- 配置完整性校验：若 db.ini 中缺少必需的 section/key 且无对应环境变量覆盖，仍然报错退出
|
||
- 启动成功时控制台打印：`配置来源: confs (db.ini @ host:port)`
|
||
|
||
> 独立运维工具 `db_tools.py`（backup/export/import/vacuum 等）仍直接读 `DATABASE_URL` 环境变量，不依赖 confs 服务就绪。
|
||
> Docker 环境通过 `docker-entrypoint.sh` 传入环境变量（见 [Dockerfile](Dockerfile)），需确保 `CONFS_HOST`/`CONFS_PORT` 指向 confs 服务地址。
|
||
|
||
## 协议与 API
|
||
|
||
### 传输协议（pymsg）
|
||
|
||
每条消息为一个完整 JSON，帧格式：`[4 字节大端长度][UTF-8 JSON]`。
|
||
请求中的 `_route` 字段指定路由，目前仅有一个路由：**`execute`**。
|
||
|
||
### execute 路由 — 通用 CRUD
|
||
|
||
```json
|
||
{
|
||
"_route": "execute",
|
||
"action": "select", // select | insert | update | delete
|
||
"table": "Users",
|
||
"columns": ["id", "name"], // select 时可选（缺省全列）
|
||
"values": {"name": "Bob"}, // insert / update 时必填
|
||
"where": { // 可选，条件见下表
|
||
"age": {"ge": 18, "lt": 60},
|
||
"name": "Alice"
|
||
},
|
||
"order_by": [{"column": "id", "desc": true}],
|
||
"limit": 10,
|
||
"offset": 0
|
||
}
|
||
```
|
||
|
||
### 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": [
|
||
{"id": 1, "username": "Alice", "create_at": "2026-07-28T14:30:00"}
|
||
]
|
||
}
|
||
```
|
||
- select → 行对象数组
|
||
- insert → `[{"lastrowid": N}]`
|
||
- update / delete → `[{"rowcount": N}]`
|
||
|
||
失败：
|
||
```json
|
||
{"success": false, "error": "错误描述"}
|
||
```
|
||
|
||
## 日期/时间列的处理
|
||
|
||
JSON 没有日期类型，db 内置了类型适配层：按反射得到的列类型自动在
|
||
ISO 8601 字符串与数据库 DATETIME / DATE / TIME 之间转换。
|
||
|
||
**写入（values / where 条件值）**：
|
||
|
||
| 输入 | 说明 |
|
||
|------|------|
|
||
| `"2026-07-28T22:30:00"` | ISO 8601（T 分隔），推荐 |
|
||
| `"2026-07-28 22:30:00"` | 空格分隔同样接受（SQLite 原生格式） |
|
||
| `"2026-07-28T22:30:00+08:00"` | 带时区偏移：自动转 UTC 后存储 |
|
||
| `"2026-07-28T14:30:00Z"` | Z 后缀等同 +00:00 |
|
||
| `"$now"` | 哨兵值：服务端当前 UTC 时间（DATETIME/DATE 列可用） |
|
||
|
||
存储统一为 UTC 朴素时间（无时区）。非法日期字符串会返回明确的错误：
|
||
`{"success": false, "error": "列 last_login 的日期时间值无法解析（需 ISO 8601 或 \"$now\"）: ..."}`
|
||
|
||
**读取（select 响应）**：日期时间值一律序列化为 ISO 8601（T 分隔），
|
||
如 `"2026-07-28T22:30:00"`，客户端可用 `datetime.fromisoformat()` 直接解析。
|
||
|
||
**注意**：`like` 运算符是字符串模式匹配，对日期列使用时不做转换。
|
||
|
||
```json
|
||
// 示例：登录后更新 last_login 为服务端当前时间
|
||
{
|
||
"_route": "execute",
|
||
"action": "update",
|
||
"table": "Users",
|
||
"values": {"last_login": "$now"},
|
||
"where": {"username": "alice"}
|
||
}
|
||
```
|
||
|
||
## 使用示例（Python + pymsg）
|
||
|
||
```python
|
||
import json
|
||
from pymsg import MsgClient
|
||
|
||
client = MsgClient("127.0.0.1", 8005, "myservice")
|
||
|
||
# 查询
|
||
resp = client.call(json.dumps({
|
||
"_route": "execute",
|
||
"action": "select",
|
||
"table": "Users",
|
||
"where": {"id": {"gt": 0}},
|
||
"order_by": [{"column": "id", "desc": True}],
|
||
"limit": 5,
|
||
}))
|
||
print(json.loads(resp))
|
||
|
||
# 插入
|
||
client.call(json.dumps({
|
||
"_route": "execute",
|
||
"action": "insert",
|
||
"table": "Users",
|
||
"values": {"username": "bob", "password_hash": "..."},
|
||
}))
|
||
```
|
||
|
||
## 安全机制
|
||
|
||
- 强制使用 SQLAlchemy 参数绑定，杜绝 SQL 注入。
|
||
- 仅反射已有表，无法创建或删除表（保证结构安全）。
|
||
- 所有异常自动回滚事务。
|
||
- 仅监听 `127.0.0.1`，仅本机服务可访问。
|
||
|
||
## 内置日志能力
|
||
|
||
db 内置统一日志收集：全系统各服务（waf / gateway / static / dnfs / db / ai_review / captcha / core / smtp）的日志集中写入 db 的 `Logs` 表。
|
||
|
||
### 为什么集成在 db 而不是独立服务
|
||
|
||
独立 logs 服务会导致**循环调用**：db 记日志→调 logs，logs 存库→调 db，无限递归。集成在 db 内部后，db 自己记日志直接写表，彻底消除循环依赖。
|
||
|
||
### 日志表（启动时自建）
|
||
|
||
```sql
|
||
CREATE TABLE IF NOT EXISTS Logs (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
ts TEXT NOT NULL,
|
||
service TEXT NOT NULL,
|
||
level TEXT NOT NULL,
|
||
message TEXT NOT NULL,
|
||
trace_id TEXT,
|
||
extra TEXT
|
||
);
|
||
-- 索引: (service,ts) / (level,ts) / (ts) / (trace_id)
|
||
```
|
||
|
||
- `level`：`DEBUG` / `INFO` / `WARNING` / `ERROR` / `CRITICAL`
|
||
- `trace_id`：可选，请求追踪 ID
|
||
- `extra`：可选，JSON 字符串，存放附加结构化字段
|
||
|
||
### 写入方式
|
||
|
||
**外部服务**：复制 `example_log_client.py` 为各自目录下的 `log_client.py`，改 `service_name` 即用。客户端底层通过 pymsg `RpcClient`（高层 RPC 封装）调 `execute` 路由逐条 INSERT 到 Logs 表：
|
||
|
||
```python
|
||
from log_client import logs
|
||
|
||
logs.info("dnfs", "节点新增成功 id=42")
|
||
logs.error("dnfs", "数据库连接失败", extra={"errno": 2002})
|
||
logs.push("gateway", "WARNING", "登录失败次数超限",
|
||
trace_id="req-abc", extra={"ip": "1.2.3.4"})
|
||
```
|
||
|
||
客户端特性：
|
||
- 无第三方依赖（仅依赖 pymsg 项目内置库），异步非阻塞：内存队列 + 后台线程批量推送
|
||
- **RpcClient 高层 RPC**：直接传 dict 对象，无需手动拼 `_route` / json.dumps / json.loads
|
||
- fail-safe：db 不可用时静默丢弃，绝不影响业务；后台线程外层 try/except 捕获所有异常，任何网络/序列化错误都不会杀死日志线程（防止队列无限增长导致 OOM）
|
||
- 缓冲批量：队列满 50 条或超时 2 秒时自动 flush
|
||
- 时间戳准确：日志入队时即生成 UTC ISO 时间戳，而非发送时生成，队列积压时时间仍反映事件真实发生时间
|
||
- 队列保护：队列上限 10000 条，满时丢弃最旧条目（drop-oldest），`push()` 永远不阻塞业务线程
|
||
- 安全序列化：`extra` 字段包含不可 JSON 序列化对象时自动 fallback 为 `str()`/`repr()`，不会因 TypeError 中断
|
||
- flush 可靠：`flush()` 使用 Event 通知而非固定 sleep，精确等待后台线程完成批量发送
|
||
- 长度保护：服务端 `_log_internal()` 对内部日志有 `message` 上限 8192 字节、`extra` 上限 4096 字节的自动截断；外部客户端建议在业务层控制日志长度
|
||
|
||
**db 自身**：`_log_internal()` 直接写库，无网络往返（防递归），并在写入前做长度截断。
|
||
|
||
### 查询日志
|
||
|
||
日志查询直接走 `execute` 路由 select Logs 表，或用工具脚本 `_view_db.sqlite3.py`：
|
||
|
||
```bash
|
||
python3 _view_db.sqlite3.py # 列出所有表及行数
|
||
python3 _view_db.sqlite3.py Logs -n 20 # 最近 20 条日志
|
||
python3 _view_db.sqlite3.py Logs -w "level='ERROR'"
|
||
python3 _view_db.sqlite3.py -s "SELECT service, COUNT(*) FROM Logs GROUP BY service"
|
||
```
|
||
|
||
### 自动清理
|
||
|
||
- 服务启动时开启后台守护线程，每 1 小时清理一次
|
||
- 默认保留 30 天日志，超期自动删除
|
||
- 日志总条数超过 100 万时，按时间倒序保留最新 100 万条
|
||
|
||
## 切换数据库
|
||
|
||
推荐方式：修改 confs 的 [db.ini](../confs/configs/db.ini) 中 `[database] url`，然后通过 confs `reload` 路由重载（或重启 db）。也可通过环境变量 `DATABASE_URL` 覆盖（Docker 部署时常用）：
|
||
|
||
- SQLite（默认）：`sqlite:///database.db`（相对路径）或 `sqlite:////app/data/database.db`（绝对路径，Docker 中使用）
|
||
- PostgreSQL：`postgresql://user:pass@localhost/dbname`（需 `pip install psycopg2-binary`）
|
||
- MySQL：`mysql+pymysql://user:pass@localhost/dbname`（需 `pip install pymysql`）
|
||
|
||
SQLAlchemy 会自动生成适配各数据库的 SQL 方言，无需修改接口调用。
|
||
|
||
## 文件说明
|
||
|
||
| 文件 | 说明 |
|
||
|------|------|
|
||
| `dbv2.py` | 服务主程序（pymsg RpcServer/TCP，唯一在跑的实现） |
|
||
| `example_log_client.py` | 日志客户端 SDK 模板，各服务复制为 `log_client.py` 使用（基于 RpcClient，含线程保护/队列上限/安全序列化等健壮性设计） |
|
||
| `db_tools.py` | SQLite 运维工具：backup / export / import / vacuum / wal-checkpoint / info，供 docker.sh 等脚本调用 |
|
||
| `create_sqlite3.sql` | Users / Notes / Logs 建表脚本 |
|
||
| `_view_db.sqlite3.py` | 非交互式查看 database.db 的工具脚本 |
|
||
| `_set.sh` / `_run.sh` | 环境安装 / 启动脚本 |
|