菜鸟教程 – 学的不仅是技术，更是梦想！

技术网站，更是交流网站

Python 3 教程

【超详细教程】Python 连接 MySQL 全流程实战

likuolei

【超详细教程】Python 连接 MySQL 全流程实战（2026最新版）

Python + MySQL 是 Web 后端、数据分析、爬虫、AI 项目中最经典的组合。无论你是新手想快速上手，还是老鸟想升级生产级方案，这篇教程都能一站式解决，从 0 到生产可用。

基于 2026 年 3 月 最新情况（Python 3.13+、MySQL 8.4/9.0、PyMySQL 1.1.2、mysql-connector-python 9.6.0、SQLAlchemy 2.1+），一步步带你走完全流程。

1. 驱动选择对比（2026 年主流推荐）

驱动库类型安装命令优点缺点推荐场景
PyMySQL (1.1.2)纯 Pythonpip install PyMySQL无 C 依赖、安装简单、兼容 MySQLdb纯 Python 略慢新手首选、脚本、学习
mysql-connector-python (9.6.0)官方纯 Pythonpip install mysql-connector-python官方维护、特性最全、MySQL 8.4+ 优化好体积稍大需要官方特性、复杂场景
mysqlclientC 扩展pip install mysqlclient (需 libmysql)最快、兼容性极高安装需系统库高性能生产、CPU 密集
SQLAlchemy 2.1+ORM 框架pip install sqlalchemy[asyncio]声明式、跨数据库、异步支持、迁移工具学习曲线稍陡生产项目首选
aiomysql异步pip install aiomysql基于 PyMySQL，高并发需要 asyncioFastAPI、异步项目

2026 年推荐路线：

  • 入门/脚本：PyMySQL
  • 生产/中大型项目：SQLAlchemy 2.1 + PyMySQL（或 mysql-connector-python）
  • 高并发异步：aiomysql + SQLAlchemy async

下面以 PyMySQL 为主讲解基础，后面再升级到 SQLAlchemy。

2. 环境准备（5 分钟搞定）

2.1 安装 MySQL

  • 本地：下载 MySQL 8.4 Community（推荐）或 9.0
  • Windows：官网 MSI Installer
  • macOS：brew install mysql
  • Linux：sudo apt install mysql-server
  • 云数据库（推荐新手）：阿里云 RDS、腾讯云 DB、PlanetScale、Neon（免费层够用）
  • 创建数据库（用 MySQL Workbench 或命令行）：
CREATE DATABASE IF NOT EXISTS py_test CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
CREATE USER IF NOT EXISTS 'pyuser'@'%' IDENTIFIED BY 'PyPassword2026!';
GRANT ALL PRIVILEGES ON py_test.* TO 'pyuser'@'%';
FLUSH PRIVILEGES;

2.2 Python 环境

确保 python --version 是 3.12+

pip install PyMySQL python-dotenv pymysqlpool tenacity  # 基础 + 生产增强
# 如果用 ORM
pip install sqlalchemy[asyncio] aiomysql

3. 基础连接（Hello MySQL）

创建 connect_demo.py：

import pymysql
from dotenv import load_dotenv
import os

load_dotenv()   # 加载 .env 文件

# 推荐写法：使用上下文管理器，自动关闭连接
def get_connection():
    return pymysql.connect(
        host=os.getenv('DB_HOST', 'localhost'),
        port=int(os.getenv('DB_PORT', 3306)),
        user=os.getenv('DB_USER', 'pyuser'),
        password=os.getenv('DB_PASSWORD', 'PyPassword2026!'),
        db=os.getenv('DB_NAME', 'py_test'),
        charset='utf8mb4',
        cursorclass=pymysql.cursors.DictCursor,   # 返回 dict 而非 tuple
        autocommit=False                          # 手动控制事务
    )

# 测试连接
with get_connection() as conn:
    with conn.cursor() as cursor:
        cursor.execute("SELECT VERSION() as version")
        result = cursor.fetchone()
        print(f"✅ 连接成功！MySQL 版本: {result['version']}")

创建 .env 文件（永不提交 Git）：

DB_HOST=localhost
DB_PORT=3306
DB_USER=pyuser
DB_PASSWORD=PyPassword2026!
DB_NAME=py_test

运行：python connect_demo.py → 看到版本号即成功！

4. CRUD 全操作实战（参数化查询防注入）

创建表（先执行一次）：

USE py_test;
CREATE TABLE IF NOT EXISTS users (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    username VARCHAR(50) UNIQUE NOT NULL,
    age INT,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

4.1 增（INSERT）

def insert_user(username: str, age: int):
    with get_connection() as conn:
        with conn.cursor() as cursor:
            sql = "INSERT INTO users (username, age) VALUES (%s, %s)"
            cursor.execute(sql, (username, age))
            conn.commit()
            print(f"✅ 插入成功，ID: {cursor.lastrowid}")

4.2 查（SELECT）

def get_users(min_age: int = 0):
    with get_connection() as conn:
        with conn.cursor() as cursor:
            sql = "SELECT id, username, age FROM users WHERE age >= %s"
            cursor.execute(sql, (min_age,))
            return cursor.fetchall()   # 返回 list[dict]

4.3 改（UPDATE）

def update_age(user_id: int, new_age: int):
    with get_connection() as conn:
        with conn.cursor() as cursor:
            sql = "UPDATE users SET age = %s WHERE id = %s"
            affected = cursor.execute(sql, (new_age, user_id))
            conn.commit()
            print(f"✅ 更新 {affected} 条记录")

4.4 删（DELETE）

def delete_user(user_id: int):
    with get_connection() as conn:
        with conn.cursor() as cursor:
            sql = "DELETE FROM users WHERE id = %s"
            cursor.execute(sql, (user_id,))
            conn.commit()

批量操作（超级实用）：

def batch_insert(users_list):
    with get_connection() as conn:
        with conn.cursor() as cursor:
            sql = "INSERT INTO users (username, age) VALUES (%s, %s)"
            cursor.executemany(sql, users_list)   # [( '张三',18), ('李四',20)]
            conn.commit()

5. 事务处理（原子性）

def transfer_money(from_user: int, to_user: int, amount: int):
    with get_connection() as conn:
        try:
            with conn.cursor() as cursor:
                # 扣款
                cursor.execute("UPDATE accounts SET balance = balance - %s WHERE id = %s", (amount, from_user))
                # 入账
                cursor.execute("UPDATE accounts SET balance = balance + %s WHERE id = %s", (amount, to_user))
                conn.commit()
                print("✅ 转账成功")
        except Exception as e:
            conn.rollback()
            print(f"❌ 转账失败，已回滚: {e}")
            raise

6. 生产级增强

6.1 连接池（推荐 pymysqlpool 或 SQLAlchemy）

from pymysqlpool import ConnectionPool

pool = ConnectionPool(
    size=10,
    maxsize=20,
    host=..., user=..., password=..., db=...
)

# 使用
with pool.get_connection() as conn:
    ...

最佳生产方案：SQLAlchemy Engine + Pool

from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker

engine = create_engine(
    "mysql+pymysql://user:pass@localhost/py_test?charset=utf8mb4",
    pool_size=20,
    max_overflow=10,
    pool_timeout=30,
    pool_recycle=3600   # 防止 MySQL 8小时断开
)

Session = sessionmaker(bind=engine)

with Session() as session:
    result = session.execute(text("SELECT VERSION()"))
    print(result.scalar())

6.2 ORM 完整示例（SQLAlchemy 2.1+）

from sqlalchemy import Column, Integer, String, DateTime, create_engine
from sqlalchemy.orm import declarative_base, sessionmaker
from datetime import datetime

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String(50), unique=True, nullable=False)
    age = Column(Integer)
    created_at = Column(DateTime, default=datetime.utcnow)

# 创建表
Base.metadata.create_all(engine)

CRUD 变成对象操作，代码更优雅。

7. 异步连接（FastAPI 必备）

import asyncio
import aiomysql

async def main():
    pool = await aiomysql.create_pool(
        host='localhost', user='pyuser', password='xxx', db='py_test'
    )
    async with pool.acquire() as conn:
        async with conn.cursor(aiomysql.DictCursor) as cur:
            await cur.execute("SELECT * FROM users")
            print(await cur.fetchall())
    pool.close()
    await pool.wait_closed()

asyncio.run(main())

8. 生产最佳实践（必须遵守）

  1. 永远用 .env + python-dotenv 管理密码
  2. 日志：用 logging 或 structlog 记录 SQL
  3. 重试机制（断网、死锁）：
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(5), wait=wait_exponential(multiplier=1, min=2, max=10))
def query_with_retry():
    ...
  1. 上下文管理器（with 语句）
  2. 监控：Prometheus + Grafana 监控连接数、慢查询
  3. Docker 示例（docker-compose.yml）：
services:
  mysql:
    image: mysql:8.4
    environment:
      MYSQL_ROOT_PASSWORD: root
      MYSQL_DATABASE: py_test
  app:
    build: .
    depends_on: [mysql]

9. 常见错误 & 排查（一键解决）

错误码原因解决办法
1045用户名/密码错误检查 .env 或 GRANT 权限
2003无法连接（主机/端口/防火墙）确认端口 3306 开放，bind-address
1146表不存在检查表名、数据库名
1366编码错误用 utf8mb4
2013Lost connection增大 wait_timeout 或 pool_recycle

完整小项目：用户管理系统（复制即用）

我可以直接给你一个包含注册、登录、CRUD 的 FastAPI + SQLAlchemy + MySQL 项目模板（100 行以内）。

现在行动：

  1. 把上面代码复制运行
  2. 告诉我你用的是 本地 MySQL 还是 云数据库？
  3. 你想继续看 SQLAlchemy 完整 ORM 项目、FastAPI 异步版，还是 批量导入百万数据性能优化？

把你运行后的截图或报错贴出来，我立刻帮你调试！

重阳，跟着这篇教程，10 分钟内你就能把 Python 和 MySQL 彻底打通！冲！🚀

网站 http://yingjuxia.com
文章已创建 4915

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注

相关文章

开始在上面输入您的搜索词，然后按回车进行搜索。按ESC取消。

返回顶部