Reglow Python 后端基础库

包名：reglow | 版本：0.8.0 | Python：>=3.12

概述

reglow 是基于 FastAPI + SQLAlchemy 2.0（async）的全栈后台基础库，提供 27 个即用模块（21 个基础模块 + 6 个 Python 先行内核模块：billing / open_api / tenant / workflow / region / security），通过 pip install -e . 安装后可直接导入使用；也支持开箱即用的自托管联调服务（reglow-serve）。

AI-Native 生成能力（Schema DSL / 生成引擎 / Agent 运行时 / 评测集）自 0.6.0 起迁出基础库，位于同仓库独立发行包 reglow-agent（Python-only）。基础库不含该部分，以保证「基础包四语言同构」。

目录结构

reglow/
├── core/                    # 框架核心层
│   ├── config.py            # Settings 配置（pydantic-settings，自动读 .env，生产安全强校验）
│   ├── database.py          # 异步引擎/会话/Base/AutoBigInt/Mixin（SQLite / MySQL）
│   ├── security.py          # 密码哈希、JWT 签发/验证
│   ├── dependencies.py      # get_current_employee / PermissionChecker / 客户端依赖
│   ├── data_scope.py        # 5 档数据权限（本人/本部门/部门及以下/自定义/全部）
│   ├── session_service.py   # 会话管理（Redis 可用时多端互斥，降级纯 JWT）
│   ├── redis_client.py      # Redis 客户端（熔断 + 降级）
│   ├── cache.py             # 缓存封装
│   ├── rate_limit.py        # 限流
│   ├── sms.py               # 短信发送（阿里云/腾讯云/华为云/Mock）
│   ├── storage.py           # 存储（local / S3，CDN 前缀）
│   ├── email.py             # 邮件（SMTP / SSL / STARTTLS）
│   ├── payment.py           # 支付渠道统一封装（微信/支付宝/银联，签名/回调/退款）
│   ├── wechat.py            # 微信开放能力（jscode2session / 小程序码 / access_token）
│   ├── observability.py     # 请求指标（Prometheus）与健康检查
│   ├── plugin.py            # 插件 / 路由自动发现
│   ├── protocols.py         # ModelRegistry（模块间解耦）
│   └── logging_config.py    # 结构化 JSON 日志（轮转 + TraceID）
│
├── common/                  # 跨模块共享层
│   ├── response.py          # ApiResponse[T] / PageData[T] 统一响应
│   ├── sort.py              # 列表统一服务端排序（sort_by / sort_order）
│   ├── exceptions.py        # AppException / ErrorCode 枚举（按模块分段）
│   ├── exception_handler.py # 全局异常处理注册
│   ├── middleware.py        # 中间件注册（CORS/日志/TraceID/操作日志/流式透传）
│   ├── i18n.py              # 国际化（Accept-Language 头）
│   └── utils/               # 日期/IP/UA 解析/字符串工具
│
└── modules/                 # 基础模块层（27 个模块包，六层架构）
    ├── 认证与组织
    │   ├── auth/                # 认证（登录/注册/验证码/滑动拼图/Token刷新/找回密码）
    │   ├── employee/            # 员工管理（部门/角色多关联、坐席同步）
    │   ├── user/                # 用户管理（注册用户，登录/资料/安全，区别于员工）
    │   ├── role/                # 角色管理（数据范围 1-5、部门授权）
    │   ├── menu/                # 菜单管理（树形）
    │   ├── dept/                # 部门管理（树形 + 成员）
    │   └── post/                # 岗位管理
    ├── 基础配置
    │   ├── dict/                # 字典管理（类型+数据，枚举自动建字典）
    │   ├── config/              # 参数配置 / 系统设置
    │   ├── region/              # 行政区划（六级，pcas.json 种子懒加载）
    │   ├── log/                 # 日志（登录 + 操作，操作日志中间件自动记录）
    │   └── message/             # 系统消息 + 模板 + 收件箱（站内信）
    ├── 内容与媒体
    │   ├── material/            # 素材中心（图片/视频/文件，分组/回收站/AI 清理）
    │   ├── article/             # 文章管理（分类 + 文集）
    │   ├── photo/               # 照片管理（分类 + 相册，管理端 + 用户端）
    │   ├── video/               # 视频管理（分类 + 视频集）
    │   ├── notice/              # 通知公告（发布/撤回/已读）
    │   ├── agreement/           # 协议管理（版本化 + 用户确认 + 站内信提醒）
    │   └── customer_service/    # 智能客服（知识库/会话/坐席/WebSocket/文档解析/SDK 生成）
    ├── 商业化与开放
    │   ├── payment/             # 支付订单（下单/回调/退款/超时闭合）
    │   ├── wechat/              # 微信（网页授权/小程序）
    │   ├── billing/             # 计费（免费/会员/按量，BILLING_ENABLED 开关）
    │   ├── open_api/            # 开放 API（HMAC-SHA256 签名 + 防重放 + 每日配额）
    │   └── tenant/              # 多租户（single_tenant / saas_multi_tenant 双轨）
    └── 平台与合规
        ├── ai/                  # AI 多模型适配（deepseek/aliyun/volcano/tencent/zhipu，chat/stream/深思考/联网/图片/视频）
        ├── workflow/            # 工作流（JSON DSL 节点/条件/审批链）
        └── security/            # 安全增强（审计防篡改哈希链 / 三员模板 / 国密 SM2/SM3/SM4 抽象）

AI-Native 生成能力已迁出为基础库之外的独立发行包 reglow-agent（Python-only，同仓库： reglow-python/reglow-agent/）。原 schema_dsl / agent / agent_runtime / eval / gen_templates 均在该包内，由应用侧调用 reglow_agent.register(app) 挂载。 基础库不含该部分——这是「基础包四语言同构」的前提。

例外：core/expr.py（白名单表达式求值器）是生成产物的运行时依赖，因此留在基础库。

模块六层架构

每个模块遵循统一的分层结构，依赖单向向下：

model.py        ORM 实体（继承 Base + Mixin）
    ↑
schema.py       Pydantic 契约（CreateRequest / UpdateRequest / Response）
    ↑
repository.py   数据访问层（AsyncSession，find_/create_/update_）
    ↑
service.py      业务逻辑层（编排 repository，抛 AppException）
    ↑
controller.py   接口控制层（APIRouter + 权限校验 + ApiResponse）
    ↑
router.py       路由导出（from .controller import router）

AI-Native 生成能力（已迁至 reglow-agent）

以下能力自 0.6.0 起迁出基础库，位于同仓库独立发行包 reglow-agent（Python-only，reglow-python/reglow-agent/）。 基础库不含该部分——这是「基础包四语言同构」的前提。

• Schema DSL v1：声明式模块 Schema（module / label / fields / permissions），13 种字段类型（string/text/int/bigint/float/decimal/bool/datetime/date/enum/json/file/relation），enum 自动建字典，聚合校验（保留字段/模块）。
• 生成引擎 L1：Schema → 后端六层（model/schema/repository/service/controller/router）+ Alembic 迁移 + 菜单/字典种子 + 前端（Vue3 PageTable 页面 + api.ts），一次生成模块两端可运行产物。
• 需求对话 L2：自然语言需求 → 结构化 Schema DSL（requirement_analyzer，复用 ai 模块多模型通道），支持澄清追问与迭代修正。
• Skill 体系 4 套：reglow-backend / reglow-frontend / reglow-deploy / reglow-business（reglow-agent/.agents/skills/）。
• 评测集 v1：golden 10 场景自动评分（schema 20 / artifacts 20 / compile 30 / ruff 30），CLI：python -m reglow_agent.eval.cli run --work-dir ...，通过率 ≥80% 为门槛。

安装与挂载：

pip install -e ./reglow-agent

import reglow_agent
from reglow.server import app

# 挂载 /agent/* 与 /schema/validate 等路由，并注册本包 ORM 模型（须在 create_all 之前）
reglow_agent.register(app)

例外：reglow.core.expr（白名单表达式求值器）是生成产物的运行时依赖——生成的代码 import 它而不是内联逻辑，因此它留在基础库，不随生成引擎一起迁出。

安装

# 可编辑模式（开发）
cd reglow && pip install -e .

自托管联调服务（开箱即用）

reglow 既能作为基础库发布安装，也可以直接启动一个完整的后台服务， 作为 reglow-design 等前端项目的联调后端：

reglow-serve                        # 控制台命令（pip 安装后可用），默认 0.0.0.0:8000
reglow-serve --port 9000 --reload   # 自定义端口 + 热重载
python -m reglow.server             # 等价方式
uvicorn reglow.server:app --reload  # 以模块级 app 方式启动

首次启动自动完成（幂等，再次启动跳过）：

• 自动建表（Base.metadata.create_all，覆盖全部模块的表）
• 自动播种初始化数据（菜单树、角色、部门、岗位、字典、消息模板、协议、系统参数等）
• 自动注册全部模块路由（管理端 /admin/api/v1 + 客户端 /api/v1 + 开放端）+ 健康检查 + /uploads 静态服务

默认账号：超级管理员 rootadmin / admin123、演示账号 demo / demo123。

配置：数据库 / Redis / CORS / JWT 等通过 .env 或环境变量配置（字段见下文）， 未配置时默认以 DEBUG 模式运行，无需任何前置环境（DEBUG 模式下文档 /docs 可用）。

# 联调常用配置（.env）
DB_DRIVER=sqlite                    # 无 MySQL 时用 SQLite 即可，零依赖启动
DEBUG=true
CORS_ORIGINS=http://localhost:5173  # reglow-design 等前端的 origin

使用

# main.py — 消费方应用入口
from fastapi import FastAPI, APIRouter
from reglow.common.middleware import register_middlewares
from reglow.common.exception_handler import register_exception_handlers

# 导入需要的模块路由
from reglow.modules.auth.router import router as auth_router
from reglow.modules.employee.router import router as employee_router
# ... 按需导入

app = FastAPI(title="My App")
register_middlewares(app)
register_exception_handlers(app)

admin_api = APIRouter(prefix="/admin/api/v1")
admin_api.include_router(auth_router)
admin_api.include_router(employee_router)
app.include_router(admin_api)

核心基础设施

统一响应

from reglow.common.response import ApiResponse, PageData

# 成功
return ApiResponse.success(data)
return ApiResponse.success(PageData(items=..., total=..., page=1, size=20))

# 失败
return ApiResponse.fail("操作失败")

列表服务端排序（sort_by / sort_order）

管理端分页列表接口统一支持两个可选查询参数（FastAPI 自动写入 /docs 的 OpenAPI 文档），支持多列排序：

生效规则

• 可排序列 = 该实体（表）在 SQLAlchemy 模型中自有的列（不含关联/join 出来的列），并排除敏感列：password、password_hash、salt、token、refresh_token、access_token、secret、app_secret、api_key、private_key。
• 逐项校验：每项先按上述白名单校验；非法 / 敏感 / 未知字段只跳过该项，其余合法项照常生效，不报错、不返回 400。
• 超过 3 项的尾部直接忽略（防 ORDER BY 滥用）。
• 全非法则整体回落：一个合法项都没有时（含 sort_by 缺失 / 空串），完全忽略排序参数，接口保持原有默认排序，SQL 与响应与不传参完全一致。
• 生效时：合法项按书写顺序依次作为主 → 次 → 末排序，最后追加该接口原有默认排序链，保证分页稳定、结果确定。
• 单列时（如 ?sort_by=created_at&sort_order=desc）SQL 与响应与旧版完全一致。
• 用户输入不会被直接拼入 ORDER BY：每一项先解析为真实列对象（getattr(Model, name) 且校验 name in Model.__table__.columns）再交给 SQLAlchemy。

示例

# 用户列表按注册时间倒序（旧写法，单列，行为不变）
GET /admin/api/v1/users/?page=1&size=20&sort_by=created_at&sort_order=desc

# 字典类型按名称升序（sort_order 省略，默认 asc）
GET /admin/api/v1/dict-types/?page=1&size=10&sort_by=name

# 多列：昵称升序为主、注册时间倒序为次，末尾追加原默认排序（分页稳定）
GET /admin/api/v1/users/?sort_by=nickname:asc,created_at:desc

# 混合省略方向：name 用 sort_order（asc）、created_at 显式 desc
GET /admin/api/v1/users/?sort_by=name,created_at:desc&sort_order=asc

# 含敏感列 / 非法列：只跳过该项，其余合法项照常生效（不报错）
GET /admin/api/v1/users/?sort_by=password_hash,nickname:desc

# 全部非法：整体回落该接口原有默认排序（不报错）
GET /admin/api/v1/users/?sort_by=not_a_column
GET /admin/api/v1/users/?sort_by=password_hash

# 超过 3 项：只取前 3 项，尾部忽略
GET /admin/api/v1/users/?sort_by=nickname,status,created_at,vip_level

实现见 reglow/common/sort.py（parse_sort_items / resolve_sort_column / resolve_sort_order / apply_sort），各列表接口的 repository 通过 apply_sort(query, Model, sort_by, sort_order, default_order) 作为唯一入口接入。

异常处理

from reglow.common.exceptions import AppException, ErrorCode

raise AppException(ErrorCode.NOT_FOUND, "留言不存在", 404)
raise AppException(ErrorCode.BAD_REQUEST, "参数错误", 400)

权限校验

from reglow.core.dependencies import PermissionChecker, get_current_employee

@router.get("/feedbacks", dependencies=[Depends(PermissionChecker("business:feedback:list"))])
async def list_feedbacks(db: AsyncSession = Depends(get_db)):
    ...

数据库模型

from reglow.core.database import AutoBigInt, Base, TimestampMixin, SoftDeleteMixin
from sqlalchemy import Column, String

class Feedback(TimestampMixin, SoftDeleteMixin, Base):
    __tablename__ = "biz_feedback"
    id = Column(AutoBigInt, primary_key=True, autoincrement=True)
    name = Column(String(100), nullable=False, comment="名称")

配置（.env）

所有配置由 reglow.core.config.Settings（pydantic-settings）自动从项目根目录的 .env 读取，带类型校验。未设置的项使用默认值。完整字段如下：

# ===== Application =====
APP_NAME=ReglowAdmin
APP_VERSION=0.8.0
DEBUG=False                    # 生产环境必须为 False；为 False 时强制校验 JWT_SECRET 与 CORS_ORIGINS

# ===== Database =====
DB_DRIVER=mysql                # mysql 或 sqlite
# --- mysql 模式 ---
DB_HOST=localhost
DB_PORT=3306
DB_USER=reglow
DB_PASSWORD=Reglow@2024#Secure   # 直接写明文，特殊字符（@ # : / 等）会自动 URL 编码后拼入连接串
DB_NAME=reglow_admin
# --- sqlite 模式（DB_DRIVER=sqlite 时使用，忽略上面的 host/port/user/password）---
DB_SQLITE_PATH=./db/reglow_admin.db

# ===== Redis =====
REDIS_URL=redis://localhost:6379/0

# ===== JWT =====
JWT_SECRET=change-me-in-production   # DEBUG=False 时必须改；否则启动报错
JWT_ALGORITHM=HS256
JWT_ACCESS_EXPIRE_MINUTES=120        # access token 有效期（分钟）
JWT_REFRESH_EXPIRE_MINUTES=10080     # refresh token 有效期（分钟，默认 7 天）

# ===== CORS =====
# 逗号分隔，每个 origin 必须带 scheme（http:// 或 https://），否则浏览器无法匹配
CORS_ORIGINS=http://localhost:5173,http://localhost:8080,http://localhost:3000

# ===== Logging =====
# 注意：需在应用入口调用 setup_logging() 才会生效，否则走 Python 默认日志
LOG_LEVEL=INFO                 # DEBUG / INFO / WARNING / ERROR
LOG_DIR=./logs                 # 留空则不写文件，仅控制台
LOG_FILE_MAX_BYTES=52428800    # 50MB，单文件轮转阈值
LOG_FILE_BACKUP_COUNT=7        # 保留历史文件数
LOG_JSON=True                  # True=结构化 JSON 日志（便于 ELK/Loki），False=纯文本

# ===== Upload =====
UPLOAD_DIR=./uploads
MAX_UPLOAD_SIZE=1073741824     # 1024MB，multipart 单文件上限

# ===== Storage =====
STORAGE_BACKEND=local          # local 或 s3
# --- s3 模式 ---
S3_ENDPOINT=
S3_ACCESS_KEY=
S3_SECRET_KEY=
S3_BUCKET=
S3_REGION=
CDN_DOMAIN=

# ===== 部署模式（AI-Native）=====
DEPLOY_MODE=single_tenant      # single_tenant / saas_multi_tenant
TENANT_GUARD=enforce           # 多租户错配自检逃生阀：enforce（默认，库里检出 tenant_id 非空数据且未实现隔离即拒绝启动）/ off（仅 WARN，不阻断，禁止静默）
BILLING_ENABLED=false          # 计费开关（saas_multi_tenant + true 时启用准入门控）
SECURITY_CRYPTO_PROVIDER=sm    # 国密抽象（sm / sha256 兼容实现）

多租户错配自检（tenant mismatch guard）

服务启动时（DB 可连接之后、开始监听之前）自动执行（reglow/modules/tenant/guard.py，契约见 reglow-library/contract/tenant.json 的 startupGuard）：扫描当前库中所有含 tenant_id 列的表， 逐表探测是否存在 tenant_id IS NOT NULL 的数据。

逃生阀 TENANT_GUARD=off：仅输出 WARN、不阻断（用于「确认在单租户下运行、库里仅是历史残留 tenant_id 数据」的部署，避免升级即无法启动）；逃生阀必须留痕，不做静默放行。

关于 DB_PASSWORD 的特殊字符

database_url / database_url_sync 会把 DB_USER、DB_PASSWORD 拼进 SQLAlchemy 连接 URL（mysql+aiomysql://{user}:{password}@{host}:{port}/{db}）。库内部已用 urllib.parse.quote_plus 对二者做 percent-encode，因此 .env 里直接写明文密码即可，含 @、#、:、/ 等保留字符也无须手动转义。SQLAlchemy 解析 URL 时会自动解码回原始密码传给驱动。

生产环境强校验

DEBUG=False 时，Settings 的 validate_production_security 校验器会：

• 拒绝默认 JWT_SECRET=change-me-in-production，必须通过环境变量注入强密钥；
• 拒绝 CORS_ORIGINS 含通配符 *。

日志初始化

LOG_* 配置不会自动生效，必须在应用入口（main.py）显式调用：

from reglow.core.logging_config import setup_logging
from reglow.core.config import get_settings

settings = get_settings()
setup_logging(
    level=settings.LOG_LEVEL,
    log_dir=settings.LOG_DIR,
    max_bytes=settings.LOG_FILE_MAX_BYTES,
    backup_count=settings.LOG_FILE_BACKUP_COUNT,
    use_json=settings.LOG_JSON,
)

质量状态

• 全量 1425 个测试用例通过（tests/，SQLite 内存库）
• 代码覆盖率 97%（coverage，concurrency=["thread"] 配置以正确追踪 pytest-asyncio）
• ruff 静态检查通过（代码质量门禁）

参考文档

• Reglow 基础库改造与集成手册
• reglow-app/backend 示例项目