MySQL vs PyMySQL 深度解析：以 elliot-blog 项目为例

📅 2026-06-11 👤 Elliot 👁️ 6 次阅读 ⏱️ 10 分钟阅读 ❤️ 0 赞

MySQL PyMySQL Flask SQLAlchemy 数据库 Python ORM

MySQL vs PyMySQL 深度解析：以 elliot-blog 项目为例

作者: Elliot
日期: 2026-06-11
标签: MySQL, PyMySQL, Flask, SQLAlchemy, 数据库, Python

项目背景

在维护多个项目的过程中，我发现团队对 MySQL 和 PyMySQL 的概念经常混淆。本文以 elliot-blog 项目为实例，详细拆解 MySQL 数据库在 Flask 应用中的运作机制，并彻底厘清 MySQL 与 PyMySQL 的关系。

elliot-blog 项目架构

技术栈

层级	技术	版本	作用
Web框架	Flask	3.0.3	HTTP请求处理
ORM	Flask-SQLAlchemy	3.1.1	对象关系映射
数据库驱动	PyMySQL	1.1.0	Python连接MySQL
数据库	MySQL	8.0+	数据存储
迁移工具	Flask-Migrate	4.0.7	数据库版本管理

文件目录结构

elliot-blog/
├── app/                    # 应用核心代码
│   ├── __init__.py         # Flask应用工厂 + 数据库初始化
│   ├── models.py           # 数据模型定义（ORM层）
│   ├── blog.py             # 博客路由/视图（业务层）
│   ├── auth.py             # 认证路由
│   ├── admin.py            # 后台管理视图
│   ├── rss.py              # RSS功能
│   ├── task.py             # 任务管理
│   └── tools.py            # 工具页面
├── config.py               # 配置文件（含数据库连接URI）
├── run.py                  # 启动入口
├── requirements.txt        # 依赖清单
├── migrations/             # Alembic数据库迁移脚本
├── static/                 # 静态文件（CSS/JS/图片）
├── templates/              # Jinja2 HTML模板
└── venv/                   # Python虚拟环境

MySQL 在项目中的运作方式

四层架构模型

┌─────────────────────────────────────┐
│  第1层：应用层（业务逻辑）            │  ← app/blog.py, app/auth.py
│  BlogPost.query.all()               │
├─────────────────────────────────────┤
│  第2层：ORM层（对象关系映射）          │  ← Flask-SQLAlchemy
│  db.Model, db.Column, db.session    │
├─────────────────────────────────────┤
│  第3层：驱动层（协议通信）             │  ← PyMySQL
│  建立TCP连接、发送SQL、接收结果       │
├─────────────────────────────────────┤
│  第4层：数据库层（数据存储）           │  ← MySQL Server
│  实际存储数据、执行查询、管理事务       │
└─────────────────────────────────────┘

运作流程详解

1. 配置阶段（config.py）

class Config:
    # 数据库连接URI
    SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or \
        'mysql+pymysql://root:password@localhost/elliot_blog'

    # 关闭追踪修改（性能优化）
    SQLALCHEMY_TRACK_MODIFICATIONS = False

URI 格式解析：

mysql+pymysql://root:password@localhost:3306/elliot_blog
│      │         │    │        │        │    │
│      │         │    │        │        │    └── 数据库名
│      │         │    │        │        └─────── 端口（默认3306）
│      │         │    │        └──────────────── 主机地址
│      │         │    └───────────────────────── 密码
│      │         └────────────────────────────── 用户名
│      └──────────────────────────────────────── 驱动名
└─────────────────────────────────────────────── 数据库类型

2. 初始化阶段（app/init.py）

from flask_sqlalchemy import SQLAlchemy

# 创建全局ORM实例（此时未绑定应用）
db = SQLAlchemy()

def create_app(config_name='default'):
    app = Flask(__name__)

    # 加载配置（含数据库URI）
    app.config.from_object(config[config_name])

    # 初始化扩展：将db绑定到当前应用
    db.init_app(app)

    # 应用上下文内创建所有表
    with app.app_context():
        db.create_all()  # 根据models.py中的模型定义，自动建表

3. 模型定义（app/models.py）

from app import db

class BlogPost(db.Model):
    __tablename__ = 'blog_posts'  # 对应MySQL中的表名

    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(200), nullable=False)
    slug = db.Column(db.String(200), unique=True, nullable=False)
    content = db.Column(db.Text, nullable=False)
    summary = db.Column(db.String(500))
    tags = db.Column(db.String(200))
    author = db.Column(db.String(100), default='Elaine')
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    published = db.Column(db.Boolean, default=True)
    view_count = db.Column(db.Integer, default=0)
    like_count = db.Column(db.Integer, default=0)

ORM映射关系：

Python 类/属性	MySQL 表/字段	数据类型
`BlogPost`	`blog_posts`	表
`id`	`id`	`INT AUTO_INCREMENT PRIMARY KEY`
`title`	`title`	`VARCHAR(200)`
`content`	`content`	`TEXT`
`created_at`	`created_at`	`DATETIME`
`published`	`published`	`BOOLEAN`（实际为TINYINT）

4. 业务操作（app/blog.py）

# 查询所有已发布文章（按时间倒序）
posts = BlogPost.query.filter_by(published=True) \
                      .order_by(BlogPost.created_at.desc()) \
                      .all()

# 根据slug查询单篇文章
post = BlogPost.query.filter_by(slug=slug).first_or_404()

# 新增文章
new_post = BlogPost(title='标题', content='内容', slug='url-slug')
db.session.add(new_post)    # 加入会话（未执行）
db.session.commit()          # 提交事务（真正执行INSERT）

# 更新阅读量
post.view_count += 1
db.session.commit()          # 执行UPDATE

核心文件解析

config.py —— 数据库配置中心

import os
from dotenv import load_dotenv

load_dotenv()

class Config:
    SECRET_KEY = os.environ.get('SECRET_KEY') or 'dev-secret-key'

    # 主数据库（生产环境）
    SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or \
        'mysql+pymysql://root:password@localhost/elliot_blog'

    SQLALCHEMY_TRACK_MODIFICATIONS = False

class DevelopmentConfig(Config):
    DEBUG = True
    # 开发环境使用独立数据库
    SQLALCHEMY_DATABASE_URI = os.environ.get('DEV_DATABASE_URL') or \
        'mysql+pymysql://root:password@localhost/elliot_blog_dev'

class ProductionConfig(Config):
    DEBUG = False
    # 生产环境从环境变量读取

class TestingConfig(Config):
    TESTING = True
    # 测试环境使用内存SQLite（无需MySQL）
    SQLALCHEMY_DATABASE_URI = 'sqlite:///:memory:'

config = {
    'development': DevelopmentConfig,
    'production': ProductionConfig,
    'testing': TestingConfig,
    'default': DevelopmentConfig
}

app/init.py —— 应用工厂

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_login import LoginManager

# 全局扩展实例（延迟初始化）
db = SQLAlchemy()
login_manager = LoginManager()

def create_app(config_name='default'):
    app = Flask(__name__)
    app.config.from_object(config[config_name])

    # 初始化扩展
    db.init_app(app)
    login_manager.init_app(app)

    # 注册蓝图
    from app.blog import blog_bp
    app.register_blueprint(blog_bp, url_prefix='/blog')

    # 应用上下文内自动建表
    with app.app_context():
        db.create_all()

    return app

# 用户加载回调
@login_manager.user_loader
def load_user(user_id):
    from app.models import User
    return User.query.get(int(user_id))

app/models.py —— 数据模型

项目包含以下数据模型：

模型	对应表	用途
`User`	`users`	管理员用户
`BlogPost`	`blog_posts`	博客文章
`Task`	`tasks`	任务管理
`RSSFeed`	`rss_feeds`	RSS订阅源
`RSSEntry`	`rss_entries`	RSS文章条目
`FriendshipLink`	`friendship_links`	友情链接
`PortfolioSnapshot`	`portfolio_snapshots`	投资组合快照
`Note`	`notes`	便签

MySQL vs PyMySQL：本质区别

这是两个完全不同的概念，经常混淆：

核心差异

维度	MySQL	PyMySQL
本质	数据库管理系统（DBMS）	Python 数据库驱动/连接器
作用	存储、查询、管理数据	让 Python 程序能连接 MySQL
运行位置	独立进程/服务（本地或远程服务器）	嵌入在 Python 应用程序中
通信协议	服务端，监听3306端口	客户端，实现MySQL客户端协议
安装方式	`apt install mysql-server`	`pip install pymysql`
类比	仓库（存东西的地方）	货车（运输东西的工具）

关系图

┌─────────────────────────────────────────────┐
│              Python 应用程序                  │
│  ┌───────────────────────────────────────┐  │
│  │         Flask 应用代码                 │  │
│  │   BlogPost.query.all()               │  │
│  └───────────────────────────────────────┘  │
│  ┌───────────────────────────────────────┐  │
│  │      Flask-SQLAlchemy (ORM)           │  │
│  │   翻译：Python对象 ↔ SQL语句           │  │
│  └───────────────────────────────────────┘  │
│  ┌───────────────────────────────────────┐  │
│  │         PyMySQL (驱动)                 │  │
│  │   实现MySQL客户端协议                   │  │
│  │   TCP连接、发送SQL、接收结果            │  │
│  └───────────────────────────────────────┘  │
└─────────────────────────────────────────────┘
                    │
                    │ TCP/IP 协议
                    │ (端口3306)
                    ▼
┌─────────────────────────────────────────────┐
│         MySQL Server (数据库服务)            │
│  ┌───────────────────────────────────────┐  │
│  │     查询解析器 → 优化器 → 执行器       │  │
│  └───────────────────────────────────────┘  │
│  ┌───────────────────────────────────────┐  │
│  │         存储引擎（InnoDB）              │  │
│  │      实际存储数据文件                   │  │
│  └───────────────────────────────────────┘  │
└─────────────────────────────────────────────┘

代码层面的体现

requirements.txt 中的依赖关系：

Flask==3.0.3              ← Web框架
Flask-SQLAlchemy==3.1.1   ← ORM工具
PyMySQL==1.1.0            ← 数据库驱动（连接MySQL）
cryptography==42.0.5      ← PyMySQL的依赖（加密连接）

注意： requirements.txt 中没有 mysql-server，因为 MySQL 是系统级服务，不是 Python 包。

为什么需要 PyMySQL？

Python 标准库没有内置 MySQL 连接能力。PyMySQL 实现了 MySQL客户端/服务器协议，让 Python 可以：

建立连接：通过 TCP/IP 连接到 MySQL 服务器（默认端口3306）
身份验证：发送用户名/密码进行登录
发送SQL：将 SQL 语句编码为 MySQL 协议包发送
接收结果：解析 MySQL 返回的二进制协议数据
类型转换：将 MySQL 数据类型映射为 Python 数据类型

其他可选驱动

驱动	实现方式	特点	适用场景
PyMySQL	纯 Python	无需额外依赖，跨平台，易安装	开发/生产首选
`mysqlclient`	C 扩展（libmysqlclient）	性能更好，但需编译环境	高性能需求
`mysql-connector-python`	纯 Python（Oracle官方）	官方维护，功能完整	企业级应用

数据流全景图

以「发布一篇博客文章」为例，数据如何在各层流动：

1. 用户提交表单
   ↓
2. Flask 接收 POST 请求（blog.py）
   ↓
3. 创建 BlogPost 对象（models.py）
   new_post = BlogPost(title='...', content='...')
   ↓
4. SQLAlchemy 转换为 INSERT 语句
   INSERT INTO blog_posts (title, content, slug, ...) VALUES (...)
   ↓
5. PyMySQL 将 SQL 编码为 MySQL 协议包
   ├─ 建立 TCP 连接（localhost:3306）
   ├─ 发送协议握手包
   ├─ 发送认证信息
   └─ 发送 COM_QUERY 命令包
   ↓
6. MySQL Server 接收并执行
   ├─ 查询解析器：解析 SQL 语法
   ├─ 优化器：选择执行计划
   ├─ 执行器：调用 InnoDB 存储引擎
   └─ 写入数据页（内存）→ 刷盘（磁盘）
   ↓
7. MySQL 返回结果包（OK Packet）
   ↓
8. PyMySQL 解码结果包
   ├─ 解析协议头
   └─ 提取 affected_rows, last_insert_id
   ↓
9. SQLAlchemy 更新对象状态
   new_post.id = 52  ← 获取自增ID
   ↓
10. Flask 返回 HTTP 响应
    redirect(url_for('blog.view_post', slug=new_post.slug))

常见问题与最佳实践

1. 连接池配置

# config.py
class Config:
    SQLALCHEMY_ENGINE_OPTIONS = {
        'pool_size': 10,              # 连接池大小
        'max_overflow': 20,          # 超出池大小后的额外连接
        'pool_timeout': 30,          # 获取连接的超时时间
        'pool_recycle': 3600,        # 连接回收时间（防止超时）
    }

2. 环境变量管理敏感信息

# .env 文件（不提交到Git）
SECRET_KEY=your-secret-key-here
DATABASE_URL=mysql+pymysql://blog_user:strong_password@localhost/elliot_blog

# config.py
from dotenv import load_dotenv
load_dotenv()  # 加载 .env 文件

SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL')

3. 数据库迁移（Flask-Migrate）

# 初始化迁移仓库
flask db init

# 创建迁移脚本（模型变更后）
flask db migrate -m "add_user_table"

# 执行迁移（升级数据库）
flask db upgrade

# 回滚迁移
flask db downgrade

4. 性能优化

# 使用索引
class BlogPost(db.Model):
    __tablename__ = 'blog_posts'

    # 为常用查询字段添加索引
    slug = db.Column(db.String(200), unique=True, index=True)
    created_at = db.Column(db.DateTime, default=datetime.utcnow, index=True)
    published = db.Column(db.Boolean, default=True, index=True)

5. 常见错误排查

错误	原因	解决
`ModuleNotFoundError: No module named 'pymysql'`	PyMySQL未安装	`pip install pymysql`
`Can't connect to MySQL server`	MySQL服务未启动	`sudo systemctl start mysql`
`Access denied for user`	用户名/密码错误	检查URI中的凭据
`Unknown database`	数据库不存在	先创建数据库：`CREATE DATABASE elliot_blog;`
`Too many connections`	连接池耗尽	增加 `pool_size` 或检查连接释放

总结

一句话区分

MySQL = 数据库本身（仓库，存数据的）
PyMySQL = Python 连接 MySQL 的驱动（货车，运数据的）

elliot-blog 的 MySQL 架构

Flask 应用
  ├─ Flask-SQLAlchemy（ORM，操作模型对象）
  ├─ PyMySQL（驱动，负责网络通信和协议转换）
  └─ MySQL 8.0（数据库，实际持久化存储）

关键配置回顾

# requirements.txt
Flask-SQLAlchemy==3.1.1   # ORM
PyMySQL==1.1.0            # 驱动

# config.py
SQLALCHEMY_DATABASE_URI = 'mysql+pymysql://user:pass@host/db'
#         │      │         │    │    │      │    │
#         │      │         │    │    │      │    └── 数据库名
#         │      │         │    │    │      └─────── 主机
#         │      │         │    │    └────────────── 密码
#         │      │         │    └─────────────────── 用户名
#         │      │         └──────────────────────── 驱动标识
#         │      └────────────────────────────────── 数据库类型
#         └───────────────────────────────────────── 协议头

理解这个分层架构，对于排查数据库问题、优化性能、以及后续迁移到其他数据库（如 PostgreSQL）都至关重要。

延伸阅读
- SQLAlchemy 官方文档
- PyMySQL GitHub
- Flask-SQLAlchemy 文档

MySQL vs PyMySQL 深度解析：以 elliot-blog 项目为例

目录

项目背景

elliot-blog 项目架构

技术栈

文件目录结构

MySQL 在项目中的运作方式

四层架构模型

运作流程详解

1. 配置阶段（config.py）

2. 初始化阶段（app/init.py）

3. 模型定义（app/models.py）

4. 业务操作（app/blog.py）

核心文件解析

config.py —— 数据库配置中心

app/init.py —— 应用工厂

app/models.py —— 数据模型

MySQL vs PyMySQL：本质区别

核心差异

关系图

代码层面的体现

为什么需要 PyMySQL？

其他可选驱动

数据流全景图

常见问题与最佳实践

1. 连接池配置

2. 环境变量管理敏感信息

3. 数据库迁移（Flask-Migrate）

4. 性能优化

5. 常见错误排查

总结

一句话区分

elliot-blog 的 MySQL 架构

关键配置回顾

📚 相关文章