---

上一篇【第018篇】文件与IO操作——读写CSV、JSON、Excel
下一篇【第020篇】网络请求实战：requests库深度使用

---

系列说明：本系列共 30 篇，旨在帮助Python学习者从零基础到精通。本系列强调实战导向，每篇文章都配有可运行的代码示例。本文为第 019 篇，聚焦于Python数据库操作，介绍SQLite基础和SQLAlchemy ORM的实战使用。

---

摘要

数据库是现代软件开发中不可或缺的核心组件，无论是Web应用、数据分析还是人工智能项目，都离不开数据的存储与管理。SQLite作为一种轻量级嵌入式数据库，以其零配置、无服务器、单一文件存储的特性，成为Python入门数据库编程的最佳选择。据统计，SQLite是世界上部署最广泛的数据库引擎，被广泛应用于移动应用、浏览器和桌面软件中。

SQLAlchemy则是Python生态中最功能强大且广泛使用的SQL工具包和ORM（对象关系映射）框架。它提供了Core和ORM两套API，Core更接近原生SQL而灵活高效，ORM则将数据库表映射为Python对象，使数据操作更加直观和面向对象。SQLAlchemy 2.0版本进行了重大升级，统一了Core和ORM的使用体验，进一步降低了学习曲线。

本文将带领读者从零开始掌握SQLite数据库的基础操作，深入理解SQLAlchemy ORM的核心概念，并通过一个完整的学生成绩管理系统实战项目，将理论知识转化为实际技能。通过学习本文，读者将能够独立完成中小型项目的数据库设计与实现，为后续学习更复杂的Web框架和数据处理技术奠定坚实基础。

一、SQLite数据库基础

1.1 SQLite简介与应用场景

SQLite是一款轻量级的嵌入式关系型数据库管理系统，由D.Richard Hipp于2000年发布。它的核心设计理念是“简单、高效、可靠”，整个数据库存储在单一的磁盘文件中，无需独立的数据库服务器进程，这使得SQLite成为嵌入式系统和小型应用的理想选择。

SQLite的主要特点包括：零配置部署，无需安装或配置；跨平台支持，数据库文件可在不同操作系统间自由迁移；事务支持完整，符合ACID特性；单文件存储便于备份和分发；支持标准SQL语法，学习成本低。SQLite的应用场景非常广泛，涵盖了移动应用（iOS和Android的内置数据库）、浏览器存储（Chrome的Web SQL和IndexedDB底层）、桌面软件配置存储、小型网站和博客系统、数据分析和原型开发等领域。

在Python中，SQLite的使用非常简单。Python 3标准库内置了sqlite3模块，无需额外安装任何包即可直接使用数据库功能。这使得SQLite成为学习数据库编程的最佳起点，读者可以在不搭建复杂数据库环境的情况下，掌握数据库操作的核心概念和技能。

1.2 创建数据库与表结构

在Python中使用SQLite数据库非常简单，无需显式创建数据库文件，当执行第一个SQL语句时会自动创建数据库文件。下面通过具体代码演示数据库的创建和表结构的定义过程。

import sqlite3
from pathlib import Path


def create_database(db_path: str = "school.db") -> sqlite3.Connection:
    """
    创建数据库连接，如果数据库不存在会自动创建
    
    Args:
        db_path: 数据库文件路径
        
    Returns:
        sqlite3.Connection对象
    """
    conn = sqlite3.connect(db_path)
    print(f"数据库连接已建立: {db_path}")
    return conn


def create_tables(conn: sqlite3.Connection) -> None:
    """
    创建学生表和成绩表
    
    Args:
        conn: 数据库连接对象
    """
    cursor = conn.cursor()
    
    # 创建学生表
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS students (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name VARCHAR(50) NOT NULL,
            gender VARCHAR(10),
            age INTEGER,
            class_name VARCHAR(50),
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    """)
    
    # 创建课程表
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS courses (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name VARCHAR(100) NOT NULL,
            teacher VARCHAR(50),
            credits INTEGER DEFAULT 0
        )
    """)
    
    # 创建成绩表
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS scores (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            student_id INTEGER NOT NULL,
            course_id INTEGER NOT NULL,
            score REAL NOT NULL,
            exam_date DATE,
            FOREIGN KEY (student_id) REFERENCES students(id),
            FOREIGN KEY (course_id) REFERENCES courses(id),
            UNIQUE(student_id, course_id)
        )
    """)
    
    conn.commit()
    print("数据表创建成功！")


if __name__ == "__main__":
    db_path = Path(__file__).parent / "school.db"
    conn = create_database(str(db_path))
    create_tables(conn)
    conn.close()
    print("数据库初始化完成！")

上述代码展示了使用Python标准库sqlite3模块进行数据库操作的基本模式。首先通过sqlite3.connect()函数建立数据库连接，如果指定的数据库文件不存在，SQLite会自动创建该文件。然后通过cursor.execute()方法执行SQL语句，使用CREATE TABLE语句定义表结构。表结构中的主键、外键、约束等概念与传统关系型数据库完全一致。

值得注意的是，代码中使用了IF NOT EXISTS子句，这可以防止在表已存在时重复创建导致的错误。AUTOINCREMENT关键字使得主键ID可以自动递增，FOREIGN KEY约束建立了表之间的关联关系，UNIQUE约束确保学生和课程的组合唯一，避免重复成绩记录。

1.3 增删改查基本操作

在掌握了数据库和表的创建方法后，本节将详细介绍SQLite的CRUD操作，即创建（Create）、读取（Read）、更新（Update）和删除（Delete）。这些是数据库操作中最基础也是最核心的操作。

import sqlite3
from datetime import date
from typing import Optional


class StudentDAO:
    """学生数据访问对象，提供学生表的CRUD操作"""
    
    def __init__(self, db_path: str = "school.db"):
        self.db_path = db_path
    
    def _get_connection(self) -> sqlite3.Connection:
        """获取数据库连接的内部方法"""
        conn = sqlite3.connect(self.db_path)
        conn.row_factory = sqlite3.Row  # 使查询结果可以通过列名访问
        return conn
    
    def create(self, name: str, gender: str, age: int, 
               class_name: str) -> int:
        """
        插入新学生记录
        
        Args:
            name: 学生姓名
            gender: 性别
            age: 年龄
            class_name: 班级
            
        Returns:
            新插入记录的行ID
        """
        conn = self._get_connection()
        cursor = conn.cursor()
        
        cursor.execute("""
            INSERT INTO students (name, gender, age, class_name)
            VALUES (?, ?, ?, ?)
        """, (name, gender, age, class_name))
        
        conn.commit()
        student_id = cursor.lastrowid
        conn.close()
        
        print(f"成功添加学生: {name}, 学号: {student_id}")
        return student_id
    
    def read(self, student_id: int) -> Optional[dict]:
        """
        根据ID查询学生信息
        
        Args:
            student_id: 学生ID
            
        Returns:
            学生信息字典，不存在则返回None
        """
        conn = self._get_connection()
        cursor = conn.cursor()
        
        cursor.execute("SELECT * FROM students WHERE id = ?", (student_id,))
        row = cursor.fetchone()
        conn.close()
        
        if row:
            return dict(row)
        return None
    
    def read_all(self, class_filter: Optional[str] = None) -> list:
        """
        查询所有学生或按班级筛选
        
        Args:
            class_filter: 班级名称筛选条件
            
        Returns:
            学生信息列表
        """
        conn = self._get_connection()
        cursor = conn.cursor()
        
        if class_filter:
            cursor.execute(
                "SELECT * FROM students WHERE class_name = ? ORDER BY id",
                (class_filter,)
            )
        else:
            cursor.execute("SELECT * FROM students ORDER BY id")
        
        rows = cursor.fetchall()
        conn.close()
        
        return [dict(row) for row in rows]
    
    def update(self, student_id: int, name: Optional[str] = None,
               age: Optional[int] = None) -> bool:
        """
        更新学生信息
        
        Args:
            student_id: 学生ID
            name: 新姓名（可选）
            age: 新年龄（可选）
            
        Returns:
            是否更新成功
        """
        updates = []
        values = []
        
        if name is not None:
            updates.append("name = ?")
            values.append(name)
        if age is not None:
            updates.append("age = ?")
            values.append(age)
        
        if not updates:
            print("没有需要更新的字段")
            return False
        
        values.append(student_id)
        sql = f"UPDATE students SET {', '.join(updates)} WHERE id = ?"
        
        conn = self._get_connection()
        cursor = conn.cursor()
        cursor.execute(sql, values)
        conn.commit()
        
        success = cursor.rowcount > 0
        conn.close()
        
        if success:
            print(f"成功更新学生ID {student_id}")
        else:
            print(f"未找到学生ID {student_id}")
        
        return success
    
    def delete(self, student_id: int) -> bool:
        """
        删除学生记录
        
        Args:
            student_id: 学生ID
            
        Returns:
            是否删除成功
        """
        conn = self._get_connection()
        cursor = conn.cursor()
        
        cursor.execute("DELETE FROM students WHERE id = ?", (student_id,))
        conn.commit()
        
        success = cursor.rowcount > 0
        conn.close()
        
        if success:
            print(f"成功删除学生ID {student_id}")
        else:
            print(f"未找到学生ID {student_id}")
        
        return success


# 使用示例
if __name__ == "__main__":
    dao = StudentDAO("school.db")
    
    # 插入学生
    student_id = dao.create("张三", "男", 18, "计算机一班")
    
    # 查询学生
    student = dao.read(student_id)
    print(f"查询结果: {student}")
    
    # 更新学生
    dao.update(student_id, age=19)
    
    # 查询所有学生
    all_students = dao.read_all()
    print(f"当前学生总数: {len(all_students)}")

上述代码展示了使用Python操作SQLite数据库的标准模式。关键点包括：使用连接作为上下文管理器确保资源正确释放；使用参数化查询（?占位符）防止SQL注入攻击，这是数据库安全的基本要求；设置row_factory使得查询结果可以像字典一样通过列名访问，提高了代码的可读性。

1.4 事务处理

事务是数据库操作中保证数据一致性和完整性的重要机制。SQLite完全支持ACID事务特性，即原子性（Atomicity）、一致性（Consistency）、隔离性（Isolation）和持久性（Durability）。在Python的sqlite3模块中，事务默认是自动开启的，但可以通过设置isolation_level来控制事务行为。

import sqlite3
from contextlib import contextmanager


@contextmanager
def transaction(db_path: str = "school.db"):
    """
    事务上下文管理器
    
    用法:
        with transaction() as conn:
            cursor = conn.cursor()
            cursor.execute(...)
    """
    conn = sqlite3.connect(db_path)
    try:
        yield conn
        conn.commit()
        print("事务提交成功！")
    except Exception as e:
        conn.rollback()
        print(f"事务回滚: {e}")
        raise
    finally:
        conn.close()


def batch_insert_students(db_path: str = "school.db"):
    """批量插入学生记录，使用事务确保数据一致性"""
    
    students_data = [
        ("李四", "女", 19, "计算机一班"),
        ("王五", "男", 18, "计算机一班"),
        ("赵六", "女", 20, "计算机二班"),
        ("孙七", "男", 19, "计算机二班"),
        ("周八", "女", 18, "计算机二班"),
    ]
    
    with transaction(db_path) as conn:
        cursor = conn.cursor()
        
        for name, gender, age, class_name in students_data:
            cursor.execute("""
                INSERT INTO students (name, gender, age, class_name)
                VALUES (?, ?, ?, ?)
            """, (name, gender, age, class_name))
            print(f"插入: {name}")
        
        # 模拟错误，测试事务回滚
        # raise ValueError("模拟插入失败")
    
    print("批量插入完成！")


def transfer_score(db_path: str = "school.db"):
    """模拟成绩迁移事务"""
    
    with transaction(db_path) as conn:
        cursor = conn.cursor()
        
        # 查询原成绩
        cursor.execute("""
            SELECT student_id, course_id, score 
            FROM scores 
            WHERE score < 60
        """)
        failing_scores = cursor.fetchall()
        
        # 为不及格成绩添加补考标记
        for row in failing_scores:
            cursor.execute("""
                UPDATE scores 
                SET score = score + 10 
                WHERE student_id = ? AND course_id = ? AND score < 60
            """, (row['student_id'], row['course_id']))
        
        print(f"已为 {len(failing_scores)} 条不及格成绩添加加分")


if __name__ == "__main__":
    batch_insert_students("school.db")

事务处理的核心原则是：一组相关的数据库操作要么全部成功，要么全部失败。当所有操作都成功时，使用commit()提交事务使更改永久化；当任何操作失败时，使用rollback()回滚事务，撤销所有未提交的更改。

在生产环境中，事务的正确使用对于保证数据一致性至关重要。特别是在涉及多表操作、批量数据处理或业务逻辑复杂的场景中，必须确保事务的完整性和一致性。transfer_score函数展示了典型的业务场景：在同一事务中先查询后更新，保证数据状态的一致性。

二、SQLAlchemy ORM介绍

2.1 ORM概念与优势

ORM（Object-Relational Mapping，对象关系映射）是一种程序设计技术，用于实现面向对象编程语言中不同类型系统间的数据转换。在数据库领域，ORM将关系型数据库的表结构映射为编程语言中的类，将表中的每条记录映射为类的实例（对象），将字段映射为对象的属性。这种映射关系使得开发者可以使用面向对象的方式操作数据库，无需编写复杂的SQL语句。

ORM的主要优势体现在以下几个方面。首先是代码可读性和可维护性的提升。使用ORM，数据库表变成了Python类，表之间的外键关系变成了对象之间的引用，开发者可以用直观的方式理解和管理数据。其次是开发效率的提高。ORM自动处理了SQL语句的生成、参数转义和结果映射，开发者可以专注于业务逻辑的实现。此外，ORM提供了数据库无关性，同一套代码可以轻松切换到不同的数据库后端。最后，ORM内置了对SQL注入等安全问题的防护，通过参数化查询机制自动规避了潜在的安全风险。

当然，ORM并非完美无缺。对于复杂的查询场景，手写SQL可能比ORM更高效；ORM的学习曲线也需要一定时间投入。但在绝大多数应用场景中，ORM带来的便利性远超其潜在的性能开销。

2.2 SQLAlchemy安装与版本

SQLAlchemy是Python生态中最成熟的数据库工具包，目前最新版本为2.0.x。相比之前的版本，SQLAlchemy 2.0进行了重大架构升级，提供了更统一的API、更清晰的编程模型和更好的性能。SQLAlchemy分为Core和ORM两部分，Core提供底层SQL表达式语言，ORM则是建立在其之上的对象关系映射层。

安装SQLAlchemy非常简单，只需使用pip命令即可。对于标准的同步操作，基本安装即可满足需求：

# 基本安装（推荐）
pip install sqlalchemy

# 如果需要异步支持（如配合asyncio使用）
pip install sqlalchemy[asyncio]

# 检查安装版本
python -c "import sqlalchemy; print(sqlalchemy.__version__)"

对于SQLite数据库，由于使用的是Python内置的sqlite3模块，无需额外安装数据库驱动。对于MySQL、PostgreSQL等其他数据库，需要额外安装对应的数据库驱动，如pymysql、psycopg2等。

# 环境检查脚本
import sys
print(f"Python版本: {sys.version}")

import sqlalchemy
print(f"SQLAlchemy版本: {sqlalchemy.__version__}")

# 检查SQLite支持
from sqlalchemy import create_engine
engine = create_engine("sqlite:///:memory:")
with engine.connect() as conn:
    result = conn.execute(sqlalchemy.text("SELECT 1 + 1 AS result"))
    print(f"SQLite测试: {result.fetchone()[0]}")

2.3 Engine、Session、Base核心概念

理解SQLAlchemy的核心概念是掌握其使用的关键。SQLAlchemy 2.0的架构主要由Engine、Session和Base三个核心组件构成，它们协同工作，提供了从数据库连接到对象操作的完整链路。

Engine是SQLAlchemy与数据库通信的起点。当调用create_engine()时，SQLAlchemy会根据数据库URL创建一个Engine实例。Engine负责维护连接池（Connection Pool），处理数据库连接的生命周期，以及将Python代码翻译为底层数据库的通信协议。Engine本身不直接执行SQL，而是通过获取的Connection对象来执行操作。

from sqlalchemy import create_engine, text

# 创建Engine
# SQLite的特殊语法：sqlite:///相对路径 或 sqlite:////绝对路径
engine = create_engine(
    "sqlite:///school.db",  # 相对路径
    echo=True,               # 打印生成的SQL语句，便于调试
    pool_pre_ping=True,      # 连接前检测连接是否有效
)

# 使用Engine执行原始SQL
with engine.connect() as conn:
    result = conn.execute(text("SELECT 'Hello, SQLAlchemy!' AS greeting"))
    print(result.fetchone())

# 显式关闭Engine
engine.dispose()

Session是ORM操作的核心对象，它提供了一组数据库事务的上下文。与Engine不同，Session代表一个工作单元，追踪对象的变更并在适当时机将更改同步到数据库。Session的使用遵循"打开、使用、关闭"的模式，最好配合上下文管理器使用以确保资源正确释放。

from sqlalchemy.orm import sessionmaker

# 创建Session工厂
SessionLocal = sessionmaker(bind=engine)

# 创建Session实例
session = SessionLocal()

try:
    # 在此处进行数据库操作
    pass
finally:
    session.close()

# 推荐：使用上下文管理器
with SessionLocal() as session:
    # Session会自动关闭
    pass

Base是所有ORM模型类的基类，通过declarative_base()函数创建。使用ORM时，所有模型类都需要继承Base，SQLAlchemy会自动为这些类创建与数据库表的映射关系。Base还负责维护一个元数据注册表，跟踪所有已定义的模型和表结构。

from sqlalchemy.orm import declarative_base

# 创建Base
Base = declarative_base()

# 继承Base定义模型
class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    
    def __repr__(self):
        return f"<User(id={self.id}, name='{self.name}')>"

理解这三个核心概念的关系很重要：Engine负责底层数据库连接和通信；Session提供了面向对象的事务工作单元；Base定义了ORM模型的元数据结构。三者配合使用，构成了SQLAlchemy ORM的完整使用模式。

三、数据模型定义

3.1 模型类与表映射

在SQLAlchemy ORM中，数据模型（Model）是连接Python对象和数据库表的桥梁。通过在模型类中定义类属性并继承declarative_base()创建的基类，SQLAlchemy会自动处理类到表的映射关系。这种声明式的定义方式清晰、直观，是SQLAlchemy ORM的核心编程范式。

from datetime import datetime
from sqlalchemy import Column, Integer, String, Float, DateTime, Boolean
from sqlalchemy.orm import declarative_base

Base = declarative_base()


class Student(Base):
    """
    学生模型
    
    映射到数据库中的 students 表
    """
    __tablename__ = "students"
    
    # 主键：自增整数
    id = Column(Integer, primary_key=True, autoincrement=True)
    
    # 姓名：非空字符串，最大50字符
    name = Column(String(50), nullable=False, index=True)
    
    # 性别：字符串
    gender = Column(String(10), default="未知")
    
    # 年龄：整数
    age = Column(Integer)
    
    # 班级：字符串
    class_name = Column(String(50))
    
    # 邮箱：唯一字符串
    email = Column(String(100), unique=True)
    
    # 是否激活
    is_active = Column(Boolean, default=True)
    
    # 创建时间：自动设置为当前时间
    created_at = Column(DateTime, default=datetime.now)
    
    # 更新时间：自动更新
    updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now)
    
    def __repr__(self):
        return f"<Student(id={self.id}, name='{self.name}', class='{self.class_name}')>"
    
    def to_dict(self):
        """转换为字典，便于JSON序列化"""
        return {
            "id": self.id,
            "name": self.name,
            "gender": self.gender,
            "age": self.age,
            "class_name": self.class_name,
            "email": self.email,
            "is_active": self.is_active,
            "created_at": self.created_at.isoformat() if self.created_at else None,
        }


# 创建所有表
def create_all_tables(engine):
    """在数据库中创建所有继承自Base的表"""
    Base.metadata.create_all(engine)
    print("所有表创建成功！")


# 删除所有表
def drop_all_tables(engine):
    """删除数据库中所有继承自Base的表"""
    Base.metadata.drop_all(engine)
    print("所有表已删除！")

上述代码展示了SQLAlchemy模型定义的标准模式。__tablename__属性指定了对应的数据库表名；Column对象定义了各个字段的数据类型和约束条件；nullable=False表示字段不能为空；index=True为该字段创建索引以提高查询效率；default指定默认值；unique=True确保字段值唯一。

3.2 字段类型与约束

SQLAlchemy提供了丰富的字段类型，涵盖了数据库中常见的数据类型。正确选择字段类型对于数据存储效率和查询性能都有重要影响。以下是常用字段类型的对照表：

SQLAlchemy类型	Python类型	说明
Integer	int	整数类型
BigInteger	int	大整数
SmallInteger	int	小范围整数
Float	float	浮点数
Numeric	Decimal	精确小数
String	str	变长字符串
Text	str	长文本
Boolean	bool	布尔值
Date	date	日期
DateTime	datetime	日期时间
Time	time	时间
LargeBinary	bytes	二进制数据
Enum	str	枚举值

字段约束是保证数据完整性的重要机制。常用的约束包括：primary_key指定主键；nullable控制是否可为空；unique确保唯一性；default设置默认值；index创建索引提升查询性能；ondelete和onupdate指定级联操作行为。合理使用约束可以减少应用层的验证逻辑，提高数据质量。

3.3 关系定义（一对多、多对多）

关系型数据库的核心价值在于能够表达表与表之间的关联关系。SQLAlchemy ORM通过relationship()函数提供了优雅的方式来实现这些关系映射。正确理解和使用关系是掌握SQLAlchemy ORM的关键。

一对多关系是最常见的关联模式。例如，一个班级有多名学生，那么班级与学生之间就是一对多关系。在数据库层面，学生表中需要有一个指向班级表的外键；在ORM层面，通过relationship可以方便地从班级对象获取其所有学生。

from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship, declarative_base

Base = declarative_base()


class Department(Base):
    """系部模型"""
    __tablename__ = "departments"
    
    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=False)
    code = Column(String(20), unique=True)
    
    # 反向引用：一个系部有多个班级
    classes = relationship("ClassModel", back_populates="department")
    
    def __repr__(self):
        return f"<Department(name='{self.name}')>"


class ClassModel(Base):
    """班级模型"""
    __tablename__ = "classes"
    
    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=False)
    department_id = Column(Integer, ForeignKey("departments.id"), nullable=False)
    
    # 外键关系：一个班级属于一个系部
    department = relationship("Department", back_populates="classes")
    # 反向引用：一个班级有多个学生
    students = relationship("Student", back_populates="class_model")
    
    def __repr__(self):
        return f"<Class(name='{self.name}')>"


class Student(Base):
    """学生模型"""
    __tablename__ = "students"
    
    id = Column(Integer, primary_key=True)
    name = Column(String(50), nullable=False)
    class_id = Column(Integer, ForeignKey("classes.id"), nullable=False)
    
    # 外键关系：一个学生属于一个班级
    class_model = relationship("ClassModel", back_populates="students")
    
    def __repr__(self):
        return f"<Student(name='{self.name}')>"

多对多关系需要借助关联表来实现。例如，学生和课程之间就是多对多关系——一个学生可以选修多门课程，一门课程也可以被多个学生选修。在数据库层面，需要创建专门的关联表；在ORM层面，SQLAlchemy的relationship支持直接操作这种关系。

from sqlalchemy import Column, Integer, String, ForeignKey, Table
from sqlalchemy.orm import relationship, declarative_base

Base = declarative_base()


# 关联表：学生-课程
enrollments = Table(
    "enrollments",
    Base.metadata,
    Column("student_id", Integer, ForeignKey("students.id"), primary_key=True),
    Column("course_id", Integer, ForeignKey("courses.id"), primary_key=True),
    Column("enroll_date", Date, default=date.today),
    Column("grade", Float, nullable=True)  # 最终成绩
)


class Student(Base):
    """学生模型"""
    __tablename__ = "students"
    
    id = Column(Integer, primary_key=True)
    name = Column(String(50), nullable=False)
    
    # 多对多关系：学生选修的课程
    courses = relationship(
        "Course",
        secondary=enrollments,  # 指定关联表
        back_populates="students"
    )
    
    def __repr__(self):
        return f"<Student(name='{self.name}')>"


class Course(Base):
    """课程模型"""
    __tablename__ = "courses"
    
    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=False)
    credits = Column(Integer, default=0)
    
    # 多对多关系：选修该课程的学生
    students = relationship(
        "Student",
        secondary=enrollments,
        back_populates="courses"
    )
    
    def __repr__(self):
        return f"<Course(name='{self.name}', credits={self.credits})>"

四、CRUD操作实战

4.1 创建记录

使用SQLAlchemy ORM创建记录的过程非常直观。只需实例化模型类，设置属性值，然后通过Session添加到数据库会话中，最后提交事务即可完成插入操作。SQLAlchemy会自动处理SQL语句的生成和参数绑定。

from datetime import date
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from model_definition import Student, Course, Department  # 假设已定义模型

# 创建数据库引擎
engine = create_engine("sqlite:///school.db", echo=True)
Session = sessionmaker(bind=engine)


def create_records():
    """演示创建记录的各种方式"""
    
    session = Session()
    
    try:
        # 方式1：创建单个对象并添加
        student1 = Student(name="张三", gender="男", age=20, class_name="计算机一班")
        session.add(student1)
        
        # 方式2：批量创建多个对象
        students = [
            Student(name="李四", gender="女", age=19, class_name="计算机一班"),
            Student(name="王五", gender="男", age=21, class_name="计算机二班"),
            Student(name="赵六", gender="女", age=20, class_name="计算机二班"),
        ]
        session.add_all(students)
        
        # 方式3：使用已有数据创建记录
        course = Course(name="Python程序设计", code="CS101", credits=4)
        session.add(course)
        
        # 提交事务
        session.commit()
        
        # 获取自动生成的主键
        print(f"插入学生ID: {student1.id}")
        print(f"插入课程ID: {course.id}")
        
    except Exception as e:
        session.rollback()
        print(f"创建失败: {e}")
        raise
    finally:
        session.close()


if __name__ == "__main__":
    create_records()

add()方法将对象标记为待插入状态，但不会立即执行SQL；add_all()用于批量添加多个对象；commit()方法触发SQL执行并提交事务。

4.2 查询数据

查询是数据库操作中使用最频繁的操作。SQLAlchemy ORM提供了功能强大且直观的查询接口，支持从简单的全表查询到复杂的条件过滤、排序和分页操作。

from sqlalchemy import create_engine, desc
from sqlalchemy.orm import sessionmaker
from model_definition import Student, Course, Score

engine = create_engine("sqlite:///school.db")
Session = sessionmaker(bind=engine)


def basic_queries():
    """基础查询操作"""
    
    session = Session()
    
    try:
        # 查询所有记录
        all_students = session.query(Student).all()
        print(f"学生总数: {len(all_students)}")
        
        # 查询第一条记录
        first_student = session.query(Student).first()
        print(f"第一个学生: {first_student.name}")
        
        # 获取指定主键的记录
        student = session.get(Student, 1)  # 获取ID为1的学生
        print(f"ID为1的学生: {student.name if student else '不存在'}")
        
        # 统计记录数量
        count = session.query(Student).count()
        print(f"统计数量: {count}")
        
    finally:
        session.close()


def filter_queries():
    """条件过滤查询"""
    
    session = Session()
    
    try:
        # 等值过滤：==
        students = session.query(Student).filter(
            Student.class_name == "计算机一班"
        ).all()
        
        # 不等于：!= 或 ~
        students = session.query(Student).filter(
            Student.gender != "男"
        ).all()
        
        # 模糊匹配：like / ilike
        students = session.query(Student).filter(
            Student.name.like("张%")
        ).all()
        
        # 多条件组合：and_, or_
        from sqlalchemy import and_, or_
        students = session.query(Student).filter(
            and_(
                Student.class_name == "计算机一班",
                Student.age >= 20
            )
        ).all()
        
        # IN查询
        from sqlalchemy import in_
        students = session.query(Student).filter(
            Student.id.in_([1, 2, 3])
        ).all()
        
        print(f"过滤结果: {len(students)} 条")
        
    finally:
        session.close()


def order_and_pagination():
    """排序与分页查询"""
    
    session = Session()
    
    try:
        # 基本排序：升序
        students = session.query(Student).order_by(Student.age).all()
        
        # 降序排序
        students = session.query(Student).order_by(
            desc(Student.age)
        ).all()
        
        # 分页查询
        page = 1  # 第几页
        page_size = 10  # 每页记录数
        
        total = session.query(Student).count()
        total_pages = (total + page_size - 1) // page_size
        
        students = session.query(Student).order_by(
            Student.id
        ).offset((page - 1) * page_size).limit(page_size).all()
        
        print(f"第 {page}/{total_pages} 页，共 {total} 条记录")
        
    finally:
        session.close()


if __name__ == "__main__":
    basic_queries()
    filter_queries()
    order_and_pagination()

SQLAlchemy的查询接口设计得非常优雅，几乎所有的SQL操作都可以通过链式调用方法来实现。filter()方法用于添加查询条件；order_by()用于排序；offset()和limit()用于分页。

4.3 更新记录

更新数据库记录有多种方式，取决于更新的范围和复杂度。对于单个对象的更新，SQLAlchemy提供了最直观的方式——直接修改对象属性并提交事务。对于批量更新，则可以使用update()语句直接执行。

from sqlalchemy import create_engine, update
from sqlalchemy.orm import sessionmaker
from model_definition import Student

engine = create_engine("sqlite:///school.db")
Session = sessionmaker(bind=engine)


def update_single_record():
    """更新单条记录"""
    
    session = Session()
    
    try:
        # 方式1：通过对象更新
        student = session.get(Student, 1)
        if student:
            student.age = 22
            student.class_name = "计算机三班"
            session.commit()
            print(f"已更新: {student.name}")
        
    finally:
        session.close()


def update_batch():
    """批量更新记录"""
    
    session = Session()
    
    try:
        # 使用Query更新
        updated_count = session.query(Student).filter(
            Student.class_name == "计算机一班"
        ).update({"class_name": "CS2021级1班"})
        
        print(f"更新了 {updated_count} 条记录")
        session.commit()
        
    except Exception as e:
        session.rollback()
        print(f"更新失败: {e}")
    finally:
        session.close()


if __name__ == "__main__":
    update_single_record()
    update_batch()

更新操作需要特别注意事务管理。在SQLAlchemy中，对象的属性修改不会立即同步到数据库，必须通过commit()显式提交才能生效。

4.4 删除记录

删除记录与更新类似，可以针对单个对象删除，也可以进行批量删除。在实际应用中，为了保证数据完整性，通常需要先处理与删除对象相关联的其他记录。

from sqlalchemy import create_engine, delete
from sqlalchemy.orm import sessionmaker
from model_definition import Student, Score

engine = create_engine("sqlite:///school.db")
Session = sessionmaker(bind=engine)


def delete_single_record():
    """删除单条记录"""
    
    session = Session()
    
    try:
        student = session.get(Student, 5)
        if student:
            session.delete(student)
            session.commit()
            print(f"已删除ID为5的学生")
        else:
            print("该学生不存在")
            
    except Exception as e:
        session.rollback()
        print(f"删除失败: {e}")
    finally:
        session.close()


def delete_batch():
    """批量删除记录"""
    
    session = Session()
    
    try:
        # 使用Query删除
        deleted_count = session.query(Student).filter(
            Student.age > 25,
            Student.is_active == False
        ).delete()
        
        print(f"批量删除了 {deleted_count} 条记录")
        session.commit()
        
    except Exception as e:
        session.rollback()
        print(f"删除失败: {e}")
    finally:
        session.close()


if __name__ == "__main__":
    delete_single_record()
    delete_batch()

在实际项目中，除非确实需要删除数据，否则推荐使用"软删除"策略——即添加一个deleted_at或is_deleted字段来标记记录已被逻辑删除，而非物理删除。

五、实战项目：学生成绩管理系统

5.1 需求分析与数据库设计

本节将通过一个完整的学生成绩管理系统实战项目，将前面学到的SQLite和SQLAlchemy ORM知识融会贯通。该系统需要实现以下核心功能：学生信息管理、课程信息管理、成绩录入与查询、成绩统计与分析。

根据需求分析，系统需要包含以下实体：学生（Student）、课程（Course）、成绩（Score）、班级（Class）和教师（Teacher）。实体之间的关系如下：班级与学生为一对多关系；教师与课程为一对多关系；学生与课程为多对多关系（通过成绩表关联）。

5.2 完整代码实现

"""
学生成绩管理系统
使用SQLite + SQLAlchemy ORM实现

作者：Python学习者
版本：1.0
Python版本：3.11+
"""

from datetime import date, datetime
from typing import Optional
from sqlalchemy import (
    create_engine, Column, Integer, String, Float, 
    ForeignKey, Date, DateTime, UniqueConstraint
)
from sqlalchemy.orm import (
    sessionmaker, relationship, declarative_base
)

Base = declarative_base()


# ==================== 数据模型定义 ====================

class Student(Base):
    """学生模型"""
    __tablename__ = "students"
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(50), nullable=False, index=True)
    gender = Column(String(10))
    birth_date = Column(Date)
    student_no = Column(String(20), unique=True, nullable=False)
    class_id = Column(Integer, ForeignKey("classes.id"))
    
    # 关系
    class_model = relationship("ClassModel", back_populates="students")
    scores = relationship("Score", back_populates="student")
    
    def __repr__(self):
        return f"<Student(no='{self.student_no}', name='{self.name}')>"
    
    def to_dict(self):
        return {
            "id": self.id,
            "name": self.name,
            "gender": self.gender,
            "birth_date": self.birth_date.isoformat() if self.birth_date else None,
            "student_no": self.student_no,
            "class_name": self.class_model.name if self.class_model else None
        }
    
    @property
    def age(self) -> Optional[int]:
        """计算学生年龄"""
        if self.birth_date:
            today = date.today()
            age = today.year - self.birth_date.year
            if today.month < self.birth_date.month or \
               (today.month == self.birth_date.month and today.day < self.birth_date.day):
                age -= 1
            return age
        return None
    
    @property
    def average_score(self) -> Optional[float]:
        """计算平均成绩"""
        if not self.scores:
            return None
        total = sum(s.score for s in self.scores if s.score is not None)
        count = sum(1 for s in self.scores if s.score is not None)
        return round(total / count, 2) if count > 0 else None


class Course(Base):
    """课程模型"""
    __tablename__ = "courses"
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(100), nullable=False)
    code = Column(String(20), unique=True, nullable=False)
    credits = Column(Integer, default=0)
    
    # 关系
    scores = relationship("Score", back_populates="course")
    
    def __repr__(self):
        return f"<Course(code='{self.code}', name='{self.name}')>"


class Score(Base):
    """成绩模型"""
    __tablename__ = "scores"
    __table_args__ = (
        UniqueConstraint("student_id", "course_id", name="uq_student_course"),
    )
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    student_id = Column(Integer, ForeignKey("students.id"), nullable=False)
    course_id = Column(Integer, ForeignKey("courses.id"), nullable=False)
    score = Column(Float, nullable=False)
    exam_date = Column(Date, default=date.today)
    
    # 关系
    student = relationship("Student", back_populates="scores")
    course = relationship("Course", back_populates="scores")
    
    def __repr__(self):
        return f"<Score(student={self.student_id}, course={self.course_id}, score={self.score})>"
    
    @property
    def grade_level(self) -> str:
        """根据成绩返回等级"""
        if self.score >= 90:
            return "A"
        elif self.score >= 80:
            return "B"
        elif self.score >= 70:
            return "C"
        elif self.score >= 60:
            return "D"
        else:
            return "F"


class ClassModel(Base):
    """班级模型"""
    __tablename__ = "classes"
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(50), nullable=False, unique=True)
    grade = Column(Integer, nullable=False)
    
    # 关系
    students = relationship("Student", back_populates="class_model")
    
    def __repr__(self):
        return f"<Class(name='{self.name}', grade={self.grade})>"


# ==================== 数据库管理器 ====================

class DatabaseManager:
    """数据库管理器，处理数据库初始化和会话管理"""
    
    def __init__(self, db_path: str = "student_system.db"):
        self.engine = create_engine(
            f"sqlite:///{db_path}",
            echo=False,
            pool_pre_ping=True
        )
        self.SessionLocal = sessionmaker(bind=self.engine)
    
    def create_tables(self):
        """创建所有数据表"""
        Base.metadata.create_all(self.engine)
        print("数据库表创建完成！")
    
    def get_session(self):
        """获取数据库会话"""
        return self.SessionLocal()


# ==================== 测试和演示 ====================

def init_sample_data(db_manager: DatabaseManager):
    """初始化示例数据"""
    
    session = db_manager.get_session()
    
    try:
        # 检查是否已有数据
        if session.query(Student).count() > 0:
            print("数据已存在，跳过初始化")
            return
        
        # 创建班级
        classes = [
            ClassModel(name="计算机24级1班", grade=2024),
            ClassModel(name="计算机24级2班", grade=2024),
        ]
        session.add_all(classes)
        session.flush()
        
        # 创建课程
        courses = [
            Course(name="Python程序设计", code="CS101", credits=4),
            Course(name="数据结构与算法", code="CS102", credits=4),
            Course(name="数据库原理", code="CS103", credits=3),
        ]
        session.add_all(courses)
        session.flush()
        
        # 创建学生
        students_data = [
            ("张三", "2024001", "男", date(2005, 3, 15), classes[0].id),
            ("李四", "2024002", "女", date(2005, 7, 22), classes[0].id),
            ("王五", "2024003", "男", date(2004, 11, 8), classes[0].id),
            ("赵六", "2024004", "女", date(2005, 1, 30), classes[1].id),
            ("钱七", "2024005", "男", date(2004, 9, 12), classes[1].id),
        ]
        
        students = []
        for name, no, gender, birth, class_id in students_data:
            student = Student(
                name=name, student_no=no, gender=gender,
                birth_date=birth, class_id=class_id
            )
            students.append(student)
        session.add_all(students)
        session.flush()
        
        # 创建成绩
        scores_data = [
            (students[0], courses[0], 92),
            (students[0], courses[1], 88),
            (students[0], courses[2], 95),
            (students[1], courses[0], 85),
            (students[1], courses[1], 78),
            (students[2], courses[0], 67),
            (students[2], courses[2], 72),
            (students[3], courses[0], 90),
            (students[3], courses[1], 82),
            (students[4], courses[0], 55),
            (students[4], courses[2], 63),
        ]
        
        for student, course, score in scores_data:
            score_obj = Score(
                student_id=student.id,
                course_id=course.id,
                score=score,
                exam_date=date(2024, 12, 20)
            )
            session.add(score_obj)
        
        session.commit()
        print("示例数据初始化完成！")
        
    except Exception as e:
        session.rollback()
        print(f"初始化失败: {e}")
        raise
    finally:
        session.close()


def demo_operations():
    """演示各种操作"""
    
    db_manager = DatabaseManager("student_system.db")
    db_manager.create_tables()
    init_sample_data(db_manager)
    
    session = db_manager.get_session()
    
    try:
        print("\n=== 学生列表 ===")
        students = session.query(Student).all()
        for s in students:
            print(f"  {s.student_no} - {s.name} - {s.class_model.name} - "
                  f"年龄:{s.age} - 平均分:{s.average_score}")
        
        print("\n=== 成绩等级分布 ===")
        all_scores = session.query(Score).all()
        grade_counts = {"A": 0, "B": 0, "C": 0, "D": 0, "F": 0}
        for sc in all_scores:
            grade_counts[sc.grade_level] += 1
        for grade, count in grade_counts.items():
            print(f"  {grade}: {count}人")
        
    finally:
        session.close()


if __name__ == "__main__":
    demo_operations()

5.3 运行效果演示

运行上述代码后，系统将完成以下操作流程：

数据库表创建完成！
示例数据初始化完成！

=== 学生列表 ===
  2024001 - 张三 - 计算机24级1班 - 年龄:19 - 平均分:91.67
  2024002 - 李四 - 计算机24级1班 - 年龄:19 - 平均分:81.5
  2024003 - 王五 - 计算机24级1班 - 年龄:20 - 平均分:69.5
  2024004 - 赵六 - 计算机24级2班 - 年龄:19 - 平均分:86.0
  2024005 - 钱七 - 计算机24级2班 - 年龄:20 - 平均分:59.0

=== 成绩等级分布 ===
  A: 4人
  B: 2人
  C: 1人
  D: 2人
  F: 2人

通过这个实战项目，我们完整展示了SQLite和SQLAlchemy ORM的协同使用，包括数据模型定义、关系映射、CRUD操作、聚合查询和统计分析等核心功能。

六、常见问题与注意事项

6.1 数据库连接管理

数据库连接管理是使用数据库时最需要注意的问题之一。连接泄漏会导致程序性能下降，严重时可能导致数据库服务不可用。以下是SQLAlchemy中连接管理的最佳实践。

首先，始终使用上下文管理器或try-finally结构确保连接被正确关闭。其次，对于高并发应用，需要合理配置连接池大小。最后，对于长时间运行的后台任务，应该定期刷新会话以避免过期连接问题。

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# 推荐：使用上下文管理器
engine = create_engine("sqlite:///app.db")
Session = sessionmaker(bind=engine)

with Session() as session:
    result = session.query(Student).all()
# session已自动关闭

6.2 性能优化建议

在使用SQLAlchemy进行数据库操作时，以下几点可以帮助提升性能。首先，尽量使用批量操作代替循环中的单条插入或更新。其次，对于只读的查询，可以使用expire_on_commit=False避免不必要的对象重新加载。合理使用索引可以显著提升查询性能。

# 批量插入优化
students = [Student(name=f"学生{i}") for i in range(1000)]
session.bulk_save_objects(students)
session.commit()

6.3 安全注意事项

数据库安全是不容忽视的重要话题。SQL注入攻击是最常见的数据库安全威胁，SQLAlchemy通过参数化查询机制自动防止了这类攻击，但在编写代码时仍需注意绝不将敏感信息硬编码在代码中。

# 安全：使用参数化查询
result = session.query(Student).filter(
    Student.name == user_input  # SQLAlchemy自动处理转义
).all()

七、总结

本文系统地介绍了Python数据库操作的核心知识，从SQLite数据库的基础使用，到SQLAlchemy ORM的进阶应用，通过理论讲解与实战项目相结合的方式，帮助读者全面掌握Python数据库编程技能。

文章首先介绍了SQLite作为轻量级嵌入式数据库的特点和使用方法，包括数据库连接、表创建、CRUD操作和事务处理等基础知识。然后深入讲解了SQLAlchemy ORM的核心概念，包括Engine、Session、Base三大组件的作用和协作方式，以及数据模型的定义方法和关系映射的实现技巧。

通过一个完整的学生成绩管理系统实战项目，将所学知识融会贯通，展示了如何构建结构清晰、功能完整的数据库应用。最后总结了数据库连接管理、性能优化和安全防护等方面的最佳实践。

---

上一篇【第018篇】文件与IO操作——读写CSV、JSON、Excel
下一篇【第020篇】网络请求实战：requests库深度使用

---

参考资料

• SQLAlchemy官方文档
• SQLAlchemy 2.0统一教程
• SQLAlchemy中文文档
• 廖雪峰的SQLAlchemy教程
• Python sqlite3官方文档
• Python SQLite最佳实践

---