💥 新手友好、全程实战、可直接复刻运行

做Python后端开发，SQLAlchemy是绕不开的数据库神器！作为Python生态最主流的ORM框架，彻底告别手写原生SQL，用面向对象的方式操作数据库，适配所有主流数据库，企业级项目必备！

一、什么是SQLAlchemy？

1.1 核心定义

SQLAlchemy /ˈælkəmi/（艾尔克米，炼金术），是Python最强大、最流行的ORM（对象关系映射）框架。

核心作用：将Python对象与数据库数据表做映射，开发者只需操作Python对象，无需手写原生SQL（兼容混合使用），自动转换为数据库语句，极大提升开发效率与代码可维护性。

1.2 核心优势

• 跨库兼容：一套代码无缝适配 MySQL、PostgreSQL、SQLite、Oracle 等主流数据库

• 解耦性强：代码与数据库解绑，切换数据库无需大规模改代码

• 自动类型转换：Python数据类型自动映射数据库字段类型

• 功能强大：完善的事务、关联查询、分组聚合、索引约束能力

• 企业级首选：FastAPI/Flask/Django 后端项目标配数据库框架

1.3 两大核心模块

SQLAlchemy整体分为两层，学习重点为 ORM层：

• Core（底层核心）：基础引擎、连接管理、SQL表达式构建，ORM的底层依赖

• ORM（上层封装）：对象关系映射，面向对象操作数据库，日常开发核心使用

二、SQLAlchemy核心概念（必懂）

2.1 五大核心组件

• Engine 引擎：数据库连接核心入口，管理连接池、数据库地址、账号密码，所有操作的根基

• Session 会话：数据库交互桥梁，所有增删改查操作必须通过Session完成

• Base 基类：所有数据库模型类的父类，子类自动映射为数据库表

• Model 模型：Python自定义类，对应数据库一张表，类属性对应表字段

• Column 字段：定义表字段，支持类型、主键、非空、唯一、默认值、索引等约束

2.2 ORM映射核心原理

ORM（对象关系映射）核心对应关系：

• Python 模型类 ➡️ 数据库 数据表

• 类属性 ➡️ 数据表 字段

• 类实例对象 ➡️ 数据表 一行数据

三、环境搭建与安装

3.1 安装依赖

根据使用的数据库安装对应驱动，SQLite无需额外驱动（Python自带）

# 安装核心框架
pip install sqlalchemy

# MySQL驱动（二选一）
pip install pymysql
pip install mysql-connector-python

# PostgreSQL驱动
pip install psycopg2-binary

# Oracle驱动
pip install cx_Oracle

3.2 验证安装成功

import sqlalchemy
import pymysql

# 打印版本，无报错即安装成功
print("SQLAlchemy版本：", sqlalchemy.__version__)
print("pymysql版本：", pymysql.__version__)

四、数据库前置准备

SQLAlchemy仅负责操作数据表，不会自动创建数据库，需手动提前创建：

# 登录MySQL
mysql -u root -p

# 创建测试数据库（支持utf8mb4全字符集）
CREATE DATABASE IF NOT EXISTS sqlalchemy_demo CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;

# 查看数据库
SHOW DATABASES;

五、项目基础配置（通用模板）

新建 sqlalchemy_config.py，统一管理引擎、会话、基类，全局复用

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine

# 1. 数据库连接字符串：mysql+驱动://账号:密码@地址:端口/数据库名?编码
db_url = "mysql+pymysql://root:123456@localhost:3306/sqlalchemy_demo?charset=utf8mb4"

# 2. 创建引擎，echo=True 打印执行SQL（调试开启，生产关闭）
engine = create_engine(db_url, echo=True)

# 3. 创建会话工厂，绑定引擎
SessionFactory = sessionmaker(bind=engine)

# 4. 创建ORM基类，所有模型继承此类
Base = declarative_base()

六、定义数据库模型（Model）

新建 base_model.py，定义用户表模型，包含常用字段与约束

from sqlalchemy import Column, Integer, String, DateTime
from datetime import datetime
from sqlalchemy_config import Base

# 用户表模型
class User(Base):
    # 指定数据库表名
    __tablename__ = "user"

    # 主键ID：自增、非空、主键
    id = Column(Integer, primary_key=True, autoincrement=True, comment="用户ID")
    # 用户名：最长50字符、非空、唯一
    username = Column(String(50), nullable=False, unique=True, comment="用户名")
    # 密码：最长100字符、非空
    password = Column(String(100), nullable=False, comment="登录密码")
    # 创建时间：默认当前时间
    create_time = Column(DateTime, default=datetime.now, comment="创建时间")

    # 重写打印方法，方便调试查看数据
    def __repr__(self):
        return f"User(id={self.id}, username={self.username}, create_time={self.create_time})"

6.1 常用字段类型

• Integer：整型（MySQL INT）

• String(n)：字符串（MySQL VARCHAR(n)）

• DateTime：时间日期（MySQL DATETIME）

• Float：浮点型

• Boolean：布尔型（MySQL TINYINT(1)）

6.2 常用字段约束

参数	作用说明
primary_key=True	设置为主键
autoincrement=True	整数主键自增
nullable=False	字段非空
unique=True	字段唯一约束
default=xxx	代码层面默认值
index=True	创建普通索引
comment="xxx"	字段注释

七、自动创建数据表

执行以下代码，根据模型自动生成数据表（重复执行不会报错、不会重复创建）

from sqlalchemy_config import engine, Base
from base_model import User

# 根据所有继承Base的模型创建数据表
Base.metadata.create_all(engine)
print("✅ 所有数据表创建成功！")

八、全套CRUD实战操作（核心重点）

统一使用with 上下文管理器，自动关闭会话，避免连接泄露

8.1 新增数据 Create

支持单条新增、批量对象新增、批量字典新增、2.0新版语法

from sqlalchemy_config import SessionFactory
from base_model import User
from sqlalchemy import insert

# 1. 单条新增
with SessionFactory() as session:
    user = User(username="zhangsan", password="123456")
    session.add(user)
    session.commit()
    print("✅ 单条数据新增成功")

# 2. 批量对象新增
with SessionFactory() as session:
    user1 = User(username="lisi", password="123456")
    user2 = User(username="wangwu", password="654321")
    session.add_all([user1, user2])
    session.commit()
    print("✅ 批量对象新增成功")

# 3. 批量字典新增（高效大数据量）
with SessionFactory() as session:
    user_list = [
        {"username": "sunqi", "password": "123123"},
        {"username": "zhouba", "password": "456456"}
    ]
    session.bulk_insert_mappings(User, user_list)
    session.commit()
    print("✅ 批量字典新增成功")

# 4. SQLAlchemy2.0 新版insert语法
with SessionFactory() as session:
    stmt = insert(User).values([
        {"username": "zhengshi", "password": "aaa123"},
        {"username": "chenshi", "password": "bbb456"}
    ])
    session.execute(stmt)
    session.commit()
    print("✅ 新版语法新增成功")

8.2 查询数据 Read（最全查询合集）

8.2.1 基础查询

from sqlalchemy_config import SessionFactory
from base_model import User

with SessionFactory() as session:
    # 1. 查询所有数据
    all_users = session.query(User).all()
    print("所有用户：", all_users)

    # 2. 根据主键查询单条
    user = session.get(User, 1)
    print("ID=1的用户：", user)

    # 3. 查询第一条数据
    first_user = session.query(User).first()
    print("第一条用户数据：", first_user)

    # 4. 查询指定字段
    field_users = session.query(User.id, User.username, User.create_time).all()
    print("指定字段数据：", field_users)

8.2.2 条件查询

from sqlalchemy_config import SessionFactory
from base_model import User
from sqlalchemy import and_, or_

with SessionFactory() as session:
    # 等于
    res1 = session.query(User).filter(User.username == "zhangsan").first()
    # 不等于
    res2 = session.query(User).filter(User.username != "zhangsan").all()
    # 模糊查询
    res3 = session.query(User).filter(User.username.like("%zhang%")).all()
    # IN范围查询
    res4 = session.query(User).filter(User.id.in_([1,2,3])).all()
    # 大小比较
    res5 = session.query(User).filter(User.id > 2).all()
    # 多条件AND
    res6 = session.query(User).filter(and_(User.id > 1, User.username.like("z%"))).all()
    # 多条件OR
    res7 = session.query(User).filter(or_(User.id == 1, User.username == "lisi")).all()
    # 简洁等值查询filter_by
    res8 = session.query(User).filter_by(username="zhangsan", password="123456").first()

8.2.3 排序、分页、统计

from sqlalchemy_config import SessionFactory
from base_model import User

with SessionFactory() as session:
    # 升序排序
    asc_users = session.query(User).order_by(User.id).all()
    # 降序排序
    desc_users = session.query(User).order_by(User.id.desc()).all()

    # 分页查询：第2页，每页2条
    page_num = 2
    page_size = 2
    page_users = session.query(User).order_by(User.id).limit(page_size).offset((page_num-1)*page_size).all()

    # 统计数量
    count = session.query(User).filter(User.username.like("z%")).count()
    print("z开头用户数量：", count)

8.2.4 分组聚合查询

from sqlalchemy_config import SessionFactory
from sqlalchemy import func, select
from base_model import User

with SessionFactory() as session:
    # 分组统计
    stmt = select(
        User.username,
        func.count(User.id).label("user_count")
    ).group_by(User.username)
    res = session.execute(stmt).mappings().all()
    print("分组统计结果：", res)

    # 分组筛选 having
    stmt2 = select(
        User.username,
        func.count(User.id).label("count")
    ).group_by(User.username).having(func.count(User.id) > 0)
    res2 = session.execute(stmt2).mappings().all()
    print("分组筛选结果：", res2)

8.3 更新数据 Update

from sqlalchemy_config import SessionFactory
from base_model import User
from sqlalchemy import update

# 1. 单条更新
with SessionFactory() as session:
    user = session.query(User).filter(User.username == "zhangsan").first()
    if user:
        user.password = "update123456"
        session.commit()
        print("✅ 单条更新成功")

# 2. 批量更新
with SessionFactory() as session:
    session.query(User).filter(User.username.like("w%")).update({"password": "batch123"})
    session.commit()
    print("✅ 批量更新成功")

# 3. 2.0新版更新语法
with SessionFactory() as session:
    stmt = update(User).where(User.id > 3).values(password="newpass123")
    session.execute(stmt)
    session.commit()
    print("✅ 新版语法更新成功")

8.4 删除数据 Delete

from sqlalchemy_config import SessionFactory
from base_model import User
from sqlalchemy import delete

# 1. 单条删除
with SessionFactory() as session:
    user = session.query(User).filter(User.username == "wangwu").first()
    if user:
        session.delete(user)
        session.commit()
        print("✅ 单条删除成功")

# 2. 批量删除
with SessionFactory() as session:
    session.query(User).filter(User.id > 10).delete()
    session.commit()
    print("✅ 批量删除成功")

# 3. 2.0新版删除语法
with SessionFactory() as session:
    stmt = delete(User).where(User.username.like("c%"))
    session.execute(stmt)
    session.commit()
    print("✅ 新版语法删除成功")

九、新手避坑总结

• 所有增删改操作必须执行 session.commit()，否则数据不入库

• 调试阶段开启 echo=True，可查看底层原生SQL，快速排错

• 优先使用 with 上下文管理会话，自动释放连接，杜绝连接池溢出

• 批量操作优先用 bulk_insert_mappings，性能远高于循环add

• 数据库需手动创建，SQLAlchemy只负责表结构与数据操作

十、本章小结

本文完整覆盖 SQLAlchemy 入门核心知识点：

• 理解ORM思想、五大核心组件作用

• 掌握环境搭建、数据库连接、模型定义、表创建

• 精通全套CRUD操作：增删改查、条件、排序、分页、分组聚合

• 适配SQLAlchemy2.0新版语法，贴合企业开发规范

后续可进阶：事务管理、多表关联查询、数据库迁移、FastAPI项目集成！

码字不易，点赞收藏，持续更新Python后端实战教程！