Reglow Python 后端基础库

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

概述

reglow 是基于 FastAPI + SQLAlchemy 2.0（async）的全栈后台基础库，提供 19 个即用业务模块，通过 pip install -e . 安装后可直接导入使用。

目录结构

reglow/
├── core/                    # 框架核心层
│   ├── config.py            # Settings 配置（pydantic-settings，自动读 .env）
│   ├── database.py          # 异步引擎/会话/Base/AutoBigInt/Mixin
│   ├── security.py          # 密码哈希、JWT 签发/验证
│   ├── dependencies.py      # get_current_employee / PermissionChecker
│   ├── redis_client.py      # Redis 客户端
│   ├── cache.py             # 缓存封装
│   ├── rate_limit.py        # 限流
│   ├── sms.py               # 短信发送
│   ├── observability.py     # 健康检查路由
│   └── logging_config.py    # 日志配置
│
├── common/                  # 跨模块共享层
│   ├── response.py          # ApiResponse[T] / PageData[T] 统一响应
│   ├── exceptions.py        # AppException / ErrorCode 枚举
│   ├── exception_handler.py # 全局异常处理注册
│   ├── middleware.py        # 中间件注册（CORS/日志/i18n）
│   └── i18n.py              # 国际化（Accept-Language 头）
│
└── modules/                 # 业务模块层（19 个模块）
    ├── auth/                # 认证（登录/注册/验证码/Token刷新）
    ├── employee/            # 员工管理
    ├── user/                # 用户管理（注册用户，区别于员工）
    ├── role/                # 角色管理（含数据范围）
    ├── menu/                # 菜单管理（树形）
    ├── dept/                # 部门管理（树形）
    ├── post/                # 岗位管理
    ├── dict/                # 字典管理（类型+数据）
    ├── config/              # 参数配置
    ├── log/                 # 日志（登录+操作）
    ├── material/            # 素材中心（图片/视频/文件）
    ├── agreement/           # 协议管理
    ├── notice/              # 通知公告
    ├── message/             # 系统消息+模板+收件箱
    ├── article/             # 文章管理（含分类+文集）
    ├── photo/               # 照片管理（含分类+相册）
    ├── video/               # 视频管理（含分类+视频集）
    ├── ai/                  # AI（对话/图片生成/视频生成）
    └── customer_service/    # 智能客服（知识库/会话/SDK）

模块六层架构

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

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）

安装

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

使用

# 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("操作失败")

异常处理

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.3.1
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=./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=

关于 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,
)

参考文档

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