数据库教程(二):使用编程语言调用数据库,以Python为例

1. 介绍

SQLAlchemy 是 Python SQL 工具包和对象关系映射器 (ORM)，它为应用程序开发人员提供了 SQL 的全部功能和灵活性。它提供了两种主要的工作模式：

1. 1. SQLAlchemy Core: 提供了一个灵活的 SQL 表达式语言，允许你以 Python 的方式构建 SQL 语句，并直接与数据库进行交互。它更接近于原始 SQL，提供了高度的控制力。
2. 2. SQLAlchemy ORM (Object Relational Mapper): 在 Core 的基础上，提供了一种将 Python 对象映射到数据库表的方式。它允许你使用 Python 对象来操作数据库，而无需编写原始 SQL 语句，从而提高了开发效率和代码可读性。

本教程将带你从 SQLAlchemy 的基础概念开始，逐步深入到 Core 和 ORM 的使用，并通过实际案例演示其强大功能。

2. 环境准备

在开始之前，请确保你已经安装了 SQLAlchemy。如果你还没有安装，可以使用 pip 进行安装：

pip install sqlalchemy

我们还将使用 SQLite 作为数据库，因为它轻量且无需额外配置。如果你想连接其他数据库（如 MySQL、PostgreSQL），你需要安装相应的数据库驱动（例如 pymysql 或 psycopg2）。

3. SQLAlchemy Core 基础

SQLAlchemy Core 提供了构建 SQL 语句的抽象层，让你能够以 Python 的方式来操作数据库。它主要由以下几个核心组件构成：

• • Engine (引擎): 数据库连接的入口点，负责管理数据库连接池和方言。
• • Connection (连接): 与数据库进行通信的实际对象。
• • MetaData (元数据): 存储数据库模式（Schema）信息的容器，例如表、列等。
• • Table (表): 映射到数据库中的表。
• • Column (列): 映射到数据库表中的列。

3.1. 创建 Engine 和 Connection

create_engine() 函数是 SQLAlchemy 的起点，它返回一个 Engine 实例，用于管理数据库连接。连接字符串的格式通常是 dialect+driver://user:password@host:port/dbname。

对于 SQLite，连接字符串非常简单，只需要指定数据库文件的路径即可：

from sqlalchemy import create_engine# 创建一个 SQLite 内存数据库引擎# 如果是文件数据库，可以是 'sqlite:///./test.db'engine = create_engine("sqlite:///:memory:")# 建立连接connection = engine.connect()print("成功连接到数据库！")# 关闭连接connection.close()print("连接已关闭")

3.2. 定义表结构 (MetaData, Table, Column)

使用 MetaData 对象来收集数据库中所有表的信息。Table 对象用于定义数据库表，Column 对象用于定义表的列。

from sqlalchemy import MetaData, Table, Column, Integer, String, Datemetadata = MetaData()# 定义一个名为 'users' 的表users_table = Table("users", metadata,    Column("id", Integer, primary_key=True),    Column("name", String(50), nullable=False),    Column("email", String(100), unique=True),    Column("signup_date", Date))# 定义一个名为 'products' 的表products_table = Table("products", metadata,    Column("id", Integer, primary_key=True),    Column("name", String(100), nullable=False),    Column("price", Integer))print("表结构已定义")

3.3. 创建表 (DDL)

定义好表结构后，可以使用 metadata.create_all(engine) 方法将这些表创建到数据库中。

from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, Date# 创建引擎engine = create_engine("sqlite:///:memory:")metadata = MetaData()users_table = Table("users", metadata,    Column("id", Integer, primary_key=True),    Column("name", String(50), nullable=False),    Column("email", String(100), unique=True),    Column("signup_date", Date))products_table = Table("products", metadata,    Column("id", Integer, primary_key=True),    Column("name", String(100), nullable=False),    Column("price", Integer))# 创建所有表metadata.create_all(engine)print("所有表已成功创建！")

3.4. 插入数据 (DML - INSERT)

使用 insert() 构造器来创建插入语句，并通过 connection.execute() 执行。

from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, Date, insertfrom datetime import dateengine = create_engine("sqlite:///:memory:")metadata = MetaData()users_table = Table("users", metadata,    Column("id", Integer, primary_key=True),    Column("name", String(50), nullable=False),    Column("email", String(100), unique=True),    Column("signup_date", Date))metadata.create_all(engine)with engine.connect() as connection:# 插入单条数据    insert_stmt = insert(users_table).values(        name="Alice", email="alice@example.com", signup_date=date(2023, 1, 1)    )    result = connection.execute(insert_stmt)print(f"插入了 {result.rowcount} 条记录，新用户ID: {result.inserted_primary_key[0]}")# 批量插入数据    insert_many_stmt = insert(users_table)    users_data = [        {"name": "Bob", "email": "bob@example.com", "signup_date": date(2023, 2, 15)},        {"name": "Charlie", "email": "charlie@example.com", "signup_date": date(2023, 3, 10)}    ]    connection.execute(insert_many_stmt, users_data)print(f"批量插入了 {len(users_data)} 条记录")    connection.commit() # 提交事务print("数据插入完成并已提交")

3.5. 查询数据 (DML - SELECT)

使用 select() 构造器来创建查询语句。可以通过 where()、order_by()、limit() 等方法添加条件。

from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, Date, insert, selectfrom datetime import dateengine = create_engine("sqlite:///:memory:")metadata = MetaData()users_table = Table("users", metadata,    Column("id", Integer, primary_key=True),    Column("name", String(50), nullable=False),    Column("email", String(100), unique=True),    Column("signup_date", Date))metadata.create_all(engine)with engine.connect() as connection:# 插入一些数据用于查询    connection.execute(insert(users_table), [        {"name": "Alice", "email": "alice@example.com", "signup_date": date(2023, 1, 1)},        {"name": "Bob", "email": "bob@example.com", "signup_date": date(2023, 2, 15)},        {"name": "Charlie", "email": "charlie@example.com", "signup_date": date(2023, 3, 10)},        {"name": "David", "email": "david@example.com", "signup_date": date(2023, 4, 20)}    ])    connection.commit()# 查询所有用户    select_all_stmt = select(users_table)print("\n--- 所有用户 ---")for row in connection.execute(select_all_stmt):print(row)# 查询特定用户 (WHERE)    select_alice_stmt = select(users_table).where(users_table.c.name == "Alice")print("\n--- 查询 Alice ---")    alice = connection.execute(select_alice_stmt).fetchone()print(alice)# 查询部分列 (SELECT specific columns)    select_names_emails_stmt = select(users_table.c.name, users_table.c.email)print("\n--- 查询姓名和邮箱 ---")for row in connection.execute(select_names_emails_stmt):print(row)# 查询并排序 (ORDER BY)    select_ordered_stmt = select(users_table).order_by(users_table.c.signup_date.desc())print("\n--- 按注册日期降序排序 ---")for row in connection.execute(select_ordered_stmt):print(row)# 查询并限制结果 (LIMIT)    select_limited_stmt = select(users_table).limit(2)print("\n--- 限制 2 条结果 ---")for row in connection.execute(select_limited_stmt):print(row)

3.6. 更新数据 (DML - UPDATE)

使用 update() 构造器来创建更新语句，并通过 connection.execute() 执行。

from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, Date, insert, update, selectfrom datetime import dateengine = create_engine("sqlite:///:memory:")metadata = MetaData()users_table = Table("users", metadata,    Column("id", Integer, primary_key=True),    Column("name", String(50), nullable=False),    Column("email", String(100), unique=True),    Column("signup_date", Date))metadata.create_all(engine)with engine.connect() as connection:# 插入一些数据用于更新    connection.execute(insert(users_table), [        {"name": "Alice", "email": "alice@example.com", "signup_date": date(2023, 1, 1)},        {"name": "Bob", "email": "bob@example.com", "signup_date": date(2023, 2, 15)}    ])    connection.commit()# 更新 Alice 的邮箱    update_stmt = update(users_table).where(users_table.c.name == "Alice").values(email="alice_new@example.com")    result = connection.execute(update_stmt)    connection.commit()print(f"更新了 {result.rowcount} 条记录")# 验证更新    updated_alice = connection.execute(select(users_table).where(users_table.c.name == "Alice")).fetchone()print(f"更新后的 Alice: {updated_alice}")

3.7. 删除数据 (DML - DELETE)

使用 delete() 构造器来创建删除语句，并通过 connection.execute() 执行。

from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, Date, insert, delete, selectfrom datetime import dateengine = create_engine("sqlite:///:memory:")metadata = MetaData()users_table = Table("users", metadata,    Column("id", Integer, primary_key=True),    Column("name", String(50), nullable=False),    Column("email", String(100), unique=True),    Column("signup_date", Date))metadata.create_all(engine)with engine.connect() as connection:# 插入一些数据用于删除    connection.execute(insert(users_table), [        {"name": "Alice", "email": "alice@example.com", "signup_date": date(2023, 1, 1)},        {"name": "Bob", "email": "bob@example.com", "signup_date": date(2023, 2, 15)}    ])    connection.commit()# 删除 Bob    delete_stmt = delete(users_table).where(users_table.c.name == "Bob")    result = connection.execute(delete_stmt)    connection.commit()print(f"删除了 {result.rowcount} 条记录")# 验证删除    remaining_users = connection.execute(select(users_table)).fetchall()print(f"剩余用户: {remaining_users}")

3.8. 事务管理

SQLAlchemy Core 默认使用事务。每次 connection.execute() 调用都会在一个事务中执行。当你使用 with engine.connect() as connection: 语句时，如果块内的所有操作都成功，事务会自动提交；如果发生异常，事务会自动回滚。

你也可以手动控制事务：

from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, Date, insertfrom datetime import dateengine = create_engine("sqlite:///:memory:")metadata = MetaData()users_table = Table("users", metadata,    Column("id", Integer, primary_key=True),    Column("name", String(50), nullable=False),    Column("email", String(100), unique=True),    Column("signup_date", Date))metadata.create_all(engine)connection = engine.connect()# 开始一个事务transaction = connection.begin()try:    connection.execute(insert(users_table).values(name="Frank", email="frank@example.com", signup_date=date(2023, 5, 1)))# 模拟一个错误# connection.execute(insert(users_table).values(name="Frank", email="frank@example.com", signup_date=date(2023, 5, 1))) # 再次插入相同 email 会导致唯一约束错误    transaction.commit() # 提交事务print("事务成功提交")except Exception as e:    transaction.rollback() # 回滚事务print(f"事务回滚: {e}")finally:    connection.close()

4. SQLAlchemy ORM 基础

SQLAlchemy ORM 允许你将 Python 类映射到数据库表，并通过操作 Python 对象来间接操作数据库。这使得数据库操作更加面向对象，减少了直接编写 SQL 的需求。

4.1. 声明式基类 (Declarative Base)

ORM 的核心是声明式系统，它允许你将 Python 类定义为数据库表的映射。declarative_base() 函数创建了一个基类，你的所有 ORM 模型都将继承自它。

from sqlalchemy import create_engine, Column, Integer, String, Datefrom sqlalchemy.orm import declarative_base, sessionmakerfrom datetime import date# 1. 创建 Engineengine = create_engine("sqlite:///:memory:")# 2. 创建声明式基类Base = declarative_base()# 3. 定义 ORM 模型 (映射到数据库表)classUser(Base):    __tablename__ = "users"id = Column(Integer, primary_key=True)    name = Column(String(50), nullable=False)    email = Column(String(100), unique=True)    signup_date = Column(Date)def__repr__(self):returnf"<User(id={self.id}, name='{self.name}', email='{self.email}')>"# 4. 创建所有表Base.metadata.create_all(engine)print("ORM 模型对应的表已成功创建！")

4.2. Session (会话)

Session 是 ORM 中最重要的概念之一。它代表了与数据库的一次对话，负责管理对象的生命周期（加载、保存、更新、删除）以及事务。所有的数据库操作都通过 Session 进行。

sessionmaker 是一个工厂函数，用于创建 Session 类。

from sqlalchemy import create_engine, Column, Integer, String, Datefrom sqlalchemy.orm import declarative_base, sessionmakerfrom datetime import dateengine = create_engine("sqlite:///:memory:")Base = declarative_base()classUser(Base):    __tablename__ = "users"id = Column(Integer, primary_key=True)    name = Column(String(50), nullable=False)    email = Column(String(100), unique=True)    signup_date = Column(Date)def__repr__(self):returnf"<User(id={self.id}, name='{self.name}', email='{self.email}')>"Base.metadata.create_all(engine)# 创建 Session 工厂Session = sessionmaker(bind=engine)# 创建一个 Session 实例session = Session()# 使用 Session 进行操作 (稍后会详细介绍)# 关闭 Sessionsession.close()print("Session 已创建并关闭")

4.3. ORM CRUD 操作

4.3.1. 创建 (Create)

创建新的 Python 对象，然后将其添加到 Session 并提交。

from sqlalchemy import create_engine, Column, Integer, String, Datefrom sqlalchemy.orm import declarative_base, sessionmakerfrom datetime import dateengine = create_engine("sqlite:///:memory:")Base = declarative_base()classUser(Base):    __tablename__ = "users"id = Column(Integer, primary_key=True)    name = Column(String(50), nullable=False)    email = Column(String(100), unique=True)    signup_date = Column(Date)def__repr__(self):returnf"<User(id={self.id}, name='{self.name}', email='{self.email}')>"Base.metadata.create_all(engine)Session = sessionmaker(bind=engine)with Session() as session:# 创建 User 对象    new_user = User(name="Eve", email="eve@example.com", signup_date=date(2023, 6, 1))# 添加到 Session    session.add(new_user)# 提交到数据库    session.commit()print(f"新用户已创建: {new_user}")print(f"新用户ID: {new_user.id}") # 提交后可以获取到 ID# 批量添加    users_to_add = [        User(name="Frank", email="frank@example.com", signup_date=date(2023, 7, 10)),        User(name="Grace", email="grace@example.com", signup_date=date(2023, 8, 5))    ]    session.add_all(users_to_add)    session.commit()print(f"批量添加了 {len(users_to_add)} 个用户")

4.3.2. 查询 (Read)

使用 session.query() 或 session.scalars(select(...)) 来构建查询。SQLAlchemy 2.0 推荐使用 select() 构造器。

from sqlalchemy import create_engine, Column, Integer, String, Date, selectfrom sqlalchemy.orm import declarative_base, sessionmakerfrom datetime import dateengine = create_engine("sqlite:///:memory:")Base = declarative_base()classUser(Base):    __tablename__ = "users"id = Column(Integer, primary_key=True)    name = Column(String(50), nullable=False)    email = Column(String(100), unique=True)    signup_date = Column(Date)def__repr__(self):returnf"<User(id={self.id}, name='{self.name}', email='{self.email}')>"Base.metadata.create_all(engine)Session = sessionmaker(bind=engine)with Session() as session:# 插入一些数据用于查询    session.add_all([        User(name="Alice", email="alice@example.com", signup_date=date(2023, 1, 1)),        User(name="Bob", email="bob@example.com", signup_date=date(2023, 2, 15)),        User(name="Charlie", email="charlie@example.com", signup_date=date(2023, 3, 10))    ])    session.commit()# 查询所有用户print("\n--- 所有用户 ---")for user in session.scalars(select(User)).all():print(user)# 按 ID 查询单个用户print("\n--- 按 ID 查询用户 (ID=1) ---")    user_by_id = session.get(User, 1) # 推荐使用 get() 方法按主键查询print(user_by_id)# 按条件查询用户 (WHERE)print("\n--- 查询名为 Bob 的用户 ---")    user_bob = session.scalars(select(User).where(User.name == "Bob")).first()print(user_bob)# 查询并排序 (ORDER BY)print("\n--- 按注册日期降序排序 ---")for user in session.scalars(select(User).order_by(User.signup_date.desc())).all():print(user)# 查询并限制结果 (LIMIT)print("\n--- 限制 1 条结果 ---")    user_limited = session.scalars(select(User).limit(1)).first()print(user_limited)

4.3.3. 更新 (Update)

从 Session 中加载对象，修改其属性，然后提交 Session。

from sqlalchemy import create_engine, Column, Integer, String, Date, selectfrom sqlalchemy.orm import declarative_base, sessionmakerfrom datetime import dateengine = create_engine("sqlite:///:memory:")Base = declarative_base()classUser(Base):    __tablename__ = "users"id = Column(Integer, primary_key=True)    name = Column(String(50), nullable=False)    email = Column(String(100), unique=True)    signup_date = Column(Date)def__repr__(self):returnf"<User(id={self.id}, name='{self.name}', email='{self.email}')>"Base.metadata.create_all(engine)Session = sessionmaker(bind=engine)with Session() as session:# 插入一个用户用于更新    session.add(User(name="David", email="david@example.com", signup_date=date(2023, 9, 1)))    session.commit()# 加载要更新的用户    user_to_update = session.scalars(select(User).where(User.name == "David")).first()if user_to_update:# 修改对象属性        user_to_update.email = "david_new@example.com"        user_to_update.signup_date = date(2023, 9, 10)# 提交 Session        session.commit()print(f"用户已更新: {user_to_update}")else:print("未找到用户 David")

4.3.4. 删除 (Delete)

从 Session 中加载对象，然后使用 session.delete() 删除，并提交 Session。

from sqlalchemy import create_engine, Column, Integer, String, Date, selectfrom sqlalchemy.orm import declarative_base, sessionmakerfrom datetime import dateengine = create_engine("sqlite:///:memory:")Base = declarative_base()classUser(Base):    __tablename__ = "users"id = Column(Integer, primary_key=True)    name = Column(String(50), nullable=False)    email = Column(String(100), unique=True)    signup_date = Column(Date)def__repr__(self):returnf"<User(id={self.id}, name='{self.name}', email='{self.email}')>"Base.metadata.create_all(engine)Session = sessionmaker(bind=engine)with Session() as session:# 插入一个用户用于删除    session.add(User(name="Eve", email="eve@example.com", signup_date=date(2023, 10, 1)))    session.commit()# 加载要删除的用户    user_to_delete = session.scalars(select(User).where(User.name == "Eve")).first()if user_to_delete:# 从 Session 中删除对象        session.delete(user_to_delete)# 提交 Session        session.commit()print(f"用户已删除: {user_to_delete}")else:print("未找到用户 Eve")# 验证删除    remaining_users = session.scalars(select(User)).all()print(f"剩余用户: {remaining_users}")

5. Core 与 ORM 的选择

建议：

• • 对于需要精细控制 SQL 语句、进行复杂查询优化或执行 DDL 操作的场景，优先考虑 SQLAlchemy Core。
• • 对于大多数业务逻辑、CRUD 操作以及需要快速开发的场景，SQLAlchemy ORM 是更好的选择。它提供了更高级别的抽象，让你可以专注于业务逻辑而不是数据库细节。
• • 两者可以混合使用，例如使用 Core 进行 DDL 操作，然后使用 ORM 进行数据操作。

6. SQLAlchemy ORM 进阶：关联关系

在现实世界的应用中，表与表之间通常存在关联关系。SQLAlchemy ORM 提供了强大的工具来处理这些关系。

6.1. 一对多关系 (One-to-Many)

一对多关系是最常见的关系类型。例如，一个用户可以有多个订单。

from sqlalchemy import create_engine, Column, Integer, String, ForeignKeyfrom sqlalchemy.orm import declarative_base, sessionmaker, relationshipBase = declarative_base()classUser(Base):    __tablename__ = "users"id = Column(Integer, primary_key=True)    name = Column(String(50), nullable=False)# 定义关系：一个用户可以有多个订单# back_populates 用于建立双向关系    orders = relationship("Order", back_populates="user")def__repr__(self):returnf"<User(id={self.id}, name='{self.name}')>"classOrder(Base):    __tablename__ = "orders"id = Column(Integer, primary_key=True)    product_name = Column(String(100), nullable=False)# 定义外键：指向 users 表的 id 列    user_id = Column(Integer, ForeignKey("users.id"))# 定义关系：一个订单属于一个用户    user = relationship("User", back_populates="orders")def__repr__(self):returnf"<Order(id={self.id}, product='{self.product_name}', user_id={self.user_id})>"# 演示使用engine = create_engine("sqlite:///:memory:")Base.metadata.create_all(engine)Session = sessionmaker(bind=engine)with Session() as session:# 创建用户和订单    alice = User(name="Alice")    order1 = Order(product_name="Laptop", user=alice)    order2 = Order(product_name="Mouse", user=alice)    session.add(alice) # 添加用户会自动添加关联的订单    session.commit()# 查询用户及其订单    user = session.get(User, 1)print(f"用户: {user.name}")print(f"订单: {user.orders}")

6.2. 多对多关系 (Many-to-Many)

多对多关系需要一个中间表（关联表）来存储两个表之间的映射。例如，一个学生可以选修多门课程，一门课程也可以被多个学生选修。

from sqlalchemy import create_engine, Column, Integer, String, Table, ForeignKeyfrom sqlalchemy.orm import declarative_base, sessionmaker, relationshipBase = declarative_base()# 定义关联表student_course_association = Table("student_course", Base.metadata,    Column("student_id", Integer, ForeignKey("students.id"), primary_key=True),    Column("course_id", Integer, ForeignKey("courses.id"), primary_key=True))classStudent(Base):    __tablename__ = "students"id = Column(Integer, primary_key=True)    name = Column(String(50), nullable=False)# 定义多对多关系    courses = relationship("Course", secondary=student_course_association, back_populates="students")def__repr__(self):returnf"<Student(id={self.id}, name='{self.name}')>"classCourse(Base):    __tablename__ = "courses"id = Column(Integer, primary_key=True)    title = Column(String(100), nullable=False)# 定义多对多关系    students = relationship("Student", secondary=student_course_association, back_populates="courses")def__repr__(self):returnf"<Course(id={self.id}, title='{self.title}')>"# 演示使用engine = create_engine("sqlite:///:memory:")Base.metadata.create_all(engine)Session = sessionmaker(bind=engine)with Session() as session:# 创建学生和课程    s1 = Student(name="Alice")    s2 = Student(name="Bob")    c1 = Course(title="Python Programming")    c2 = Course(title="Database Systems")# 建立关联    s1.courses.append(c1)    s1.courses.append(c2)    s2.courses.append(c1)    session.add_all([s1, s2])    session.commit()# 查询    student = session.get(Student, 1)print(f"学生 {student.name} 选修的课程: {[c.title for c in student.courses]}")    course = session.get(Course, 1)print(f"选修课程 {course.title} 的学生: {[s.name for s in course.students]}")

6.3. 级联操作 (Cascades)

级联操作定义了当对父对象执行操作时，如何影响关联的子对象。常见的级联类型包括：

• • save-update: (默认) 当父对象被保存或更新时，子对象也会被保存或更新。
• • delete: 当父对象被删除时，子对象也会被删除。
• • all, delete-orphan: 当子对象从父对象的集合中移除时，子对象会被删除。

# 在 relationship 中定义级联orders = relationship("Order", back_populates="user", cascade="all, delete-orphan")

7. SQLAlchemy 高级特性与性能优化

7.1. 加载策略 (Loading Strategies)

当查询带有关系的对象时，SQLAlchemy 提供了不同的加载策略来平衡查询次数和数据量。

• • Lazy Loading (延迟加载): (默认) 只有在访问关联属性时才发出 SQL 查询。这可能导致 "N+1 查询问题"。
• • Joined Eager Loading (连接预加载): 使用 JOIN 在单次查询中获取关联对象。适用于一对一或多对一关系。
• • Selectin Eager Loading (Selectin 预加载): 发出第二次查询来获取所有关联对象。适用于一对多或多对多关系，通常性能最好。

from sqlalchemy.orm import joinedload, selectinload# 使用 joinedloaduser = session.scalars(select(User).options(joinedload(User.orders))).first()# 使用 selectinloadusers = session.scalars(select(User).options(selectinload(User.orders))).all()

7.2. 性能优化建议

1. 1. 避免 N+1 问题: 始终根据需要选择合适的预加载策略（joinedload 或 selectinload）。
2. 2. 只查询需要的列: 不要总是使用 select(User)，如果只需要姓名，使用 select(User.name)。
3. 3. 批量操作: 使用 session.add_all() 或 Core 的批量插入来减少往返次数。
4. 4. 使用索引: 确保数据库中经常查询和连接的列上有索引。
5. 5. 连接池: SQLAlchemy 默认管理连接池，但在高并发环境下，可能需要调整 pool_size 和 max_overflow。

7.3. 数据库迁移 (Alembic)

在实际开发中，数据库模式会随时间变化。Alembic 是 SQLAlchemy 的官方数据库迁移工具，它允许你像管理代码版本一样管理数据库模式的变化。

基本流程：

1. 1. 安装 Alembic: pip install alembic
2. 2. 初始化: alembic init migrations
3. 3. 配置 alembic.ini 中的数据库连接。
4. 4. 在 env.py 中导入你的模型 Base.metadata。
5. 5. 生成迁移脚本: alembic revision --autogenerate -m "create users table"
6. 6. 应用迁移: alembic upgrade head

8. 总结与最佳实践

SQLAlchemy 是一个功能极其丰富的库，掌握它需要时间和实践。以下是一些核心建议：

• • 优先使用 ORM 进行业务逻辑开发，因为它更易于维护和测试。
• • 在性能关键点切换到 Core，或者使用原生 SQL。
• • 始终使用参数化查询（SQLAlchemy 默认会这样做），以防止 SQL 注入。
• • 保持 Session 的生命周期尽可能短，通常是一个请求一个 Session。
• • 使用 Alembic 管理数据库变更，不要手动修改生产数据库。