Python Web 开发：数据库操作全攻略

引言 / 什么是 Python Web 开发中的数据库操作

在 Python Web 开发中，数据库是存储和管理数据的核心组件。无论是用户信息、订单数据还是配置参数，几乎所有业务逻辑都依赖数据库的支持。Python 通过丰富的库（如 pymysql、SQLAlchemy、psycopg2）提供了与多种数据库（MySQL、PostgreSQL、SQLite）的交互能力。

本文将聚焦 MySQL 数据库，通过一个完整的用户信息管理系统案例，详细讲解如何使用 Python 实现数据库的连接、数据增删改查（CRUD）操作，帮助你快速掌握后端开发中的数据管理技能。

准备工作

1. 环境要求：

   • Python 3.6+
   • MySQL 8.0+（或本地安装的 MySQL 服务）
   • 推荐使用虚拟环境：python -m venv venv && source venv/bin/activate

2. 安装依赖库：

   pip install pymysql sqlalchemy
   

   • pymysql：纯 Python 实现的 MySQL 客户端库。
   • SQLAlchemy：功能强大的 ORM 工具，支持多种数据库。

3. 创建测试数据库：

   CREATE DATABASE IF NOT EXISTS user_management CHARACTER SET utf8mb4;
   USE user_management;
   CREATE TABLE users (
       id INT AUTO_INCREMENT PRIMARY KEY,
       username VARCHAR(50) NOT NULL UNIQUE,
       email VARCHAR(100) NOT NULL UNIQUE,
       age INT DEFAULT 18
   );
   

基础操作 / 核心用法

步骤一：连接 MySQL 数据库

使用 pymysql 直接连接数据库：

import pymysql

# 连接配置
connection = pymysql.connect(
    host='localhost',
    user='root',
    password='your_password',
    database='user_management',
    charset='utf8mb4',
    cursorclass=pymysql.cursors.DictCursor  # 返回字典格式结果
)

try:
    with connection.cursor() as cursor:
        # 测试连接
        cursor.execute("SELECT VERSION()")
        version = cursor.fetchone()
        print(f"数据库版本: {version['VERSION()']}")
finally:
    connection.close()

提示：生产环境中建议将数据库配置存储在环境变量或配置文件中，避免硬编码。

步骤二：实现数据增删改查（CRUD）

1. 创建数据（Create）

def add_user(username, email, age=18):
    connection = pymysql.connect(**DB_CONFIG)
    try:
        with connection.cursor() as cursor:
            sql = "INSERT INTO users (username, email, age) VALUES (%s, %s, %s)"
            cursor.execute(sql, (username, email, age))
        connection.commit()  # 提交事务
        print("用户添加成功！")
    except Exception as e:
        connection.rollback()  # 回滚事务
        print(f"添加失败: {e}")
    finally:
        connection.close()

# 示例调用
add_user("alice", "alice@example.com", 25)

2. 查询数据（Read）

def get_users():
    connection = pymysql.connect(**DB_CONFIG)
    try:
        with connection.cursor() as cursor:
            cursor.execute("SELECT * FROM users")
            users = cursor.fetchall()  # 获取所有结果
            for user in users:
                print(f"ID: {user['id']}, 用户名: {user['username']}")
    finally:
        connection.close()

# 条件查询
def get_user_by_id(user_id):
    connection = pymysql.connect(**DB_CONFIG)
    try:
        with connection.cursor() as cursor:
            cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
            user = cursor.fetchone()  # 获取单条结果
            return user
    finally:
        connection.close()

3. 更新数据（Update）

def update_user_age(user_id, new_age):
    connection = pymysql.connect(**DB_CONFIG)
    try:
        with connection.cursor() as cursor:
            sql = "UPDATE users SET age = %s WHERE id = %s"
            cursor.execute(sql, (new_age, user_id))
        connection.commit()
        print("用户年龄更新成功！")
    except Exception as e:
        connection.rollback()
        print(f"更新失败: {e}")
    finally:
        connection.close()

4. 删除数据（Delete）

def delete_user(user_id):
    connection = pymysql.connect(**DB_CONFIG)
    try:
        with connection.cursor() as cursor:
            cursor.execute("DELETE FROM users WHERE id = %s", (user_id,))
        connection.commit()
        print("用户删除成功！")
    except Exception as e:
        connection.rollback()
        print(f"删除失败: {e}")
    finally:
        connection.close()

进阶技巧

使用 SQLAlchemy 实现 ORM 操作

SQLAlchemy 是 Python 中最流行的 ORM 工具，能通过类映射数据库表，简化操作：

1. 定义模型类：

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)
    username = Column(String(50), unique=True)
    email = Column(String(100), unique=True)
    age = Column(Integer, default=18)

# 创建引擎和会话
engine = create_engine('mysql+pymysql://root:your_password@localhost/user_management')
Session = sessionmaker(bind=engine)
session = Session()

2. CRUD 操作：

# 添加用户
new_user = User(username="bob", email="bob@example.com", age=30)
session.add(new_user)
session.commit()

# 查询用户
users = session.query(User).filter_by(age=30).all()
for user in users:
    print(user.username)

# 更新用户
user = session.query(User).get(1)
user.age = 26
session.commit()

# 删除用户
user = session.query(User).get(2)
session.delete(user)
session.commit()

事务管理与连接池

常见问题

Q：如何处理数据库连接失败？
A：使用 try-except 捕获 pymysql.Error 或 sqlalchemy.exc.OperationalError，记录错误日志并重试连接。

Q：如何防止 SQL 注入攻击？
A：始终使用参数化查询（如 %s 占位符），避免直接拼接 SQL 字符串。

Q：ORM 和原生 SQL 如何选择？
A：简单查询推荐 ORM（代码更简洁）；复杂查询或性能优化时使用原生 SQL。

小结

本文通过 MySQL 数据库 和 Python 的结合，详细讲解了：

1. 数据库连接配置与基础 CRUD 操作；
2. 使用 SQLAlchemy 实现 ORM 映射；
3. 事务管理与安全实践。

完整代码示例可在 GitHub 获取。建议从原生 SQL 开始练习，再逐步过渡到 ORM，以深入理解数据库操作原理。动手实现一个用户管理系统，是掌握 Python Web 开发数据库操作的最佳实践！