# classg-fast-boot

**Repository Path**: CLASSG-BLOG/classg-fast-boot

## Basic Information

- **Project Name**: classg-fast-boot
- **Description**: classg-fast-boot 是一个基于 FastAPI + SQLAlchemy 2.0 的开箱即用后端脚手架，采用 Schema → API → Service → DAO → Model 分层架构，参考 Java 企业级项目设计。拉代码、配数据库、跑起来，5 分钟启动新项目。
- **Primary Language**: Unknown
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-05-23
- **Last Updated**: 2026-05-23

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

<p align="center">
  <h1 align="center">classg-fast-boot</h1>
  <p align="center">🚀 开箱即用的 FastAPI 分层架构脚手架</p>
  <p align="center">
    SQLAlchemy 2.0 · 软删除 · 企业级日志 · 代码生成 · 统一响应封装 · 多环境配置
  </p>
</p>

---

## 特性

- **分层架构** — Schema → API → Service → DAO → Model，参考 Java 企业级项目设计，职责清晰
- **SQLAlchemy 2.0** — 声明式映射 + scoped_session 线程安全 + 连接池
- **逻辑删除** — Mixin + ORM 事件自动过滤，恢复时绕过事件，批量删除性能优先
- **统一响应封装** — `R.ok()` / `R.fail()` 一行代码搞定，自动携带 request_id 和时间戳
- **SQL 彩色日志** — 参数回填生成可复制 SQL，ANSI 语法高亮，按耗时变色
- **请求日志中间件** — 自动记录请求/响应体，智能处理 JSON/multipart/binary，超长截断
- **代码生成器** — 从数据库表结构一键生成 ORM 模型 + 项目基础设施
- **全局异常处理** — 业务异常、HTTP 异常、参数校验异常、未知异常统一格式
- **Request ID 追踪** — ContextVar 贯穿全链路，类似 Java MDC
- **多环境配置** — 基于 pydantic-settings 的类型安全配置，支持 dev/test/staging/prod 环境切换

## 快速开始

### 环境要求

- Python >= 3.10
- MySQL 5.7+ / 8.0+
- [uv](https://github.com/astral-sh/uv)（推荐）或 pip

### 安装

```bash
git clone https://gitee.com/CLASSG-BLOG/classg-fast-boot.git
cd classg-fast-boot

# 安装依赖
uv sync
# 或
pip install -r requirements.txt
```

### 配置数据库

复制环境变量模板，填入你的数据库连接信息：

```bash
cp .env.example .env
```

编辑 `.env`：

```env
DATABASE_URL=mysql+pymysql://root:your_password@localhost:3306/your_db?charset=utf8mb4
SQL_ECHO=true
LOG_LEVEL=INFO
```

### 多环境配置

项目支持多环境配置，通过 `APP_ENV` 切换环境：

```bash
# 复制对应环境的模板
cp .env.dev.example .env.dev      # 开发环境
cp .env.prod.example .env.prod    # 生产环境
cp .env.test.example .env.test    # 测试环境
```

配置加载优先级（由低到高）：

```
代码默认值 → .env（基础配置） → .env.{APP_ENV}（环境专属） → 系统环境变量
```

| 文件 | 用途 | 是否提交 |
|---|---|---|
| `.env.example` | 基础配置模板 | ✅ 是 |
| `.env.dev.example` | 开发环境模板 | ✅ 是 |
| `.env.prod.example` | 生产环境模板 | ✅ 是 |
| `.env.test.example` | 测试环境模板 | ✅ 是 |
| `.env` / `.env.*` | 实际配置（含密码） | ❌ 否 |

### 启动

```bash
# 默认开发环境
python main.py

# 指定环境启动
python main.py --env prod

# 通过环境变量指定
APP_ENV=prod python main.py

# uvicorn 命令行
uvicorn main:app --reload --host 127.0.0.1 --port 8000
```

访问：
- API 文档：http://127.0.0.1:8000/docs
- 健康检查：http://127.0.0.1:8000/

## 项目结构

```
classg-fast-boot/
├── main.py                 # 应用入口（中间件、路由、异常处理器注册）
├── conf/                   # 配置层
│   ├── settings.py         #   多环境配置管理（基于 pydantic-settings）
│   ├── db_conf.py          #   数据库连接池、Session 工厂、逻辑删除事件、SQL 彩色日志
│   └── log_conf.py         #   日志初始化、请求日志中间件
├── core/                   # 核心层
│   ├── exception.py        #   ClassgException 业务异常定义
│   ├── exception_handler.py#   全局异常处理器（4 类异常统一转 R.fail）
│   └── middleware.py        #   中间件注册（CORS + Request ID + 请求日志）
├── constants/              # 常量层
│   ├── result_code.py      #   HTTP 标准状态码枚举
│   └── business_code.py    #   业务扩展状态码枚举
├── models/                 # ORM 模型层
│   ├── base.py             #   BaseModel（id + create_time + update_time）+ SoftDeleteMixin
│   └── blog/               #   Blog、Tag 模型
├── schemas/                # Pydantic 模型层
│   ├── base.py             #   PaginatedResponse[T] 通用分页响应
│   └── blog/               #   Blog、Tag 的 Create / Update / Response
├── api/                    # API 路由层
│   └── blog/               #   Blog、Tag 路由端点
├── services/               # 业务逻辑层
│   └── blog/               #   Blog、Tag 服务
├── dao/                    # 数据访问层
│   └── blog/               #   Blog、Tag DAO（纯 SQLAlchemy 查询）
├── utils/                  # 工具层
│   ├── R.py                #   统一响应封装（R.ok / R.fail）
│   ├── db_gen.py           #   数据库表检查 & 模型代码生成器
│   └── templates/          #   代码生成模板（base.py / db_conf.py / log_conf.py）
├── examples/               # 示例代码
│   ├── crud_blog.py        #   Blog CRUD 案例
│   └── orm_demo.py         #   SQLAlchemy 高级查询模式案例
├── http/                   # HTTP 接口测试文件（JetBrains HTTP Client）
├── scripts/                # 脚本工具
│   └── sh/
│       ├── cleanup_comments.py #   注释清理脚本
│       └── format_code.py      #   代码格式化脚本
├── .env.example            # 基础环境变量模板
├── .env.dev.example        # 开发环境配置模板
├── .env.prod.example       # 生产环境配置模板
├── .env.test.example       # 测试环境配置模板
└── pyproject.toml          # 项目依赖
```

## 架构说明

### 调用链路

```
请求 → Router(API) → Service → DAO → Model(DB)
              ↓          ↓       ↓
           Schema     校验/编排  纯查询
```

每一层职责分明：

| 层 | 职责 | 不应该做 |
|---|---|---|
| **API** | 参数接收、调用 Service、返回 R 响应 | 写业务逻辑、直接操作 DB |
| **Service** | 业务校验、逻辑编排、抛 ClassgException | 直接写 SQLAlchemy 查询 |
| **DAO** | 纯 SQLAlchemy 查询，返回 ORM 对象或元组 | 做业务判断、抛异常 |
| **Model** | 表映射、字段定义 | 写查询逻辑 |
| **Schema** | 请求校验、响应序列化 | 包含业务逻辑 |

### 统一响应格式

所有接口统一返回以下结构（None 字段自动排除）：

```json
{
  "code": 200,
  "msg": "操作成功",
  "data": { ... },
  "request_id": "uuid",
  "request_time": "2026-05-23 10:30:00"
}
```

```python
# 成功
return R.ok(data=blog, msg="查询成功")

# 失败
return R.fail(code=404, msg="博客不存在")

# 带字段级错误
raise ClassgException(code=400, msg="参数错误", field={"title": "不能为空"})
```

### 逻辑删除

继承 `SoftDeleteMixin` 即可启用，无需手动过滤：

```python
class Blog(Base, BaseModel, SoftDeleteMixin):
    title = Column(String(50))
    # deleted 字段自动拥有
```

- **自动过滤**：所有 SELECT 自动追加 `WHERE deleted = FALSE OR deleted IS NULL`
- **逻辑删除**：`DAO.soft_delete()` 设 `deleted=True`
- **恢复**：`DAO.restore()` 用 Core API 绕过 ORM 事件
- **选择性启用**：不继承 `SoftDeleteMixin` 则不受影响（如 Tag 物理删除）

### 全局异常处理

| 异常类型 | 处理方式 |
|---|---|
| `ClassgException` | 转为 `R.fail(code=exc.code, msg=exc.msg, field=exc.field)` |
| `HTTPException` | 转为 `R.fail(code=exc.status_code, msg=str(exc.detail))` |
| `RequestValidationError` | 解析字段校验错误，转为 `R.fail(code=400, explain=detail)` |
| `Exception`（兜底） | 记录日志，转为 `R.fail(code=500)` |

### SQL 彩色日志

设置 `SQL_ECHO=true` 开启，效果：

```
SELECT blog_blog.id, blog_blog.title FROM blog_blog WHERE blog_blog.deleted = false;
⏱ 12.3ms
```

- 参数回填，生成可直接复制执行的完整 SQL
- 关键字青色、表名.列名黄色、字符串绿色、数字蓝色、NULL 灰色
- 耗时：<50ms 绿色 / <200ms 黄色 / ≥200ms 红色
- Windows 终端 ANSI 自动支持

### 请求日志中间件

自动记录每个请求的：

```
2026-05-23 10:30:00 [INFO] request_logging_middleware - → POST /blogs 200 45.2ms
  Request Body: {"title": "Hello", "content": "World"}
  Response Body: {"code": 200, "msg": "创建成功", "data": {...}}
```

- 自动跳过 `/docs`、`/redoc`、`/openapi.json`
- JSON 格式化、multipart 文件元信息、binary hex 摘要
- 超长截断（`LOG_BODY_MAX_LENGTH`，默认 2048 字节）
- 4xx/5xx 用 WARNING 级别

## API 端点

以 Blog 示例模块展示：

### Blog

| 方法 | 路径 | 说明 |
|---|---|---|
| `GET` | `/blogs` | 分页查询（支持 keyword 模糊搜索） |
| `GET` | `/blogs/{id}` | 获取单篇博客 |
| `POST` | `/blogs` | 创建博客 |
| `PUT` | `/blogs/{id}` | 部分更新博客 |
| `DELETE` | `/blogs/{id}` | 逻辑删除 |
| `POST` | `/blogs/{id}/restore` | 恢复已删除博客 |
| `POST` | `/blogs/batch-delete` | 批量逻辑删除 |

### Tag

| 方法 | 路径 | 说明 |
|---|---|---|
| `GET` | `/tags` | 分页查询 |
| `GET` | `/tags/{id}` | 获取单个标签 |
| `POST` | `/tags` | 创建标签 |
| `PUT` | `/tags/{id}` | 部分更新标签 |
| `DELETE` | `/tags/{id}` | 物理删除 |

## 代码生成器

从已有数据库表结构一键生成 ORM 模型代码：

```bash
# 快速模式（按代码中 TABLES 配置）
python utils/db_gen.py

# 指定表名，自动去前缀生成类名
python utils/db_gen.py blog_blog --strip-prefix blog

# 指定输出路径
python utils/db_gen.py blog_blog -o models/blog/blog.py

# 列出所有表
python utils/db_gen.py --tables

# 手动指定 Boolean 字段（TINYINT(1) 自动识别，其他需手动）
python utils/db_gen.py blog_blog --bool status deleted

# 字段类型覆盖
python utils/db_gen.py blog_blog --type sort:SmallInteger
```

**生成效果**：
- 自动检测并继承 `BaseModel` / `SoftDeleteMixin`
- 自动跳过基类已有字段（id, create_time, update_time）
- 自动检测软删除字段
- TINYINT(1) 自动识别为 Boolean
- 终端带行号和语法高亮预览

首次运行时自动检查并生成项目基础设施（`.env.example`、`conf/db_conf.py`、`conf/log_conf.py`、`models/base.py`），已存在则跳过。

## 脚本工具

### 注释清理（`scripts/sh/cleanup_comments.py`）

删除分隔线，将多行 `#` 注释块转为 `"""` docstring：

```bash
# 清理全项目
python scripts/sh/cleanup_comments.py

# 清理指定文件
python scripts/sh/cleanup_comments.py conf/db_conf.py main.py

# 预览模式（不写入文件）
python scripts/sh/cleanup_comments.py --dry-run
```

### 代码格式化（`scripts/sh/format_code.py`）

基于 ruff 的格式化 + lint 修复：

```bash
# 格式化全项目
python scripts/sh/format_code.py

# 格式化指定目录/文件
python scripts/sh/format_code.py conf/ main.py

# 只检查不修改（CI 用）
python scripts/sh/format_code.py --check

# 只格式化，不修复 lint
python scripts/sh/format_code.py --format-only
```

需要先安装 ruff：`pip install ruff`

## 环境变量

所有配置项均由 `conf/settings.py` 集中管理，基于 `pydantic-settings` 实现类型安全，支持环境变量覆盖。

### 应用配置

| 变量 | 说明 | 默认值 |
|---|---|---|
| `APP_ENV` | 当前环境（dev / test / staging / prod） | `dev` |
| `APP_NAME` | 应用名称 | `classg-fast-boot` |
| `APP_DEBUG` | 调试模式 | `true` |
| `APP_HOST` | 服务监听地址 | `127.0.0.1` |
| `APP_PORT` | 服务监听端口 | `8000` |
| `APP_RELOAD` | 是否开启热重载 | `true` |
| `DOCS_URL` | Swagger UI 路径，设为空字符串禁用 | `/docs` |

### 数据库配置

| 变量 | 说明 | 默认值 |
|---|---|---|
| `DATABASE_URL` | 数据库连接 URL | `mysql+pymysql://root:password@localhost:3306/your_db?charset=utf8mb4` |
| `DB_POOL_SIZE` | 连接池常驻连接数 | `10` |
| `DB_MAX_OVERFLOW` | 超出 pool_size 时最多再创建的连接数 | `20` |
| `DB_POOL_RECYCLE` | 连接存活秒数 | `3600` |
| `SQL_ECHO` | SQL 日志开关（生产务必 false） | `false` |

### 日志配置

| 变量 | 说明 | 默认值 |
|---|---|---|
| `LOG_LEVEL` | 日志级别 | `INFO` |
| `LOG_DIR` | 日志文件目录 | `logs` |
| `LOG_RETAIN_DAYS` | 日志保留天数 | `30` |
| `LOG_BODY_MAX_LENGTH` | 请求/响应体最大记录长度 | `2048` |

### 在代码中使用

```python
from conf.settings import settings

# 读取配置
db_url = settings.DATABASE_URL
if settings.is_prod:
    # 生产环境逻辑
    ...
```

## 新增业务模块

以新增 `Comment` 模块为例：

**1. 创建模型** `models/blog/comment.py`

```python
from models.base import BaseModel, Base

class Comment(Base, BaseModel):
    __tablename__ = 'blog_comment'
    content = Column(Text, comment='评论内容')
    blog_id = Column(String(50), comment='博客id')
```

或用代码生成器：

```bash
python utils/db_gen.py blog_comment --strip-prefix blog -o models/blog/comment.py
```

**2. 创建 Schema** `schemas/blog/comment_schema.py`

```python
from pydantic import BaseModel as PydanticBase

class CommentCreate(PydanticBase):
    content: str
    blog_id: str

class CommentResponse(PydanticBase):
    id: str
    content: str | None
    blog_id: str | None
    model_config = {'from_attributes': True}
```

**3. 创建 DAO** `dao/blog/comment_dao.py`

```python
from models.blog.comment import Comment

class CommentDAO:
    @staticmethod
    def get_list(db, page, size):
        query = db.query(Comment)
        total = query.count()
        items = query.order_by(Comment.create_time.desc()).offset((page-1)*size).limit(size).all()
        return items, total
```

**4. 创建 Service** `services/blog/comment_service.py`

```python
from dao.blog.comment_dao import CommentDAO

class CommentService:
    @staticmethod
    def get_list(db, page, size):
        items, total = CommentDAO.get_list(db, page, size)
        return {"total": total, "page": page, "size": size, "items": items}
```

**5. 创建 API** `api/blog/comment_api.py`

```python
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from conf.db_conf import get_db
from utils.R import R
from services.blog.comment_service import CommentService

router = APIRouter(prefix='/comments', tags=['Comment'])

@router.get('')
def list_comments(page: int = 1, size: int = 20, db: Session = Depends(get_db)):
    data = CommentService.get_list(db, page, size)
    return R.ok(data=data)
```

**6. 注册路由** 在 `main.py` 中：

```python
from api.blog.comment_api import router as comment_router
app.include_router(comment_router)
```

## 技术栈

| 类别 | 技术 |
|---|---|
| Web 框架 | FastAPI |
| ORM | SQLAlchemy 2.0 |
| 数据库驱动 | PyMySQL |
| 数据校验 | Pydantic v2 |
| 配置管理 | pydantic-settings |
| ASGI 服务器 | Uvicorn + WatchFiles |
| 环境变量 | python-dotenv |
| 代码格式化 | Ruff |

## License

MIT