SQLAlchemy 数据库操作完整指南：从模型定义到CRUD实战

SQLAlchemy 是 Python 中最著名的 ORM（对象关系映射）框架之一，它允许开发者使用 Python 类来定义数据库表，并通过简单的 Python 代码执行各种数据库操作。本文将带你全面了解 SQLAlchemy 的基本使用方法。

1. 环境准备与引擎创建

首先需要安装 SQLAlchemy，然后创建数据库引擎：

from sqlalchemy import create_engine

# 创建 SQLite 数据库引擎
engine = create_engine('sqlite:///user.db')

这里我们使用 SQLite 作为示例数据库，实际项目中可以根据需要替换为 MySQL、PostgreSQL 等。

2. 定义数据模型

使用声明式基类来定义数据模型：

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

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(20))
    nikename = Column(String(32))
    password = Column(String(32))
    email = Column(String(50))

代码说明：

__tablename__ 指定映射的数据库表名
Column 定义表字段
primary_key=True 设置主键
autoincrement=True 设置自增

3. 创建数据表

1 2	# 执行业务逻辑表的创建 Base.metadata.create_all(engine, checkfirst=True)

checkfirst=True 参数确保不会重复创建已存在的表。

4. 数据库会话管理

from sqlalchemy.orm import sessionmaker

# 创建会话类
Session = sessionmaker(bind=engine)
# 实例化会话对象
session = Session()

会话（Session）是 SQLAlchemy 与数据库交互的主要接口。

5. 增删改查（CRUD）操作详解

5.1 新增记录

# 创建用户实例
_user = User(name='xiaozhong', nikename='Zyx', 
             password='123456', email='zyx@qq.com')

# 添加到会话并提交
session.add(_user)
session.commit()

5.2 查询记录

基本查询：

from typing import List

# 查询所有符合条件的记录
result: List[User] = session.query(User).filter_by(name='xiaozhong').all()

for item in result:
    print(item.name, item.password, item.email)

指定字段查询：

# 只查询特定字段
result1: List[User] = session.query(User.name, User.email, User.password)\
    .filter_by(name='xiaozhong').all()

for item in result1:
    print(item.name, item.password, item.email)

模糊查询：

# 使用 like 进行模糊匹配
result2: List[User] = session.query(User)\
    .filter(User.name.like("xiao%")).all()

for item in result2:
    print(item.name, item.password, item.email)

正则表达式查询：

1
2
3

# 使用正则表达式匹配
result3: List[User] = session.query(User)\
    .filter(User.name.op("regexp")("^xiao")).all()

5.3 更新记录

方法一：对象属性更新

# 查询并更新对象
update_user: User = session.query(User)\
    .filter_by(name='xiaozhong').first()

update_user.email = 'zyx123@qq.com'
session.commit()

方法二：批量更新

# 直接批量更新
session.query(User).filter_by(name='xiaozhong')\
    .update({User.email: 'zyx123456@qq.com'})
session.commit()

5.4 删除记录

方法一：对象删除

# 查询并删除对象
del_user: User = session.query(User)\
    .filter_by(name='xiaozhong').first()

if del_user:
    session.delete(del_user)
    session.commit()

方法二：条件删除

# 直接条件删除
session.query(User).filter_by(name='xiaozhong').delete()
session.commit()

# 或者使用另一种语法
session.query(User).filter(User.name=='xiaozhong')\
    .delete(synchronize_session=False)
session.commit()