Python实用工具：SQLAlchemy零基础入门教程

作者：

在

一、SQLAlchemy 核心介绍

SQLAlchemy是Python生态中功能强大的ORM（对象关系映射） 库，它能将Python类与数据库表进行映射，让开发者通过操作Python对象来实现数据库的增删改查，无需编写复杂的原生SQL语句。其工作原理是建立对象模型与关系模型的映射桥梁，通过SQL表达式语言和ORM两层架构，实现对多种数据库的兼容操作。

优点方面，它支持MySQL、PostgreSQL、SQLite等主流数据库，具备灵活的查询构造能力，事务处理机制完善，且能兼顾底层SQL的优化需求；缺点是入门门槛略高于轻量级ORM库，简单场景下配置相对繁琐。SQLAlchemy采用MIT开源许可证，允许自由使用、修改和分发，无商业使用限制。

二、SQLAlchemy 安装步骤

对于技术小白来说，SQLAlchemy的安装非常简单，只需要使用Python的包管理工具pip即可完成，具体步骤如下：

检查pip环境：打开命令行终端（Windows下是CMD或PowerShell，Mac和Linux下是Terminal），输入以下命令验证pip是否可用
bash pip --version
如果能正常显示pip的版本号，说明环境没问题；如果提示“找不到命令”，则需要先配置Python的环境变量。
执行安装命令：在终端中输入以下命令，安装最新版本的SQLAlchemy
bash pip install sqlalchemy
验证安装结果：安装完成后，在终端中输入Python交互式环境，执行以下代码
python import sqlalchemy print(sqlalchemy.__version__)
如果能正常输出SQLAlchemy的版本号（例如2.0.23），则说明安装成功。

提示：如果需要连接特定的数据库（如MySQL），还需要安装对应的数据库驱动，例如pip install pymysql；连接PostgreSQL则需要安装psycopg2-binary。

三、SQLAlchemy 核心使用方式

3.1 核心概念梳理

在使用SQLAlchemy之前，我们需要先了解几个核心概念，这对后续的学习至关重要：

Engine（引擎）：负责管理数据库连接池，是SQLAlchemy与数据库交互的核心入口。
Session（会话）：用于执行数据库操作的“工作区”，所有的增删改查操作都需要通过Session来执行。
Model（模型）：继承自declarative_base的Python类，每个类对应数据库中的一张表，类的属性对应表的字段。
MetaData（元数据）：用于存储数据库表结构的相关信息，ORM模式下会自动生成。

3.2 建立数据库连接

首先我们需要创建一个数据库引擎，不同数据库的连接字符串格式略有不同，下面以常用的SQLite（无需额外配置，文件型数据库）和MySQL为例进行演示。

3.2.1 连接SQLite数据库

SQLite数据库无需安装服务端，直接通过文件路径即可连接，适合本地测试和小型项目。

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# 创建SQLite引擎，echo=True表示打印执行的SQL语句，方便调试
engine = create_engine('sqlite:///test.db', echo=True)

# 创建Session类，绑定到上面的引擎
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

代码说明：

sqlite:///test.db 表示数据库文件test.db位于当前目录下，如果文件不存在，SQLAlchemy会自动创建。
autocommit=False 表示关闭自动提交，所有操作需要手动提交事务。
autoflush=False 表示关闭自动刷新，避免不必要的数据库交互。

3.2.2 连接MySQL数据库

连接MySQL需要先安装驱动（如pymysql），然后使用对应的连接字符串。

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# 安装驱动：pip install pymysql
# 连接字符串格式：mysql+pymysql://用户名:密码@主机地址:端口号/数据库名
engine = create_engine('mysql+pymysql://root:123456@localhost:3306/test_db', echo=True)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

代码说明：

请将root替换为你的MySQL用户名，123456替换为密码，test_db替换为需要连接的数据库名（需提前在MySQL中创建）。

3.3 定义数据模型

数据模型是Python类与数据库表的映射载体，我们需要继承declarative_base来创建模型类。

from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime

# 创建基类，所有模型类都需要继承这个基类
Base = declarative_base()

# 定义User模型，对应数据库中的user表
class User(Base):
    # 定义表名
    __tablename__ = 'user'

    # 定义表字段
    id = Column(Integer, primary_key=True, autoincrement=True, comment='用户ID')
    name = Column(String(50), nullable=False, comment='用户姓名')
    age = Column(Integer, nullable=True, comment='用户年龄')
    create_time = Column(DateTime, default=datetime.now, comment='创建时间')

    # 定义__repr__方法，方便打印对象时查看信息
    def __repr__(self):
        return f"<User(id={self.id}, name='{self.name}', age={self.age})>"

代码说明：

__tablename__ 属性指定模型对应的数据库表名，如果不指定，SQLAlchemy会默认使用类名的小写形式作为表名。
Column 用于定义表字段，参数说明：
Integer/String/DateTime 表示字段的数据类型；
primary_key=True 表示该字段是主键；
autoincrement=True 表示主键自增（仅适用于整数类型）；
nullable=False 表示该字段不允许为空；
default 表示字段的默认值。

3.4 创建数据库表

定义好模型后，我们需要通过create_all方法来创建对应的数据库表，执行以下代码即可：

# 基于引擎创建所有定义的表
Base.metadata.create_all(bind=engine)

代码说明：

执行该代码后，SQLAlchemy会检查数据库中是否存在user表，如果不存在则自动创建；如果已存在，则不会重复创建，也不会修改现有表结构。

3.5 数据库基本操作（CRUD）

CRUD是数据库操作的核心，即创建（Create）、查询（Read）、更新（Update）、删除（Delete），下面我们通过Session来实现这些操作。

3.5.1 创建数据（新增用户）

新增数据的步骤是：创建Session实例 → 实例化模型类 → 将对象添加到Session → 提交事务 → 关闭Session。

# 创建Session实例
db = SessionLocal()

# 方式1：单个新增
user1 = User(name='张三', age=25)
db.add(user1)

# 方式2：批量新增
user2 = User(name='李四', age=30)
user3 = User(name='王五', age=28)
db.add_all([user2, user3])

# 提交事务，这一步才会真正将数据写入数据库
db.commit()

# 刷新对象，获取数据库自动生成的id等属性
db.refresh(user1)
print(user1)  # 输出：<User(id=1, name='张三', age=25)>

# 关闭Session
db.close()

代码说明：

db.add() 用于添加单个对象，db.add_all() 用于添加多个对象。
db.commit() 必须执行，否则所有操作都只是在本地Session中，不会同步到数据库。
db.refresh() 用于从数据库中获取最新的对象数据，例如自增的id字段。

3.5.2 查询数据（读取用户）

SQLAlchemy提供了灵活的查询方式，支持简单查询、条件查询、排序、分页等操作，查询的核心是db.query()方法。

db = SessionLocal()

# 1. 查询所有用户
all_users = db.query(User).all()
print("所有用户：", all_users)

# 2. 查询单个用户（根据主键查询）
user = db.query(User).get(1)  # get方法根据主键查询，不存在返回None
print("主键为1的用户：", user)

# 3. 条件查询（filter）
# 查询年龄大于25的用户
users_gt_25 = db.query(User).filter(User.age > 25).all()
print("年龄大于25的用户：", users_gt_25)

# 查询姓名为“李四”的用户
user_li = db.query(User).filter(User.name == '李四').first()  # first()返回第一条数据，不存在返回None
print("姓名为李四的用户：", user_li)

# 4. 排序查询（order_by）
# 按年龄升序排序
sorted_users = db.query(User).order_by(User.age.asc()).all()
print("按年龄升序排序的用户：", sorted_users)

# 5. 分页查询（slice）
# 查询第2-3条数据（索引从0开始）
page_users = db.query(User).slice(1, 3).all()
print("分页查询结果：", page_users)

db.close()

代码说明：

all() 返回所有符合条件的结果列表，first() 返回第一条结果，get() 根据主键查询。
filter() 用于添加查询条件，支持==、>、<、!=等运算符，还可以通过and_、or_组合多条件。
order_by() 用于排序，asc() 升序，desc() 降序。
slice(start, end) 用于分页，start 是起始索引，end 是结束索引（不包含）。

3.5.3 更新数据（修改用户信息）

更新数据的步骤是：查询到需要修改的对象 → 修改对象的属性 → 提交事务。

db = SessionLocal()

# 1. 先查询再更新
user = db.query(User).filter(User.name == '张三').first()
if user:
    user.age = 26  # 修改年龄
    db.commit()  # 提交事务
    db.refresh(user)
    print("更新后的用户：", user)  # 输出：<User(id=1, name='张三', age=26)>

# 2. 批量更新（无需查询对象）
db.query(User).filter(User.age > 25).update({User.age: User.age + 1})
db.commit()
print("批量更新后年龄大于25的用户：", db.query(User).filter(User.age > 25).all())

db.close()

代码说明：

方式1适合单条数据的更新，需要先查询到对象再修改属性；
方式2适合批量更新，直接通过update()方法修改，效率更高，无需查询对象。

3.5.4 删除数据（删除用户）

删除数据的步骤是：查询到需要删除的对象 → 调用delete()方法 → 提交事务。

db = SessionLocal()

# 1. 单条数据删除
user = db.query(User).get(3)  # 删除主键为3的用户
if user:
    db.delete(user)
    db.commit()
    print("删除后的所有用户：", db.query(User).all())

# 2. 批量数据删除
db.query(User).filter(User.age > 28).delete()
db.commit()
print("批量删除后剩余用户：", db.query(User).all())

db.close()

代码说明：

删除操作执行后，必须调用db.commit()才能生效；
批量删除时，通过filter()添加条件，直接删除符合条件的所有数据。

四、实际案例：用户信息管理系统

为了让大家更好地掌握SQLAlchemy的使用，我们结合一个实际案例——用户信息管理系统，实现用户的新增、查询、修改、删除功能，代码如下：

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

# 1. 创建引擎和Session
engine = create_engine('sqlite:///user_manage.db', echo=True)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# 2. 定义用户模型
class User(Base):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(50), nullable=False)
    age = Column(Integer, nullable=True)
    gender = Column(String(10), nullable=True)
    create_time = Column(DateTime, default=datetime.now)

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

# 3. 创建数据库表
Base.metadata.create_all(bind=engine)

# 4. 定义操作函数
def get_db():
    """获取数据库Session，自动关闭"""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

def add_user(name, age, gender):
    """新增用户"""
    db = next(get_db())
    user = User(name=name, age=age, gender=gender)
    db.add(user)
    db.commit()
    db.refresh(user)
    return user

def query_user(user_id=None, name=None):
    """查询用户，支持按ID或姓名查询"""
    db = next(get_db())
    if user_id:
        return db.query(User).get(user_id)
    elif name:
        return db.query(User).filter(User.name == name).all()
    else:
        return db.query(User).all()

def update_user(user_id, **kwargs):
    """更新用户信息"""
    db = next(get_db())
    user = db.query(User).get(user_id)
    if not user:
        return None
    for key, value in kwargs.items():
        if hasattr(user, key):
            setattr(user, key, value)
    db.commit()
    db.refresh(user)
    return user

def delete_user(user_id):
    """删除用户"""
    db = next(get_db())
    user = db.query(User).get(user_id)
    if not user:
        return False
    db.delete(user)
    db.commit()
    return True

# 5. 测试功能
if __name__ == '__main__':
    # 新增用户
    print("=== 新增用户 ===")
    user1 = add_user("张三", 25, "男")
    user2 = add_user("李四", 30, "女")
    print(f"新增用户：{user1}, {user2}")

    # 查询用户
    print("\n=== 查询所有用户 ===")
    all_users = query_user()
    print(all_users)

    print("\n=== 按姓名查询用户 ===")
    li_users = query_user(name="李四")
    print(li_users)

    # 更新用户
    print("\n=== 更新用户信息 ===")
    updated_user = update_user(1, age=26, gender="男")
    print(f"更新后的用户：{updated_user}")

    # 删除用户
    print("\n=== 删除用户 ===")
    result = delete_user(2)
    print(f"删除是否成功：{result}")
    print(f"删除后剩余用户：{query_user()}")

代码说明：

get_db() 函数通过生成器实现Session的自动创建和关闭，避免手动关闭的繁琐；
add_user()、query_user()、update_user()、delete_user() 四个函数分别实现用户的增删改查功能；
在if __name__ == '__main__' 代码块中，我们测试了所有功能，运行后可以看到完整的操作流程和结果。

五、相关资源地址

Pypi地址：https://pypi.org/project/SQLAlchemy
Github地址：https://github.com/sqlalchemy/sqlalchemy
官方文档地址：https://docs.sqlalchemy.org/en/20/

这个案例覆盖了SQLAlchemy的核心使用场景，小白可以直接复制代码运行，然后根据自己的需求修改字段和功能，快速上手实际开发。{ Environment.NewLine }{ Environment.NewLine }关注我，每天分享一个实用的Python自动化工具。

实用工具

Python实用工具：SQLAlchemy零基础入门教程

一、SQLAlchemy 核心介绍

二、SQLAlchemy 安装步骤

三、SQLAlchemy 核心使用方式

3.1 核心概念梳理

3.2 建立数据库连接

3.2.1 连接SQLite数据库

3.2.2 连接MySQL数据库

3.3 定义数据模型

3.4 创建数据库表

3.5 数据库基本操作（CRUD）

3.5.1 创建数据（新增用户）

3.5.2 查询数据（读取用户）

3.5.3 更新数据（修改用户信息）

3.5.4 删除数据（删除用户）

四、实际案例：用户信息管理系统

五、相关资源地址

更多文章

Python实用工具Upgini：零代码特征工程与数据集增强教程

Python实用工具库excalibur：PDF表格提取与数据处理实战教程

Python实用工具：rows库快速入门与实战指南

Python实用工具：Camelot库——轻松提取PDF表格数据的完整指南