在Python中使用sqlalchemy来操作数据库的几个小总结

简介: 在探索使用 FastAPI, SQLAlchemy, Pydantic,Redis, JWT 构建的项目的时候,其中数据库访问采用SQLAlchemy,并采用异步方式。数据库操作和控制器操作,采用基类继承的方式减少重复代码,提高代码复用性。在这个过程中设计接口和测试的时候,对一些问题进行跟踪解决,并记录供参考。

在探索使用 FastAPI, SQLAlchemy, Pydantic,Redis, JWT 构建的项目的时候,其中数据库访问采用SQLAlchemy,并采用异步方式。数据库操作和控制器操作,采用基类继承的方式减少重复代码,提高代码复用性。在这个过程中设计接口和测试的时候,对一些问题进行跟踪解决,并记录供参考。

1、SQLAlchemy事务处理

在异步环境中,批量更新操作需要使用异步方法来执行查询和提交事务。

async def update_range(self, obj_in_list: List[DtoType], db: AsyncSession) -> bool:
    """批量更新对象"""
    try:
        async with db.begin():  # 使用事务块确保批量操作的一致性
            for obj_in in obj_in_list:
                # 查询对象
                query = select(self.model).filter(self.model.id == obj_in.id)
                result = await db.execute(query)
                db_obj = result.scalars().first()
                
                if db_obj:
                    # 获取更新数据
                    update_data = obj_in.model_dump(skip_defaults=True)
                    
                    # 更新对象字段
                    for field, value in update_data.items():
                        setattr(db_obj, field, value)
        
        return True
    except SQLAlchemyError as e:
        print(e)
        # 异常处理时,事务会自动回滚
        return False

在这个改进后的代码中:

  1. 事务块:使用 async with db.begin() 创建事务块,以确保批量操作的一致性。事务块会在操作完成后自动提交,并在出现异常时回滚。
  2. 查询对象:使用 select(self.model).filter(self.model.id == obj_in.id) 进行异步查询,并使用 await db.execute(query) 执行查询。
  3. 更新对象字段:用 setattr 更新对象的字段。
  4. 异常处理:捕获 SQLAlchemyError 异常,并在异常发生时回滚事务。事务块会自动处理回滚,因此不需要手动回滚。


这种方式确保了在异步环境中批量更新操作的正确性和一致性。


在使用 async with db.begin() 进行事务管理时,事务会自动提交。如果在事务块内执行的所有操作都成功,事务会在退出时自动提交;如果出现异常,事务会自动回滚。


因此,手动调用 await db.commit() 是不必要的,因为事务块会处理这些操作。如果你不使用事务块,并希望手动控制事务的提交,可以如下修改:

async def update_range(self, obj_in_list: List[DtoType], db: AsyncSession) -> bool:
    """批量更新对象"""
    try:
        for obj_in in obj_in_list:
            query = select(self.model).filter(self.model.id == obj_in.id)
            result = await db.execute(query)
            db_obj = result.scalars().first()
            
            if db_obj:
                update_data = obj_in.model_dump(skip_defaults=True)
                
                for field, value in update_data.items():
                    setattr(db_obj, field, value)
        
        await db.commit()  # 手动提交事务
        return True
    except SQLAlchemyError as e:
        print(e)
        await db.rollback()  # 确保在出错时回滚事务
        return False

在这个手动提交事务的例子中:

  • 在更新对象的操作完成后,使用 await db.commit() 来提交事务。
  • 如果发生异常,使用 await db.rollback() 来回滚事务。


根据需求选择合适的方法进行事务管理。事务块方式通常是更安全和简洁的选择。


在异步环境中,create_update 方法需要对数据库进行异步查询、更新或创建操作。

async def create_update(
    self, obj_in: DtoType, id: PrimaryKeyType, db: AsyncSession
) -> bool:
    """创建或更新对象"""
    try:
        # 查询对象
        query = select(self.model).filter(self.model.id == id)
        result = await db.execute(query)
        db_obj = result.scalars().first()
        
        if db_obj:
            # 更新对象
            return await self.update(obj_in, db)
        else:
            # 创建对象
            return await self.create(obj_in, db)
    except SQLAlchemyError as e:
        print(e)
        # 确保在出错时回滚事务
        await db.rollback()
        return False

在这个代码中:

  1. 异步查询:使用 select(self.model).filter(self.model.id == id) 来构建查询,并用 await db.execute(query) 执行查询。
  2. 获取对象:使用 result.scalars().first() 来获取查询结果中的第一个对象。
  3. 调用更新或创建方法:根据查询结果的有无,分别调用 self.updateself.create 方法。确保这两个方法都是异步的,并在调用时使用 await
  4. 异常处理:捕获 SQLAlchemyError 异常,并在发生异常时使用 await db.rollback() 来回滚事务。


在异步环境中,批量插入对象通常需要使用异步方法来执行数据库操作。由于 bulk_insert_mappings 在 SQLAlchemy 的异步版本中可能不直接支持,你可以使用 add_all 方法来批量添加对象。

async def save_import(self, data: List[DtoType], db: AsyncSession) -> bool:
    """批量导入对象"""
    try:
        # 将 DTO 转换为模型实例
        db_objs = [self.model(**obj_in.model_dump()) for obj_in in data]
        
        # 批量添加对象
        db.add_all(db_objs)
        
        # 提交事务
        await db.commit()
        
        return True
    except SQLAlchemyError as e:
        print(e)
        await db.rollback()  # 确保在出错时回滚事务
        return False

代码说明:

  1. 转换 DTO 为模型实例:使用 [self.model(**obj_in.model_dump()) for obj_in in data]data 列表中的 DTO 转换为模型实例列表。
  2. 批量添加对象:使用 db.add_all(db_objs) 批量添加对象到数据库会话。
  3. 提交事务:使用 await db.commit() 异步提交事务。
  4. 异常处理:捕获 SQLAlchemyError 异常,使用 await db.rollback() 回滚事务以确保在出错时数据库状态的一致性。


这种方式确保了在异步环境中正确地进行批量导入操作,并处理可能出现的异常。

2、在 SQLAlchemy 中select(...).where(...)select(...).filter(...)的差异

在 SQLAlchemy 中,select(...).where(...)select(...).filter(...) 都用于构造查询条件,但它们有一些细微的差别和适用场景。

1. where(...)

  • 定义where 是 SQLAlchemy 中 select 对象的方法,用于添加查询的条件。
  • 用法query = select(self.model).where(self.model.id == id)
  • 描述where 方法用于指定 SQL WHERE 子句的条件。在大多数情况下,它的行为和 filter 是等效的。


2. filter(...)

  • 定义filter 是 SQLAlchemy 中 Query 对象的方法,用于添加查询的条件。
  • 用法query = select(self.model).filter(self.model.id == id)
  • 描述filter 方法也用于指定 SQL WHERE 子句的条件。它通常用于更复杂的查询构建中,尤其是在 ORM 查询中。


主要差异

  • 上下文whereselect 对象的一部分,通常用于构建 SQL 查询(SQLAlchemy Core)。而 filterQuery 对象的一部分,通常用于 ORM 查询(SQLAlchemy ORM)。然而,在 SQLAlchemy 2.0+ 中,selectfilter 的使用变得更加一致。
  • 语义:在使用 SQLAlchemy Core 时,where 更加明确地表示你正在添加 SQL 语句中的 WHERE 子句。在 ORM 查询中,filter 也做了类似的事情,但它提供了更多 ORM 相关的功能。


使用 where 的示例(SQLAlchemy Core):

from sqlalchemy.future import select
from sqlalchemy.ext.asyncio import AsyncSession

async def get(self, id: int, db: AsyncSession) -> Optional[ModelType]:
    query = select(self.model).where(self.model.id == id)
    result = await db.execute(query)
    return result.scalars().first()

使用 filter 的示例(SQLAlchemy ORM):

from sqlalchemy.orm import sessionmaker

async def get(self, id: int, db: AsyncSession) -> Optional[ModelType]:
    query = select(self.model).filter(self.model.id == id)
    result = await db.execute(query)
    return result.scalars().first()

总结

  • 在 SQLAlchemy Core 中where 是构建查询条件的标准方法。
  • 在 SQLAlchemy ORM 中filter 用于构建查询条件,但在 Core 中,filter 的使用相对较少。


在 SQLAlchemy 2.0 及更高版本中,selectwherefilter 的用法变得越来越一致,你可以根据自己的习惯和需求选择其中一种。在实际开发中,选择哪一种方法通常取决于你的代码上下文和个人偏好。

3、model_dump(exclude_unset=True) 和model_dump(skip_defaults=True)有什么差异

model_dump(exclude_unset=True)model_dump(skip_defaults=True) 是用于处理模型实例的序列化方法,它们的用途和行为略有不同。这两个方法通常用于将模型实例转换为字典,以便进行进一步的处理或传输。

model_dump(exclude_unset=True)

exclude_unset=True 是一个选项,通常用于序列化方法中,表示在转换模型实例为字典时,排除那些未设置的字段。

  • 功能:排除所有未显式设置(即使用默认值)的字段。
  • 使用场景:适用于需要忽略那些未被用户设置的字段,以避免在输出中包含默认值。
# 假设模型有字段 'name' 和 'age',且 'age' 使用了默认值
model_instance = MyModel(name='Alice', age=25)
# 如果 age 的默认值是 0, exclude_unset=True 将只包含 'name'
serialized_data = model_instance.model_dump(exclude_unset=True)

model_dump(skip_defaults=True)

skip_defaults=True 是另一个选项,表示在转换模型实例为字典时,排除那些使用了默认值的字段。

  • 功能:排除所有字段的值等于其默认值的字段。
  • 使用场景:适用于需要排除那些显式设置为默认值的字段,以减少输出的冗余信息。
# 假设模型有字段 'name' 和 'age',且 'age' 使用了默认值
model_instance = MyModel(name='Alice', age=25)
# 如果 age 的默认值是 0, skip_defaults=True 将只包含 'name'
serialized_data = model_instance.model_dump(skip_defaults=True)

主要区别

  • 排除条件
  • exclude_unset=True 排除那些在模型实例中未显式设置的字段(即字段值为默认值或未赋值)。
  • skip_defaults=True 排除那些字段值等于其默认值的字段。
  • 适用场景
  • 使用 exclude_unset=True 时,目的是排除那些在实例化过程中未被显式赋值的字段,这通常用于避免包含那些尚未配置的字段。
  • 使用 skip_defaults=True 时,目的是去掉那些显式设置为默认值的字段,以避免输出不必要的信息。

4、使用**kwargs 参数,在接口中实现数据软删除的处理

例如我们在删除接口中,如果传递了 kwargs 参数,则进行软删除(更新记录),否则进行硬删除(删除记录)。

    async def delete_byid(self, id: PrimaryKeyType, db: AsyncSession, **kwargs) -> bool:
        """根据主键删除一个对象

        :param kwargs: for soft deletion only
        """
        if not kwargs:
            result = await db.execute(sa_delete(self.model).where(self.model.id == id))
        else:
            result = await db.execute(
                sa_update(self.model).where(self.model.id == id) .values(** kwargs) 
            )

        await db.commit()
        return result.rowcount > 0

实例代码如下所示:

# 示例模型
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Boolean

Base = declarative_base()

class Customer(Base):
    __tablename__ = 'customer'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    is_deleted = Column(Boolean, default=False)

# 示例使用
async def main():
    async with AsyncSession(engine) as session:
        controller = BaseController(Customer)
        
        # 硬删除
        result = await controller.delete_byid(1, session)
        print(f"Hard delete successful: {result}")
        
        # 软删除
        result = await controller.delete_byid(2, session, is_deleted=True)
        print(f"Soft delete successful: {result}")

# 确保运行主程序
import asyncio
if __name__ == "__main__":
    asyncio.run(main())

注意事项

  1. 模型定义:确保你的模型中包含 is_deleted 字段,并且字段名正确。
  2. 传递参数:在调用 delete_byid 方法时,正确传递 kwargs 参数。例如,如果你要进行软删除,可以传递 is_deleted=True
  3. 调试输出:你可以添加一些调试输出(如 print(kwargs)),以确保正确传递了参数。
# 示例硬删除调用
await controller.delete_byid(1, session)

# 示例软删除调用
await controller.delete_byid(2, session, is_deleted=True

如果我们的is_deleted 字段是Int类型的,如下所示,那么处理有所不同

class Customer(Base):
    __tablename__ = "t_customer"

    id = Column(String, primary_key=True, comment="主键")
    name = Column(String, comment="姓名")
    age = Column(Integer, comment="年龄")
    creator = Column(String, comment="创建人")
    createtime = Column(DateTime, comment="创建时间")
    is_deleted = Column(Integer, comment="是否删除")

操作代码

        # 硬删除
        result = await controller.delete_byid("1", session)
        print(f"Hard delete successful: {result}")
        
        # 软删除
        result = await controller.delete_byid("2", session, is_deleted=1)
        print(f"Soft delete successful: {result}")

注意事项

  1. 模型定义:你的 Customer 模型定义看起来是正确的,确保所有字段和注释都符合你的要求。
  2. 硬删除和软删除
  • 硬删除:直接从数据库中删除记录。
  • 软删除:通过更新 is_deleted 字段来标记记录为已删除,而不是实际删除记录。
  1. 正确传递参数
  • 硬删除时,不需要传递额外参数。
  • 软删除时,传递 is_deleted=1 作为参数。

通过确保正确传递参数并且模型包含正确的字段,你应该能够正确执行软删除和硬删除操作。

5、Python处理接口的时候,Iterable 和List有什么差异

在 Python 中,IterableList 是两个不同的概念,它们有各自的特点和用途:

Iterable

Iterable 是一个更广泛的概念,指的是任何可以返回一个迭代器的对象。迭代器是一个实现了 __iter__() 方法的对象,能够逐个返回元素。几乎所有的容器类型(如列表、元组、字典、集合等)都是可迭代的。要检查一个对象是否是可迭代的,可以使用 collections.abc.Iterable 来进行检查。

特点

  • 通用性Iterable 是一个通用的接口,表示对象可以被迭代。
  • 惰性:一些 Iterable 可能是惰性计算的(如生成器),即它们不会立即计算所有元素,而是按需生成元素。
  • 示例:列表(List)、元组(Tuple)、字典(Dict)、集合(Set)、生成器(Generator)等都是可迭代对象。
from collections.abc import Iterable

print(isinstance([1, 2, 3], Iterable))  # True
print(isinstance((1, 2, 3), Iterable))  # True
print(isinstance({1, 2, 3}, Iterable))  # True
print(isinstance({'a': 1}, Iterable))   # True
print(isinstance((x for x in range(3)), Iterable))  # True

List

List 是 Python 中的一种具体的容器类型,表示一个有序的元素集合,可以包含重复的元素。它是最常用的可变序列类型之一,支持索引访问、切片操作以及其他多种方法来操作列表中的元素。

特点

  • 具体实现List 是一个具体的类型,表示一个动态数组,可以存储多个对象。
  • 有序:列表保持元素的插入顺序。
  • 可变:可以对列表中的元素进行修改(如添加、删除、更新)。
  • 示例[1, 2, 3] 是一个列表。
my_list = [1, 2, 3]
print(my_list)  # [1, 2, 3]
my_list.append(4)  # [1, 2, 3, 4]
my_list[0] = 10  # [10, 2, 3, 4]

总结一下:

  • Iterable:一个广泛的概念,表示可以被迭代的对象,不一定是具体的数据结构。例如,生成器是可迭代的但不是列表。
  • List:一个具体的容器类型,是一种有序的可变集合。列表是 Iterable 的一种实现,但并不是所有的 Iterable 都是列表。


Iterable 是一个抽象概念,而 List 是一个具体的实现。你可以在 List 之上使用许多操作和方法来处理数据,而 Iterable 主要关注的是是否可以进行迭代。

因此接收结合的处理,我们可以使用Iterable接口更加通用一些。

    async def create_range(
        self, obj_in_list: Iterable[DtoType], db: AsyncSession
    ) -> bool:
        """批量创建对象"""
        try:
            # 将 DTO 转换为模型实例
            db_objs = [self.model(**obj_in.model_dump()) for obj_in in obj_in_list]

            # 批量添加到数据库
            db.add_all(db_objs)
            await db.commit()
            return True
        except SQLAlchemyError as e:
            print(e)
            await db.rollback()  # 确保在出错时回滚事务
            return False

以上就是在Python中使用sqlalchemy来操作数据库的时候,对一些小问题的总结,供大家参考。

作者:伍华聪

链接:https://juejin.cn/post/7397723783442432000

相关文章
|
4天前
|
存储 缓存 Shell
你知道 Python 其实自带了小型数据库吗
你知道 Python 其实自带了小型数据库吗
12 2
你知道 Python 其实自带了小型数据库吗
|
12天前
|
数据采集 关系型数据库 MySQL
如何用Python+sqlalchemy创建数据库
如何用Python+sqlalchemy创建数据库
27 2
|
28天前
|
消息中间件 关系型数据库 数据库
Python实时监测数据库表数据变化的方法
在实现时,需要考虑到应用的实时性需求、数据库性能影响以及网络延迟等因素,选择最适合的方法。每种方法都有其适用场景和限制,理解这些方法的原理和应用,将帮助开发者在实际项目中做出最合适的技术选择。
90 17
|
16天前
|
数据库 Python
python之操作数据库删除创建
python之操作数据库删除创建
|
14天前
|
关系型数据库 MySQL 数据管理
pymysql:Python操作MySQL数据库的又一利器
pymysql:Python操作MySQL数据库的又一利器
14 0
|
14天前
|
SQL 关系型数据库 MySQL
Python小技巧——将CSV文件导入到MySQL数据库
Python小技巧——将CSV文件导入到MySQL数据库
15 0
|
17天前
|
存储 数据库 Python
python的对象数据库ZODB的使用(python3经典编程案例)
该文章介绍了如何使用Python的对象数据库ZODB来进行数据存储,包括ZODB的基本操作如创建数据库、存储和检索对象等,并提供了示例代码。
22 0
|
17天前
|
JSON NoSQL 数据库
和SQLite数据库对应的NoSQL数据库:TinyDB的详细使用(python3经典编程案例)
该文章详细介绍了TinyDB这一轻量级NoSQL数据库的使用方法,包括如何在Python3环境中安装、创建数据库、插入数据、查询、更新以及删除记录等操作,并提供了多个编程案例。
34 0
|
25天前
|
SQL 关系型数据库 MySQL
Python操作pymysql数据库的流程与技巧
在现代软件开发中,Python作为一门高效且易于学习的编程语言,广泛应用于各种场景,其中包括数据库操作。**PyMySQL** 是一个流行的Python数据库接口,用于连接和操作MySQL数据库。它提供了一种简便的方法来执行SQL语句、处理数据和管理数据库事务。以下是使用PyMySQL操作MySQL数据库的流程与技巧,旨在为开发者提供一个清晰、实用的指南。
23 0
|
3天前
|
存储 关系型数据库 MySQL
一个项目用5款数据库?MySQL、PostgreSQL、ClickHouse、MongoDB区别,适用场景
一个项目用5款数据库?MySQL、PostgreSQL、ClickHouse、MongoDB——特点、性能、扩展性、安全性、适用场景比较