# SQLAlchemy-FastAPI

**Repository Path**: class_g/SQLAlchemy-FastAPI

## Basic Information

- **Project Name**: SQLAlchemy-FastAPI
- **Description**: SQLAlchemy 代码生成
- **Primary Language**: Python
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

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

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# FastAPI + SQLAlchemy 项目

一个基于 FastAPI 的 Web 后端项目，集成 SQLAlchemy ORM，附带 **数据库模型代码自动生成工具**——直接从 MySQL 表结构生成 Python ORM 模型，省去手写上百行 Column 的体力活。

---

## 项目结构

```
FastAPIProject/
├── main.py              # FastAPI 启动类（组装路由）
├── pyproject.toml       # 项目依赖（uv 管理）
├── http/                # HTTP 测试文件（IntelliJ/PyCharm 可用，按前缀分组）
│   ├── main.http        # 根路由健康检查
│   └── blog/
│       ├── blog.http    # Blog 接口测试（/blogs）
│       └── tag.http     # Tag 接口测试（/tags）
├── conf/                # 配置层（数据库、环境变量等）
│   ├── __init__.py
│   └── db_conf.py       # 数据库连接与会话管理
├── models/              # ORM 模型层（纯表映射）
│   └── blog/
│       ├── blog.py      # blog_blog 表 → Blog 类（含逻辑删除字段 deleted）
│       └── tag.py       # blog_tag 表 → Tag 类
├── dao/                 # DAO 数据访问层（纯 SQLAlchemy 查询）
│   └── blog/
│       ├── blog_dao.py  # Blog 数据访问
│       └── tag_dao.py   # Tag 数据访问
├── services/            # Service 业务逻辑层（校验、编排、抛异常）
│   └── blog/
│       ├── blog_service.py  # Blog 业务逻辑
│       └── tag_service.py   # Tag 业务逻辑
├── schemas/            # Pydantic 请求/响应模型层（入参校验、序列化）
│   ├── base.py        # 通用分页响应 PaginatedResponse
│   └── blog/
│       ├── blog_schema.py  # BlogCreate / BlogUpdate / BlogResponse
│       └── tag_schema.py   # TagCreate / TagUpdate / TagResponse
├── api/                 # API 路由层（入参校验、调用 Service）
│   └── blog/
│       ├── blog_api.py  # Blog 接口
│       └── tag_api.py   # Tag 接口
├── utils/
│   ├── __init__.py      # 导出 DBInspector、generate_model_code、gen_model
│   └── db_gen.py        # ★ 核心工具：表结构检查 & 模型代码生成
└── examples/            # 使用案例
    ├── crud_blog.py     # Blog CRUD 路由（含逻辑删除+恢复）
    └── orm_demo.py       # ORM 查询演示（可直接运行）
```

---

## 环境配置

### 1. 安装依赖

```bash
# 使用 uv（推荐）
uv sync

# 或 pip
pip install fastapi sqlalchemy uvicorn
```

### 2. 数据库连接

数据库连接字符串通过 **环境变量 `DATABASE_URL`** 配置，`conf/db_conf.py` 中内置了默认值作为兜底：

```bash
# Windows PowerShell
$env:DATABASE_URL="mysql+pymysql://root:yourpassword@localhost:3306/your_db?charset=utf8mb4"

# Linux / macOS
export DATABASE_URL="mysql+pymysql://root:yourpassword@localhost:3306/your_db?charset=utf8mb4"
```

不设置则使用内置默认值。**生产环境务必通过环境变量注入，避免密码泄露。**

---

## 启动服务

```bash
python main.py
# 访问 http://127.0.0.1:8000
# API 文档 http://127.0.0.1:8000/docs
```

---

## 核心工具：`db_gen.py` —— 数据库模型代码生成器

你不需要手写 ORM 模型。连上数据库，一条命令搞定。

### 三种使用方式

#### 方式一：快速模式（最常用）

编辑 `utils/db_gen.py` 末尾的 `TABLES` 列表，直接 `python utils/db_gen.py`：

```python
# utils/db_gen.py 末尾
TABLES = [
    {"table": "blog_blog",   "strip_prefix": "blog"},
    {"table": "blog_tag",    "strip_prefix": "blog"},
]
```

```bash
python utils/db_gen.py
# → 遍历 TABLES，自动打印结构 + 生成代码 + 写入 models/ 目录
```

每项可配置：

| 键 | 类型 | 说明 |
|----|------|------|
| `table` | `str` | 表名（**必填**） |
| `strip_prefix` | `str` | 去掉表名前缀，如 `blog` → `blog_blog` 生成 `class Blog` |
| `bool` | `list` | 手动指定 Boolean 字段，如 `["status", "deleted"]` |
| `type` | `dict` | 类型覆盖，如 `{"sort": "SmallInteger"}` |
| `print_only` | `bool` | 仅打印不写入文件，默认 `False` |

> **提示**：带 CLI 参数运行时自动切到命令行模式，不影响 TABLES 配置。

#### 方式二：命令行模式

```bash
# 基本用法：查结构 → 生成模型 → 写入 models/ 目录
python utils/db_gen.py blog_blog --strip-prefix blog

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

# 指定输出目录
python utils/db_gen.py blog_tag -d models

# 只打印不写文件
python utils/db_gen.py blog_tag -p

# 手动指定 Boolean 字段（TINYINT(1) 已自动识别）
python utils/db_gen.py blog_tag --bool status deleted

# 字段类型覆盖
python utils/db_gen.py blog_tag --type sort:SmallInteger status:Boolean

# 去前缀 + 类型覆盖 组合使用
python utils/db_gen.py blog_tag --strip-prefix blog --type status:SmallInteger

# 列出数据库中所有表名
python utils/db_gen.py --tables
```

| 参数 | 简写 | 说明 |
|------|------|------|
| `table` | 位置参数 | 表名，如 `blog_blog`、`per_user` |
| `--output` | `-o` | 输出文件路径，如 `models/my_model.py` |
| `--dir` | `-d` | 输出目录，默认 `models/` |
| `--print-only` | `-p` | 仅打印代码，不写入文件 |
| `--bool` | | Boolean 字段列表，空格分隔 |
| `--type` | | 类型覆盖，`字段名:SQLAlchemy类型` 格式 |
| `--strip-prefix` | | 去掉表名前缀，如 `blog` → `blog_tag` 生成 `class Tag` |
| `--tables` | | 列出数据库中所有表名 |

#### 方式三：Python 代码调用

```python
from utils import DBInspector, generate_model_code, gen_model

# ------ gen_model：一键生成（推荐） ------
# 查结构 + 生成代码 + 写入 models/blog/tag.py
gen_model("blog_tag", strip_prefix="blog")

# 指定输出目录
gen_model("blog_tag", output_dir="models", strip_prefix="blog")

# ------ generate_model_code：灵活控制 ------
# 只打印不写文件
code = generate_model_code("blog_blog", strip_prefix="blog")

# 写入指定路径
generate_model_code(
    "blog_blog",
    output_path="models/blog/blog.py",
    strip_prefix="blog",
    column_types={"status": "Boolean"},
    bool_columns=["is_publish"],
)

# ------ DBInspector：表结构检查 ------
inspector = DBInspector()

# 查看单表结构
info = inspector.inspect_table("blog_tag")
print(info["columns"])   # 字段列表
print(info["indexes"])   # 索引列表

# 查看所有表结构
inspector.inspect_all_tables()

# 工具方法
inspector.get_table_names()      # 所有表名列表
inspector.get_columns("blog_tag")   # 表的所有字段
inspector.get_primary_keys("blog_tag")  # 主键信息
inspector.get_indexes("blog_tag")      # 索引信息
inspector.get_foreign_keys("blog_tag") # 外键信息
```

---

## 智能特性

### 类型自动映射

| MySQL 类型 | → SQLAlchemy Column 类型 |
|------------|--------------------------|
| `VARCHAR(n)` | `String(n)` |
| `LONGTEXT/MEDIUMTEXT/TEXT` | `Text` |
| `INTEGER/INT` | `Integer` |
| `BIGINT` | `BigInteger` |
| `SMALLINT` | `SmallInteger` |
| `TINYINT` | `Integer`（常规），`Boolean`（TINYINT(1) 自动识别） |
| `FLOAT/DOUBLE` | `Float` |
| `DECIMAL/NUMERIC` | `Numeric` |
| `DATETIME/TIMESTAMP` | `DateTime` |
| `DATE` | `Date` |
| `BOOLEAN` | `Boolean` |
| `JSON` | `JSON` |
| `BLOB` | `LargeBinary` |
| `ENUM` | `Enum` |

> `TINYINT(1)` 自动识别为 `Boolean`，无需手动指定。其他 `TINYINT(n)` 保持 `Integer`。

### 表前缀自动剥离

输入 `--strip-prefix blog` 时：

| 表名 | 类名 | 输出路径 |
|------|------|----------|
| `blog_blog` | `Blog` | `models/blog/blog.py` |
| `blog_tag` | `Tag` | `models/blog/tag.py` |

不带前缀时，直接 `models/表名.py`。

### 逻辑删除自动过滤

`conf/db_conf.py` 内置了 `before_compile` 事件——只要模型定义了 `deleted = Column(Boolean)`，**所有 SELECT 查询自动追加 `WHERE deleted = False`**，你不再需要手动写 `.filter(deleted == False)`。

```python
# ✅ 正常查询——自动过滤已删除数据
db.query(Blog).all()  # SQL: SELECT ... WHERE blog_blog.deleted = 0
```

> 恢复已删除数据时，DAO 层使用 Core API（`db.execute(update(Blog)...)`）绕过 Query 事件。
> Tag 等没有 `deleted` 字段的表不受影响，照常查询。

### 终端颜色标识

表结构预览中，字段名按特征着色：

| 颜色 | 含义 |
|------|------|
| **紫色加粗** | 主键字段 |
| **红色加粗** | 逻辑删除字段（`is_deleted`、`deleted_at` 等） |
| **绿色** | Boolean 字段 |
| **暗粉色** | 时间类型字段（`DATETIME`、`DATE`、`TIMESTAMP`） |

---

## 生成代码示例

输入：

```bash
python utils/db_gen.py blog_blog --strip-prefix blog
```

生成的 `models/blog/blog.py`：

```python
from sqlalchemy import Boolean, Column, DateTime, Integer, String, Text
from conf.db_conf import Base


class Blog(Base):
    __tablename__ = "blog_blog"

    id = Column(String(50), primary_key=True, nullable=False, comment="唯一id")
    title = Column(String(50), comment="博客标题")
    summary = Column(String(50), comment="博客简介")
    content = Column(Text, comment="博客内容")
    click_count = Column(Integer, comment="博客点击数")
    picture_url = Column(Text, comment="标题图片")
    tag_ids = Column(String(255), comment="标签id")
    sort_id = Column(String(50), comment="分类id")
    is_original = Column(Boolean, nullable=False, comment="是否原创")
    author = Column(String(50), comment="作者")
    articles_part = Column(String(50), comment="文章来源")
    is_publish = Column(Boolean, comment="是否发布")
    start_comment = Column(Boolean, comment="是否开启评论")
    level = Column(String(50), nullable=False, comment="推荐等级")
    sort = Column(Integer, nullable=False, comment="排序字段")
    status = Column(Boolean, comment="状态")
    user_id = Column(String(50), comment="用户id")
    create_time = Column(DateTime, comment="创建时间")
    update_time = Column(DateTime, comment="更新时间")

    def __repr__(self):
        return f"<Blog(id={self.id!r}, title={self.title!r}, ...)>"
```

`__repr__` 包含**表中所有字段**，打印对象时一目了然。

---

## 在 FastAPI 中使用生成的模型

`main.py` 已包含完整的 CRUD 示例，核心模式是 **`Depends(get_db)` 依赖注入**——每个请求获取独立 Session，请求结束自动释放：

```python
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from conf.db_conf import get_db
from models.blog.blog import Blog

app = FastAPI()

@app.get('/blogs')
def list_blogs(page: int = 1, size: int = 20, db: Session = Depends(get_db)):
    """分页查询（逻辑删除自动过滤，无需手动加 .filter(deleted == False)）"""
    q = db.query(Blog)
    total = q.count()
    items = q.order_by(Blog.create_time.desc()).offset((page - 1) * size).limit(size).all()
    return {'total': total, 'page': page, 'size': size, 'items': items}

@app.get('/blogs/{blog_id}')
def get_blog(blog_id: str, db: Session = Depends(get_db)):
    """单条查询（自动过滤已逻辑删除），找不到抛 404"""
    blog = db.query(Blog).filter(Blog.id == blog_id).first()
    if not blog:
        raise HTTPException(status_code=404, detail='博客不存在')
    print(blog)  # → <Blog(id='xxx', title='My Blog', ...)>
    return blog

@app.post('/blogs', status_code=201)
def create_blog(title: str, content: str = '', db: Session = Depends(get_db)):
    """创建博客"""
    import uuid
    blog = Blog(id=str(uuid.uuid4()), title=title, content=content)
    db.add(blog)
    db.commit()
    db.refresh(blog)
    return blog

@app.put('/blogs/{blog_id}')
def update_blog(blog_id: str, title: str | None = None, db: Session = Depends(get_db)):
    """部分更新：只改传入字段"""
    blog = db.query(Blog).filter(Blog.id == blog_id).first()
    if not blog:
        raise HTTPException(404, detail='博客不存在')
    if title is not None:
        blog.title = title
    db.commit()
    db.refresh(blog)
    return blog

@app.delete('/blogs/{blog_id}', status_code=204)
def delete_blog(blog_id: str, db: Session = Depends(get_db)):
    """逻辑删除：标记 deleted=True，数据不丢失"""
    blog = db.query(Blog).filter(Blog.id == blog_id).first()
    if not blog:
        raise HTTPException(404, detail='博客不存在')
    blog.deleted = True   # 不真删，打标记
    db.commit()
```

> **为什么不用 `db_session` 直接调用？**  
> `db_session` 是全局变量，直接从路由函数调用存在两个隐患：
> 1. 忘记 `db_session.remove()` → 内存泄漏  
> 2. 并发场景下 session 被交叉污染  
> `Depends(get_db)` 利用 FastAPI 的依赖注入，`finally` 块在你请求结束后自动释放，安全又省心。

### 更多案例

| 文件 | 内容 |
|------|------|
| `examples/crud_blog.py` | 完整的 Blog CRUD 路由（分页/搜索/批量删除/恢复），可直接挂载到 FastAPI |
| `examples/orm_demo.py` | 独立运行的高级查询演示（含 Tag/Blog 聚合统计、事务、批量操作） |
| `main.py` | 项目入口，包含 Blog + Tag 的完整 CRUD 端点 |

```bash
# 直接运行查询演示（无需启动 FastAPI）
python examples/orm_demo.py
```

---

## 常见问题

**Q：生成后的模型没有生效？**
确保 `models/` 目录下有 `__init__.py`（若无则创建），然后重启 Python 进程。

**Q：新增了表，如何快速重新生成？**
在 `TABLES` 列表里加一条，再跑 `python utils/db_gen.py`。如果是 CLI 模式，直接 `python utils/db_gen.py 新表名 --strip-prefix xxx`。

**Q：怎么连接其他数据库？**
设置环境变量 `DATABASE_URL`，或在代码调用时传 `database_url` 参数：
```python
gen_model("my_table", database_url="postgresql://user:pass@host/db")
```
> 目前类型映射针对 MySQL 优化，PostgreSQL 需微调 `TYPE_MAPPING`。

**Q：Boolean 字段没识别对？**
TINYINT(1) 自动识别，其他情况用 `--bool 字段名` 或 `column_types={"字段名": "Boolean"}` 手动指定。