SQLAlchemy 学习笔记

简介

突然发现最近好多Vibe Coding生成的技术方案都会用SQLAlchemy，所以编写了一个简单的code顺便拆解学习一下，特记录在此，方便后续知识回顾。
SQLAlchemy 是 Python 中最流行的 ORM（对象关系映射）库，它提供了一种将 Python 对象与数据库表结构映射的方式，使开发者能够使用面向对象的方式操作数据库，而不需要直接编写 SQL 语句。
总体感觉跟微软的LINQ to SQL有点像，但其有LINQ的特殊查询语言，而在SQLAlchemy下则都是通过类和方法来操作。

核心组件

以下是使用SQLAlchemy需要用到的核心组件：

• Engine：数据库引擎，负责与数据库建立连接
• Base：所有模型类的基类
• Session：会话，用于执行数据库操作
• Model：数据模型，对应数据库中的表
• Query：查询对象，用于构建和执行数据库查询

环境搭建

安装 SQLAlchemy

pip3 install sqlalchemy

导入必要的模块

from sqlalchemy import create_engine, Column, Integer, String, Float
from sqlalchemy.orm import declarative_base, sessionmaker

数据库连接

创建数据库引擎

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

• sqlite:///products.db：SQLite 数据库连接字符串，指定数据库文件路径
• echo=False：是否打印 SQL 语句，开发时可以设置为 True 以便调试

支持的数据库类型

不同数据库的连接字符串：

• SQLite：sqlite:///database.db
• MySQL：mysql://username:password@host:port/database
• PostgreSQL：postgresql://username:password@host:port/database
• Oracle：oracle://username:password@host:port/database

数据模型定义

创建基类

# 创建基类
Base = declarative_base()

定义模型类

# 定义Product模型
class Product(Base):
    __tablename__ = 'products'  # 表名
    
    id = Column(Integer, primary_key=True, autoincrement=True)  # 主键，自增
    name = Column(String(100), nullable=False)  # 产品名称，非空
    price = Column(Float, nullable=False)  # 产品价格，非空
    category = Column(String(50), nullable=False)  # 产品类别，非空

可以看到，通过SQL检表的方法在SQLAlchemy都有对应的实现。

字段类型

跟常用数据库里的数据类型都有相应的类型对应。

• Integer：整数
• String：字符串，可指定长度
• Float：浮点数
• Boolean：布尔值
• DateTime：日期时间
• Date：日期
• Time：时间
• Text：长文本

字段参数

• primary_key：是否为主键
• autoincrement：是否自增
• nullable：是否允许为空
• default：默认值
• unique：是否唯一
• index：是否创建索引

数据表创建

# 创建数据表
Base.metadata.create_all(engine)

• 这行代码会自动创建所有继承自 Base 的模型对应的表结构
• 如果表已经存在，则不会重复创建

会话管理

创建会话工厂

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

创建会话实例

# 在类中创建会话实例
def __init__(self):
    self.session = Session()

6.3 关闭会话

def close(self):
    """关闭会话"""
    self.session.close()

CRUD 操作

创建（Create）

def create(self, name, price, category):
    """创建新产品"""
    product = Product(name=name, price=price, category=category)
    self.session.add(product)  # 添加到会话
    self.session.commit()  # 提交事务
    return product

读取（Read）

读取所有数据

def read_all(self):
    """读取所有产品"""
    return self.session.query(Product).all()

根据 ID 读取

def read_by_id(self, product_id):
    """根据ID读取产品"""
    return self.session.query(Product).filter_by(id=product_id).first()

更新（Update）

def update(self, product_id, name=None, price=None, category=None):
    """更新产品信息"""
    product = self.read_by_id(product_id)
    if product:
        if name:
            product.name = name
        if price:
            product.price = price
        if category:
            product.category = category
        self.session.commit()  # 提交事务
        return product
    return None

删除（Delete）

根据 ID 删除

def delete(self, product_id):
    """删除产品"""
    product = self.read_by_id(product_id)
    if product:
        self.session.delete(product)  # 从会话中删除
        self.session.commit()  # 提交事务
        return True
    return False

根据名称删除

def delete_by_name(self, product_name):
    """根据产品名字删除产品"""
    product = self.session.query(Product).filter_by(name=product_name).first()
    if product:
        self.session.delete(product)
        self.session.commit()
        return True
    return False

查询方法

基本查询

• query(Model)：创建查询对象
• all()：返回所有结果
• first()：返回第一个结果
• filter_by(**kwargs)：根据条件过滤
• filter(condition)：使用更复杂的条件过滤
• order_by(column)：排序
• limit(n)：限制返回数量
• offset(n)：偏移量

示例

# 查询所有产品
products = session.query(Product).all()

# 根据条件查询
product = session.query(Product).filter_by(name="iPhone 15").first()

# 复杂条件查询
products = session.query(Product).filter(Product.price > 5000).all()

# 排序
products = session.query(Product).order_by(Product.price.desc()).all()

# 限制数量
products = session.query(Product).limit(10).all()

事务管理

提交和回滚事务

self.session.commit()

self.session.rollback()

上下文管理器

with Session() as session:
    try:
        # 执行操作
        session.commit()
    except Exception:
        session.rollback()
        raise

完整示例

以下是一个完整的 SQLAlchemy 使用示例，包含了模型定义、CRUD 操作和测试代码：

from sqlalchemy import create_engine, Column, Integer, String, Float
from sqlalchemy.orm import declarative_base, sessionmaker

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

# 创建基类
Base = declarative_base()

# 定义Product模型
class Product(Base):
    __tablename__ = 'products'
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(100), nullable=False)
    price = Column(Float, nullable=False)
    category = Column(String(50), nullable=False)

# 创建数据表
Base.metadata.create_all(engine)

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

# CRUD操作类
class ProductCRUD:
    def __init__(self):
        self.session = Session()
    
    def create(self, name, price, category):
        """创建新产品"""
        product = Product(name=name, price=price, category=category)
        self.session.add(product)
        self.session.commit()
        return product
    
    def read_all(self):
        """读取所有产品"""
        return self.session.query(Product).all()
    
    def read_by_id(self, product_id):
        """根据ID读取产品"""
        return self.session.query(Product).filter_by(id=product_id).first()
    
    def update(self, product_id, name=None, price=None, category=None):
        """更新产品信息"""
        product = self.read_by_id(product_id)
        if product:
            if name:
                product.name = name
            if price:
                product.price = price
            if category:
                product.category = category
            self.session.commit()
            return product
        return None
    
    def delete(self, product_id):
        """删除产品"""
        product = self.read_by_id(product_id)
        if product:
            self.session.delete(product)
            self.session.commit()
            return True
        return False
    
    def delete_by_name(self, product_name):
        """根据产品名字删除产品"""
        product = self.session.query(Product).filter_by(name=product_name).first()
        if product:
            self.session.delete(product)
            self.session.commit()
            return True
        return False
    
    def close(self):
        """关闭会话"""
        self.session.close()

# 测试代码
if __name__ == "__main__":
    crud = ProductCRUD()
    
    # 创建测试数据
    print("创建测试数据...")    
    crud.create("iPhone 15", 7999.99, "电子产品")
    crud.create("MacBook Pro", 12999.99, "电子产品")
    crud.create("AirPods Pro", 1999.99, "电子产品")
    
    # 读取所有产品
    print("\n读取所有产品:")
    products = crud.read_all()
    for product in products:
        print(f"ID: {product.id}, 名称: {product.name}, 价格: {product.price}, 类别: {product.category}")
    
    # 根据ID读取产品
    print("\n根据ID读取产品:")
    product = crud.read_by_id(1)
    if product:
        print(f"ID: {product.id}, 名称: {product.name}, 价格: {product.price}, 类别: {product.category}")
    
    # 更新产品
    print("\n更新产品:")
    updated_product = crud.update(1, price=8999.99)
    if updated_product:
        print(f"更新后: ID: {updated_product.id}, 名称: {updated_product.name}, 价格: {updated_product.price}, 类别: {updated_product.category}")
    
    # 删除产品
    print("\n删除产品:")
    result = crud.delete(3)
    print(f"根据ID删除结果: {'成功' if result else '失败'}")
    
    # 根据产品名字删除
    print("\n根据产品名字删除:")
    result = crud.delete_by_name("AirPods Pro")
    print(f"根据名字删除结果: {'成功' if result else '失败'}")
    
    # 再次读取所有产品
    print("\n删除后读取所有产品:")
    products = crud.read_all()
    for product in products:
        print(f"ID: {product.id}, 名称: {product.name}, 价格: {product.price}, 类别: {product.category}")
    
    # 关闭会话
    crud.close()

总结

SQLAlchemy 是一个功能强大的 ORM 库，它提供了一种优雅的方式来操作数据库。通过本文可以快速的回顾：

1. 理解 SQLAlchemy 的核心概念和组件
2. 定义数据模型并创建数据表
3. 执行基本的 CRUD 操作
4. 使用各种查询方法
5. 管理事务
6. 遵循最佳实践

SQLAlchemy 不仅简化了数据库操作，还提高了代码的可维护性和可扩展性，是 Python 开发中处理数据库的首选工具之一。

相比LINQ的方式，微软在IDE里可以通过拖拽的方式来完成ORM的映射，很是方便。SQLAlchemy也有类似的第三方库可以实现，但目前有一个更好的方法就是通过AI来实现，效率会高很多。