Python与MySQL：从基础操作到实战技巧的完整指南

2025-08-18 132

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

本文涉及的产品

RDS MySQL Serverless 基础系列，0.5-2RCU 50GB

云数据库 RDS PostgreSQL，高可用系列 2核4GB

RDS MySQL Serverless 高可用系列，价值2615元额度，1个月

简介： Python操作MySQL凭借高效性能与成熟生态，广泛应用于Web开发、金融风控及物联网等领域。借助pymysql等库，开发者可快速实现数据库连接与增删改查操作，同时通过参数化查询、连接池及事务管理保障系统安全与稳定。本文详解环境搭建、实战技巧与性能优化，助你构建高并发应用。

一、为什么选择Python操作MySQL？
在Web开发中，MySQL作为开源关系型数据库的代表，与Python的结合堪称"黄金搭档"。以某电商平台为例，其用户系统日均处理10万次登录请求，Python通过pymysql库与MySQL配合，能在0.2秒内完成身份验证。这种高效协作得益于：
代理IP助力机器人赛事信息安全 (23).png
免费python教程合集：https://pan.quark.cn/s/5f868993f9a5
性能优势：MySQL的InnoDB引擎支持每秒数千次事务处理
生态成熟：Python的pymysql库拥有超过2000万次下载量
开发效率：三行代码即可建立数据库连接（示例见下文）
这种组合被广泛应用于金融风控系统、物联网数据采集等场景。某智能工厂的实时监控系统，通过Python每5秒采集2000个传感器数据，经MySQL存储后，供数据分析模块生成生产报告。

二、环境搭建：从零开始的准备工作

数据库安装与配置
以MySQL 8.0为例，在Ubuntu系统上通过以下命令安装：

sudo apt update
sudo apt install mysql-server
sudo mysql_secure_installation # 安全配置向导
安装完成后，执行mysql -u root -p登录，创建测试数据库：

CREATE DATABASE test_db CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
CREATE USER 'python_user'@'localhost' IDENTIFIED BY 'SecurePass123!';
GRANT ALL PRIVILEGES ON test_db.* TO 'python_user'@'localhost';
FLUSH PRIVILEGES;

Python库选择
主流方案对比：

库名称特点适用场景
pymysql 纯Python实现，兼容性好跨平台开发
mysql-connector-python Oracle官方驱动企业级应用
SQLAlchemy ORM框架，支持多种数据库复杂业务系统
推荐新手从pymysql入手，安装命令：

pip install pymysql
三、核心操作：增删改查实战

连接数据库的三种方式
基础版：

import pymysql
conn = pymysql.connect(
host='localhost',
user='python_user',
password='SecurePass123!',
database='test_db',
charset='utf8mb4'
)

上下文管理器版（推荐）：

with pymysql.connect(...) as conn:
with conn.cursor() as cursor:
cursor.execute("SELECT VERSION()")
print(cursor.fetchone())
连接池版（高并发场景）：

from dbutils.pooled_db import PooledDB
pool = PooledDB(
creator=pymysql,
maxconnections=5,
host='localhost',
user='python_user',
password='SecurePass123!',
database='test_db'
)
conn = pool.connection() # 从连接池获取连接

查询操作进阶
基础查询：

def get_user_by_id(user_id):
with conn.cursor() as cursor:
cursor.execute("SELECT * FROM users WHERE id=%s", (user_id,))
return cursor.fetchone()

分页查询优化：

def get_users_paginated(page, per_page=10):
offset = (page - 1) per_page
with conn.cursor() as cursor:
cursor.execute(
"SELECT FROM users ORDER BY id LIMIT %s OFFSET %s",
(per_page, offset)
)
return cursor.fetchall()

批量查询技巧：

def get_users_by_ids(user_ids):

# 生成占位符字符串，如 '%s,%s,%s'
placeholders = ','.join(['%s'] * len(user_ids))
with conn.cursor() as cursor:
    cursor.execute(
        f"SELECT * FROM users WHERE id IN ({placeholders})",
        tuple(user_ids)
    )
    return cursor.fetchall()

数据修改与事务处理
原子性操作示例：

def transfer_balance(from_id, to_id, amount):
try:
with conn.cursor() as cursor:

        # 开启事务（pymysql默认不自动提交）
        conn.begin()

        # 扣款操作
        cursor.execute(
            "UPDATE accounts SET balance=balance-%s WHERE id=%s AND balance>=%s",
            (amount, from_id, amount)
        )
        if cursor.rowcount == 0:
            raise ValueError("扣款失败：余额不足或用户不存在")

        # 存款操作
        cursor.execute(
            "UPDATE accounts SET balance=balance+%s WHERE id=%s",
            (amount, to_id)
        )

        conn.commit()
        return True
except Exception as e:
    conn.rollback()
    print(f"转账失败：{str(e)}")
    return False

批量插入优化：

def batch_insert_users(user_list):
with conn.cursor() as cursor:

    # 使用executemany批量插入
    cursor.executemany(
        "INSERT INTO users (name, email) VALUES (%s, %s)",
        [(u['name'], u['email']) for u in user_list]
    )
    print(f"成功插入 {cursor.rowcount} 条记录")

四、安全防护：防止SQL注入

参数化查询原理
对比两种查询方式：

危险写法（易受SQL注入攻击）

username = input("请输入用户名：")
cursor.execute(f"SELECT * FROM users WHERE username='{username}'")

安全写法（使用参数化查询）

cursor.execute("SELECT * FROM users WHERE username=%s", (username,))

特殊字符处理
当需要存储包含引号的数据时：

def safe_insert_article(title, content):
with conn.cursor() as cursor:

    # pymysql会自动处理特殊字符转义
    cursor.execute(
        "INSERT INTO articles (title, content) VALUES (%s, %s)",
        (title, content)
    )
    conn.commit()

最小权限原则
数据库用户应遵循最小权限原则，例如：

-- 只授予必要的权限
GRANT SELECT, INSERT ON test_db.users TO 'app_user'@'localhost';
REVOKE ALL PRIVILEGES, GRANT OPTION FROM 'app_user'@'localhost';

五、性能优化：从毫秒到微秒的突破

索引优化实战
为高频查询字段添加索引：

ALTER TABLE orders ADD INDEX idx_customer_id (customer_id);
ALTER TABLE orders ADD INDEX idx_order_date (order_date);
通过EXPLAIN分析查询性能：

def analyze_query(sql):
with conn.cursor() as cursor:
cursor.execute("EXPLAIN " + sql)
return cursor.fetchall()

示例输出：

[('1', 'SIMPLE', 'orders', 'ALL', None, None, None, None, '10000', '100.00', 'Using where')]

连接管理策略

长连接复用：通过连接池保持5-10个持久连接
短连接控制：设置connect_timeout=5避免长时间等待
负载均衡：使用ProxySQL实现读写分离

查询缓存技巧

from functools import lru_cache

@lru_cache(maxsize=100)
def get_user_cached(user_id):
with conn.cursor() as cursor:
cursor.execute("SELECT * FROM users WHERE id=%s", (user_id,))
return cursor.fetchone()

六、异常处理：构建健壮的系统

常见错误类型

错误代码说明解决方案
1045 访问被拒绝检查用户名密码和权限
1064 SQL语法错误使用try-except捕获并记录
2003 无法连接到MySQL服务器检查网络和防火墙设置
2006 MySQL服务器已关闭实现自动重连机制

重试机制实现

import time
from pymysql import MySQLError

def execute_with_retry(sql, args=None, max_retries=3):
for attempt in range(max_retries):
try:
with conn.cursor() as cursor:
cursor.execute(sql, args or ())
return cursor.fetchall()
except MySQLError as e:
if attempt == max_retries - 1:
raise
wait_time = 2 ** attempt # 指数退避
time.sleep(wait_time)

日志记录系统

import logging

logging.basicConfig(
filename='db_operations.log',
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)

def log_query(sql, args=None):
logging.info(f"Executing: {sql} with args: {args}")
try:
with conn.cursor() as cursor:
cursor.execute(sql, args or ())
result = cursor.fetchall()
logging.info(f"Affected rows: {cursor.rowcount}")
return result
except MySQLError as e:
logging.error(f"Database error: {str(e)}")
raise

七、进阶应用：ORM与异步操作

SQLAlchemy快速入门

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

Base = declarative_base()

class User(Base):
tablename = 'users'
id = Column(Integer, primary_key=True)
name = Column(String(50))
email = Column(String(100))

创建引擎和会话

engine = create_engine('mysql+pymysql://python_user:SecurePass123!@localhost/test_db')
Session = sessionmaker(bind=engine)
session = Session()

查询示例

users = session.query(User).filter(User.name.like('%John%')).all()

异步操作（aiomysql）

import asyncio
import aiomysql

async def test_async_query():
conn = await aiomysql.connect(
host='localhost',
port=3306,
user='python_user',
password='SecurePass123!',
db='test_db'
)
async with conn.cursor() as cursor:
await cursor.execute("SELECT VERSION()")
print(await cursor.fetchone())
conn.close()

asyncio.run(test_async_query())

八、实战案例：电商订单系统

数据库设计

CREATE TABLE customers (
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(100) NOT NULL,
email VARCHAR(100) UNIQUE NOT
);

CREATE TABLE products (
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(100) NOT NULL,
price DECIMAL(10,2) NOT NULL,
stock INT NOT
);

CREATE TABLE orders (
id INT AUTO_INCREMENT PRIMARY KEY,
customer_id INT NOT NULL,
order_date DATETIME DEFAULT CURRENT_TIMESTAMP,
total_amount DECIMAL(10,2) NOT NULL,
FOREIGN KEY (customer_id) REFERENCES customers(id)
);

CREATE TABLE order_items (
id INT AUTO_INCREMENT PRIMARY KEY,
order_id INT NOT NULL,
product_id INT NOT NULL,
quantity INT NOT NULL,
unit_price DECIMAL(10,2) NOT NULL,
FOREIGN KEY (order_id) REFERENCES orders(id),
FOREIGN KEY (product_id) REFERENCES products(id)
);

核心业务实现

def create_order(customer_id, items):
"""
items格式: [{'product_id': 1, 'quantity': 2}, ...]
"""
try:
with conn.cursor() as cursor:

        # 计算总金额
        total = 0
        for item in items:
            cursor.execute(
                "SELECT price FROM products WHERE id=%s FOR UPDATE",
                (item['product_id'],)
            )
            product = cursor.fetchone()
            if not product:
                raise ValueError(f"产品不存在: {item['product_id']}")
            unit_price = product[0]
            total += unit_price * item['quantity']

        # 创建订单
        cursor.execute(
            "INSERT INTO orders (customer_id, total_amount) VALUES (%s, %s)",
            (customer_id, total)
        )
        order_id = cursor.lastrowid

        # 添加订单项
        for item in items:
            cursor.execute(
                "INSERT INTO order_items (order_id, product_id, quantity, unit_price) VALUES (%s, %s, %s, %s)",
                (order_id, item['product_id'], item['quantity'], unit_price)
            )

        # 更新库存
        for item in items:
            cursor.execute(
                "UPDATE products SET stock=stock-%s WHERE id=%s AND stock>=%s",
                (item['quantity'], item['product_id'], item['quantity'])
            )
            if cursor.rowcount == 0:
                raise ValueError(f"库存不足: 产品{item['product_id']}")

        conn.commit()
        return order_id
except Exception as e:
    conn.rollback()
    raise

九、总结与展望
从基础连接操作到复杂事务处理，Python与MySQL的协作展现了强大的灵活性。某金融科技公司的实时风控系统，通过Python每秒处理2000笔交易，MySQL的行级锁机制确保了数据一致性。未来发展方向包括：

AI驱动优化：利用机器学习自动优化查询计划
Serverless架构：AWS Aurora Serverless等新型数据库服务
量子计算集成：探索量子加密在数据库安全的应用
掌握这些核心技能后，开发者可以轻松构建从个人博客到企业级应用的各类系统。建议持续关注MySQL 8.0的新特性（如JSON增强、窗口函数）和Python异步编程的发展趋势。

Python与MySQL：从基础操作到实战技巧的完整指南

危险写法（易受SQL注入攻击）

安全写法（使用参数化查询）

示例输出：

[('1', 'SIMPLE', 'orders', 'ALL', None, None, None, None, '10000', '100.00', 'Using where')]

创建引擎和会话

查询示例

热门文章

最新文章

相关课程

相关电子书

推荐镜像

探索云世界

热门

云计算

大数据

云原生

人工智能

数据库

开发与运维

活动广场

任务中心

训练营

直播

乘风者计划

下载

镜像站

技术资料

Python与MySQL：从基础操作到实战技巧的完整指南

危险写法（易受SQL注入攻击）

安全写法（使用参数化查询）

示例输出：

[('1', 'SIMPLE', 'orders', 'ALL', None, None, None, None, '10000', '100.00', 'Using where')]

创建引擎和会话

查询示例

热门文章

最新文章

相关课程

相关电子书

推荐镜像