micro_servers/db

通用数据库服务 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，不对外暴露

快速开始

安装依赖

./_set.sh     # 建 venv 并安装 requirements.txt（SQLAlchemy + 本地 ../pymsg）

启动服务

./_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 是 db 启动的强依赖：

1. confs 统一配置服务 — 必须成功从 confs 拉取 db.ini，否则 db 启动失败直接退出（exit code 1）
2. 环境变量覆盖（最高优先级）— Docker/部署环境通过环境变量覆盖 confs 中的值

confs 配置文件（db.ini）

配置文件托管于 confs/configs/db.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 服务端口（引导配置）	—

启动顺序要求

confs 必须先于 db 启动（启动顺序调整为：confs → db → ai_review/...）。若 confs 不可用：

• db 打印致命错误到 stderr（包含 confs 地址和具体错误原因），以 exit code 1 终止
• 不会静默使用默认值启动
• 配置完整性校验：若 db.ini 中缺少必需的 section/key 且无对应环境变量覆盖，同样报错退出
• 启动成功时控制台打印：配置来源: confs 统一配置服务 (db.ini @ host:port)

独立运维工具 db_tools.py（backup/export/import/vacuum 等）仍直接读 DATABASE_URL 环境变量，不依赖 confs 服务就绪。 Docker 环境通过 docker-entrypoint.sh 传入环境变量（见 Dockerfile），需确保 CONFS_HOST/CONFS_PORT 指向 confs 服务地址。

协议与 API

传输协议（pymsg）

每条消息为一个完整 JSON，帧格式：[4 字节大端长度][UTF-8 JSON]。 请求中的 _route 字段指定路由，目前仅有一个路由：execute。

execute 路由 — 通用 CRUD

{
  "_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 连接。

响应格式

成功：

{
  "success": true,
  "data": [
    {"id": 1, "username": "Alice", "create_at": "2026-07-28T14:30:00"}
  ]
}

• select → 行对象数组
• insert → [{"lastrowid": N}]
• update / delete → [{"rowcount": N}]

失败：

{"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 运算符是字符串模式匹配，对日期列使用时不做转换。

// 示例：登录后更新 last_login 为服务端当前时间
{
  "_route": "execute",
  "action": "update",
  "table": "Users",
  "values": {"last_login": "$now"},
  "where": {"username": "alice"}
}

使用示例（Python + pymsg）

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 自己记日志直接写表，彻底消除循环依赖。

日志表（启动时自建）

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 表：

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：

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 中 [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	环境安装 / 启动脚本