访问数据库

数据库基础概念

在深入Python数据库编程之前，我们先了解一些基本概念。

关系数据库 vs NoSQL

特性	关系数据库	NoSQL
数据结构	表（行和列）	键值、文档、列族、图
查询语言	SQL	各不相同的API
事务支持	ACID完整支持	最终一致性
扩展性	垂直扩展为主	水平扩展为主
典型代表	MySQL, PostgreSQL, SQLite	MongoDB, Redis, Cassandra

JDBC/DB-API 原理

┌─────────────────────────────────────────────────────────┐
│              Python数据库访问架构                          │
├─────────────────────────────────────────────────────────┤
│                                                         │
│   ┌─────────────┐                                       │
│   │  Python应用  │                                       │
│   └──────┬──────┘                                       │
│          │                                              │
│          ▼                                              │
│   ┌─────────────────┐                                   │
│   │   DB-API 2.0    │  (Python标准接口)                   │
│   │ cursor, connect │                                   │
│   └──────┬──────────┘                                   │
│          │                                              │
│          ▼                                              │
│   ┌─────────────────┐                                   │
│   │   数据库驱动     │  (PyMySQL, sqlite3, psycopg2)   │
│   │                 │                                   │
│   └──────┬──────────┘                                   │
│          │                                              │
│          ▼                                              │
│   ┌─────────────────┐                                   │
│   │   数据库服务器   │  (MySQL, PostgreSQL, SQLite)      │
│   └─────────────────┘                                   │
│                                                         │
└─────────────────────────────────────────────────────────┘

SQLite数据库

SQLite是Python内置的轻量级数据库，整个数据库存储在一个文件中，适合桌面应用和移动应用。

Python

import sqlite3

# ===== 连接数据库 =====
conn = sqlite3.connect("test.db")  # 文件不存在会自动创建
cursor = conn.cursor()

# ===== 创建表 =====
cursor.execute("""CREATE TABLE IF NOT EXISTS users (
              id INTEGER PRIMARY KEY AUTOINCREMENT,
              name TEXT NOT NULL,
              email TEXT UNIQUE,
              age INTEGER,
              created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
          )""")

# ===== 插入数据 (CRUD - Create) =====
# 单条插入
cursor.execute("INSERT INTO users (name, email, age) VALUES (?, ?, ?)",
              ("张三", "zhangsan@example.com", 25))
conn.commit()

# 批量插入
users = [
    ("李四", "lisi@example.com", 30),
    ("王五", "wangwu@example.com", 28),
    ("赵六", "zhaoliu@example.com", 35)
]
cursor.executemany("INSERT INTO users (name, email, age) VALUES (?, ?, ?)", users)
conn.commit()

# ===== 查询数据 (CRUD - Read) =====
# 查询所有
cursor.execute("SELECT * FROM users")
results = cursor.fetchall()
for row in results:
    print(f"ID: {row[0]}, Name: {row[1]}, Email: {row[2]}, Age: {row[3]}")

# 条件查询
cursor.execute("SELECT * FROM users WHERE age > ? ORDER BY age DESC", (25,))
results = cursor.fetchall()
print(f"\n年龄大于25的用户: {results}")

# 查询单条
cursor.execute("SELECT * FROM users WHERE name = ?", ("张三",))
user = cursor.fetchone()
print(f"张三的信息: {user}")

# ===== 更新数据 (CRUD - Update) =====
cursor.execute("UPDATE users SET age = ? WHERE name = ?", (26, "张三"))
conn.commit()
print(f"更新影响的行数: {cursor.rowcount}")

# ===== 删除数据 (CRUD - Delete) =====
cursor.execute("DELETE FROM users WHERE name = ?", ("赵六",))
conn.commit()
print(f"删除影响的行数: {cursor.rowcount}")

# ===== 关闭连接 =====
conn.close()

SQLAlchemy ORM

SQLAlchemy是Python最流行的ORM（对象关系映射）框架，可以让我们用面向对象的方式操作数据库。

┌─────────────────────────────────────────────────────────┐
│                    ORM工作原理                             │
├─────────────────────────────────────────────────────────┤
│                                                         │
│   ┌─────────────────┐         ┌─────────────────┐      │
│   │   Python对象     │  ←→    │   SQL表         │      │
│   │                 │  ORM   │                 │      │
│   │  User           │        │  users          │      │
│   │  ├─ id           │        │  ├─ id          │      │
│   │  ├─ name         │        │  ├─ name        │      │
│   │  ├─ email        │        │  ├─ email       │      │
│   │  └─ age          │        │  └─ age         │      │
│   └─────────────────┘        └─────────────────┘      │
│                                                         │
│   user = User(name="张三", age=25)  ←→  INSERT INTO...  │
│   db.query(User)               ←→  SELECT * FROM...    │
│   db.delete(user)              ←→  DELETE FROM...      │
│                                                         │
└─────────────────────────────────────────────────────────┘

Python

# 安装: pip install sqlalchemy
from sqlalchemy import create_engine, Column, Integer, String, DateTime, func
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, autoincrement=True)
    name = Column(String(50), nullable=False)
    email = Column(String(100), unique=True)
    age = Column(Integer)
    created_at = Column(DateTime, default=func.now())

    def __repr__(self):
        return f"<User(id={self.id}, name='{self.name}', email='{self.email}'>"

# ===== 创建引擎和会话 =====
engine = create_engine("sqlite:///test.db", echo=True)
Base.metadata.create_all(engine)

Session = sessionmaker(bind=engine)
session = Session()

# ===== 插入数据 =====
user1 = User(name="张三", email="zhangsan@example.com", age=25)
user2 = User(name="李四", email="lisi@example.com", age=30)
user3 = User(name="王五", email="wangwu@example.com", age=28)

session.add(user1)
session.add(user2)
session.add(user3)
session.commit()

# 批量添加
session.add_all([
    User(name="赵六", email="zhaoliu@example.com", age=32),
    User(name="钱七", email="qianqi@example.com", age=27)
])
session.commit()

# ===== 查询数据 =====
# 查询所有
all_users = session.query(User).all()
for user in all_users:
    print(user)

# 条件查询
young_users = session.query(User).filter(User.age < 30).all()
print(f"\n30岁以下的用户: {young_users}")

# 排序查询
sorted_users = session.query(User).order_by(User.age.desc()).all()
print(f"\n按年龄降序: {sorted_users}")

# 分页查询
page_users = session.query(User).limit(2).offset(0).all()
print(f"\n前2个用户: {page_users}")

# 聚合查询
from sqlalchemy import func
count = session.query(func.count(User.id)).scalar()
avg_age = session.query(func.avg(User.age)).scalar()
print(f"\n用户总数: {count}, 平均年龄: {avg_age:.1f}")

# ===== 更新数据 =====
user = session.query(User).filter(User.name == "张三").first()
user.age = 26
session.commit()

# ===== 删除数据 =====
user = session.query(User).filter(User.name == "赵六").first()
session.delete(user)
session.commit()

# ===== 关闭会话 =====
session.close()

MySQL数据库

MySQL是世界上最流行的开源关系数据库，下面介绍使用PyMySQL连接MySQL的方法。

Python

# 安装: pip install pymysql
import pymysql
from contextlib import contextmanager

# ===== 基本连接 =====
conn = pymysql.connect(
    host="localhost",
    port=3306,
    user="root",
    password="your_password",
    database="test_db",
    charset="utf8mb4"
)
cursor = conn.cursor()

# ===== 执行SQL =====
cursor.execute("CREATE TABLE IF NOT EXISTS products (id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(100), price DECIMAL(10,2))")

cursor.execute("INSERT INTO products (name, price) VALUES (%s, %s)", ("笔记本电脑", 5999.00))
conn.commit()

cursor.execute("SELECT * FROM products")
results = cursor.fetchall()
for row in results:
    print(row)

conn.close()

# ===== 使用上下文管理器 =====
@contextmanager
def get_connection():
    conn = pymysql.connect(
        host="localhost",
        port=3306,
        user="root",
        password="your_password",
        database="test_db",
        charset="utf8mb4"
    )
    try:
        yield conn
    finally:
        conn.close()

with get_connection() as conn:
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM products")
    print(cursor.fetchall())

事务处理

事务是数据库操作的基本单位，确保数据的一致性和完整性。

Python

import sqlite3

conn = sqlite3.connect("bank.db")
cursor = conn.cursor()

# 创建账户表
cursor.execute("""CREATE TABLE IF NOT EXISTS accounts (
              id INTEGER PRIMARY KEY,
              name TEXT,
              balance REAL
          )""")
cursor.execute("DELETE FROM accounts")
cursor.execute("INSERT INTO accounts (id, name, balance) VALUES (1, '张三', 10000)")
cursor.execute("INSERT INTO accounts (id, name, balance) VALUES (2, '李四', 5000)")
conn.commit()

# ===== 转账示例：事务确保数据一致性 =====
def transfer(from_id, to_id, amount):
    try:
        cursor.execute("SELECT balance FROM accounts WHERE id = ?", (from_id,))
        from_balance = cursor.fetchone()[0]

        if from_balance < amount:
            raise ValueError(f"余额不足: 当前余额 {from_balance}")

        # 扣除转出账户金额
        cursor.execute("UPDATE accounts SET balance = balance - ? WHERE id = ?",
                      (amount, from_id))

        # 添加转入账户金额
        cursor.execute("UPDATE accounts SET balance = balance + ? WHERE id = ?",
                      (amount, to_id))

        conn.commit()
        print(f"转账成功: 从 {from_id} 转 {amount} 到 {to_id}")

    except Exception as e:
        conn.rollback()  # 回滚事务
        print(f"转账失败: {e}")

# 测试转账
print("转账前:")
cursor.execute("SELECT * FROM accounts")
print(cursor.fetchall())

transfer(1, 2, 3000)

print("\n转账后:")
cursor.execute("SELECT * FROM accounts")
print(cursor.fetchall())

# 测试余额不足
transfer(1, 2, 20000)

conn.close()

💡 ACID特性：

Atomicity（原子性）：事务中的操作要么全部成功，要么全部失败回滚
Consistency（一致性）：事务执行前后，数据库状态保持一致
Isolation（隔离性）：并发执行的事务互不干扰
Durability（持久性）：事务提交后，结果永久保存

PyMySQL + SQLAlchemy

可以结合PyMySQL作为SQLAlchemy的MySQL驱动使用。

Python

# 安装: pip install sqlalchemy pymysql
# 使用PyMySQL作为MySQL驱动
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

Base = declarative_base()

class Product(Base):
    __tablename__ = "products"

    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(100), nullable=False)
    price = Column(Integer)

# MySQL连接字符串
engine = create_engine(
    "mysql+pymysql://root:password@localhost:3306/test_db?charset=utf8mb4"
)

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

# 创建会话
Session = sessionmaker(bind=engine)
session = Session()

# CRUD操作
session.add(Product(name="手机", price=2999))
session.commit()

products = session.query(Product).all()
print(products)

session.close()

章节总结

数据库操作速查表

操作	SQL	SQLAlchemy
创建连接	connect("db")	create_engine(), sessionmaker()
插入	INSERT INTO...	session.add(), session.add_all()
查询	SELECT * FROM...	session.query(Model)
条件查询	WHERE...	.filter(Model.col == value)
更新	UPDATE...	obj.col = new_val; session.commit()
删除	DELETE FROM...	session.delete(obj); session.commit()
提交	COMMIT	session.commit()
回滚	ROLLBACK	session.rollback()

关键知识点：

SQLite：Python内置轻量级数据库，文件存储，适合桌面和移动应用
SQLAlchemy ORM：对象关系映射，用面向对象方式操作数据库
PyMySQL：纯Python实现的MySQL驱动
事务ACID：原子性、一致性、隔离性、持久性
连接管理：使用上下文管理器确保资源释放
SQL注入防护：使用参数化查询而非字符串拼接

课后测验

课后练习

练习1：学生管理系统数据库设计

设计一个学生管理系统的数据库，包括学生表、课程表和学生选课表，实现基本的增删改查功能。

提示

学生表：id, name, age, gender
课程表：id, name, teacher, credits
选课表：student_id, course_id, score
使用SQLAlchemy定义模型类
实现添加学生、选课、查询成绩等功能

练习2：博客系统数据库设计

设计一个简单博客系统的数据库，实现用户发表文章、评论的功能。

提示

用户表：id, username, email, password_hash
文章表：id, title, content, author_id, created_at
评论表：id, content, author_id, article_id, created_at
表之间通过外键关联
使用级联删除处理关联数据