在探索使用 FastAPI, SQLAlchemy, Pydantic,Redis, JWT 構(gòu)建的項(xiàng)目的時(shí)候，其中數(shù)據(jù)庫(kù)訪問(wèn)采用SQLAlchemy，并采用異步方式。數(shù)據(jù)庫(kù)操作和控制器操作，采用基類繼承的方式減少重復(fù)代碼，提高代碼復(fù)用性。在這個(gè)過(guò)程中設(shè)計(jì)接口和測(cè)試的時(shí)候，對(duì)一些問(wèn)題進(jìn)行跟蹤解決，并記錄供參考。

1、SQLAlchemy事務(wù)處理

在異步環(huán)境中，批量更新操作需要使用異步方法來(lái)執(zhí)行查詢和提交事務(wù)。

async def update_range(self, obj_in_list: List[DtoType], db: AsyncSession) -> bool:
    """批量更新對(duì)象"""
    try:
        async with db.begin():  # 使用事務(wù)塊確保批量操作的一致性
            for obj_in in obj_in_list:
                # 查詢對(duì)象
                query = select(self.model).filter(self.model.id == obj_in.id)
                result = await db.execute(query)
                db_obj = result.scalars().first()
                if db_obj:
                    # 獲取更新數(shù)據(jù)
                    update_data = obj_in.model_dump(skip_defaults=True)
                    # 更新對(duì)象字段
                    for field, value in update_data.items():
                        setattr(db_obj, field, value)
        return True
    except SQLAlchemyError as e:
        print(e)
        # 異常處理時(shí)，事務(wù)會(huì)自動(dòng)回滾
        return False

在這個(gè)改進(jìn)后的代碼中：

• 事務(wù)塊：使用 async with db.begin() 創(chuàng)建事務(wù)塊，以確保批量操作的一致性。事務(wù)塊會(huì)在操作完成后自動(dòng)提交，并在出現(xiàn)異常時(shí)回滾。
• 查詢對(duì)象：使用 select(self.model).filter(self.model.id == obj_in.id) 進(jìn)行異步查詢，并使用 await db.execute(query) 執(zhí)行查詢。
• 更新對(duì)象字段：用 setattr 更新對(duì)象的字段。
• 異常處理：捕獲 SQLAlchemyError 異常，并在異常發(fā)生時(shí)回滾事務(wù)。事務(wù)塊會(huì)自動(dòng)處理回滾，因此不需要手動(dòng)回滾。

這種方式確保了在異步環(huán)境中批量更新操作的正確性和一致性。

在使用 async with db.begin() 進(jìn)行事務(wù)管理時(shí)，事務(wù)會(huì)自動(dòng)提交。如果在事務(wù)塊內(nèi)執(zhí)行的所有操作都成功，事務(wù)會(huì)在退出時(shí)自動(dòng)提交；如果出現(xiàn)異常，事務(wù)會(huì)自動(dòng)回滾。

因此，手動(dòng)調(diào)用 await db.commit() 是不必要的，因?yàn)槭聞?wù)塊會(huì)處理這些操作。如果你不使用事務(wù)塊，并希望手動(dòng)控制事務(wù)的提交，可以如下修改：

async def update_range(self, obj_in_list: List[DtoType], db: AsyncSession) -> bool:
    """批量更新對(duì)象"""
    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()  # 手動(dòng)提交事務(wù)
        return True
    except SQLAlchemyError as e:
        print(e)
        await db.rollback()  # 確保在出錯(cuò)時(shí)回滾事務(wù)
        return False

在這個(gè)手動(dòng)提交事務(wù)的例子中：

• 在更新對(duì)象的操作完成后，使用 await db.commit() 來(lái)提交事務(wù)。
• 如果發(fā)生異常，使用 await db.rollback() 來(lái)回滾事務(wù)。

根據(jù)需求選擇合適的方法進(jìn)行事務(wù)管理。事務(wù)塊方式通常是更安全和簡(jiǎn)潔的選擇。

 在異步環(huán)境中，create_update 方法需要對(duì)數(shù)據(jù)庫(kù)進(jìn)行異步查詢、更新或創(chuàng)建操作。

async def create_update(
    self, obj_in: DtoType, id: PrimaryKeyType, db: AsyncSession
) -> bool:
    """創(chuàng)建或更新對(duì)象"""
    try:
        # 查詢對(duì)象
        query = select(self.model).filter(self.model.id == id)
        result = await db.execute(query)
        db_obj = result.scalars().first()
        if db_obj:
            # 更新對(duì)象
            return await self.update(obj_in, db)
        else:
            # 創(chuàng)建對(duì)象
            return await self.create(obj_in, db)
    except SQLAlchemyError as e:
        print(e)
        # 確保在出錯(cuò)時(shí)回滾事務(wù)
        await db.rollback()
        return False

在這個(gè)代碼中：

• 異步查詢：使用 select(self.model).filter(self.model.id == id) 來(lái)構(gòu)建查詢，并用 await db.execute(query) 執(zhí)行查詢。
• 獲取對(duì)象：使用 result.scalars().first() 來(lái)獲取查詢結(jié)果中的第一個(gè)對(duì)象。
• 調(diào)用更新或創(chuàng)建方法：根據(jù)查詢結(jié)果的有無(wú)，分別調(diào)用 self.update 或 self.create 方法。確保這兩個(gè)方法都是異步的，并在調(diào)用時(shí)使用 await。
• 異常處理：捕獲 SQLAlchemyError 異常，并在發(fā)生異常時(shí)使用 await db.rollback() 來(lái)回滾事務(wù)。

在異步環(huán)境中，批量插入對(duì)象通常需要使用異步方法來(lái)執(zhí)行數(shù)據(jù)庫(kù)操作。由于 bulk_insert_mappings 在 SQLAlchemy 的異步版本中可能不直接支持，你可以使用 add_all 方法來(lái)批量添加對(duì)象。

async def save_import(self, data: List[DtoType], db: AsyncSession) -> bool:
    """批量導(dǎo)入對(duì)象"""
    try:
        # 將 DTO 轉(zhuǎn)換為模型實(shí)例
        db_objs = [self.model(**obj_in.model_dump()) for obj_in in data]
        # 批量添加對(duì)象
        db.add_all(db_objs)
        # 提交事務(wù)
        await db.commit()
        return True
    except SQLAlchemyError as e:
        print(e)
        await db.rollback()  # 確保在出錯(cuò)時(shí)回滾事務(wù)
        return False

代碼說(shuō)明：

• 轉(zhuǎn)換 DTO 為模型實(shí)例：使用 [self.model(**obj_in.model_dump()) for obj_in in data] 將 data 列表中的 DTO 轉(zhuǎn)換為模型實(shí)例列表。
• 批量添加對(duì)象：使用 db.add_all(db_objs) 批量添加對(duì)象到數(shù)據(jù)庫(kù)會(huì)話。
• 提交事務(wù)：使用 await db.commit() 異步提交事務(wù)。
• 異常處理：捕獲 SQLAlchemyError 異常，使用 await db.rollback() 回滾事務(wù)以確保在出錯(cuò)時(shí)數(shù)據(jù)庫(kù)狀態(tài)的一致性。

這種方式確保了在異步環(huán)境中正確地進(jìn)行批量導(dǎo)入操作，并處理可能出現(xiàn)的異常。

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

在 SQLAlchemy 中，select(...).where(...) 和 select(...).filter(...) 都用于構(gòu)造查詢條件，但它們有一些細(xì)微的差別和適用場(chǎng)景。

1. where(...)

• 定義：where 是 SQLAlchemy 中 select 對(duì)象的方法，用于添加查詢的條件。
• 用法：query = select(self.model).where(self.model.id == id)
• 描述：where 方法用于指定 SQL WHERE 子句的條件。在大多數(shù)情況下，它的行為和 filter 是等效的。

2. filter(...)

• 定義：filter 是 SQLAlchemy 中 Query 對(duì)象的方法，用于添加查詢的條件。
• 用法：query = select(self.model).filter(self.model.id == id)
• 描述：filter 方法也用于指定 SQL WHERE 子句的條件。它通常用于更復(fù)雜的查詢構(gòu)建中，尤其是在 ORM 查詢中。

主要差異

• 上下文：where 是 select 對(duì)象的一部分，通常用于構(gòu)建 SQL 查詢（SQLAlchemy Core）。而 filter 是 Query 對(duì)象的一部分，通常用于 ORM 查詢（SQLAlchemy ORM）。然而，在 SQLAlchemy 2.0+ 中，select 和 filter 的使用變得更加一致。

• 語(yǔ)義：在使用 SQLAlchemy Core 時(shí)，where 更加明確地表示你正在添加 SQL 語(yǔ)句中的 WHERE 子句。在 ORM 查詢中，filter 也做了類似的事情，但它提供了更多 ORM 相關(guān)的功能。

使用 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()

總結(jié)

• 在 SQLAlchemy Core 中：where 是構(gòu)建查詢條件的標(biāo)準(zhǔn)方法。
• 在 SQLAlchemy ORM 中：filter 用于構(gòu)建查詢條件，但在 Core 中，filter 的使用相對(duì)較少。

在 SQLAlchemy 2.0 及更高版本中，select 的 where 和 filter 的用法變得越來(lái)越一致，你可以根據(jù)自己的習(xí)慣和需求選擇其中一種。在實(shí)際開(kāi)發(fā)中，選擇哪一種方法通常取決于你的代碼上下文和個(gè)人偏好。

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

model_dump(exclude_unset=True) 和 model_dump(skip_defaults=True) 是用于處理模型實(shí)例的序列化方法，它們的用途和行為略有不同。這兩個(gè)方法通常用于將模型實(shí)例轉(zhuǎn)換為字典，以便進(jìn)行進(jìn)一步的處理或傳輸。

model_dump(exclude_unset=True)

exclude_unset=True 是一個(gè)選項(xiàng)，通常用于序列化方法中，表示在轉(zhuǎn)換模型實(shí)例為字典時(shí)，排除那些未設(shè)置的字段。

• 功能：排除所有未顯式設(shè)置（即使用默認(rèn)值）的字段。
• 使用場(chǎng)景：適用于需要忽略那些未被用戶設(shè)置的字段，以避免在輸出中包含默認(rèn)值。

# 假設(shè)模型有字段 'name' 和 'age'，且 'age' 使用了默認(rèn)值
model_instance = MyModel(name='Alice', age=25)
# 如果 age 的默認(rèn)值是 0, exclude_unset=True 將只包含 'name'
serialized_data = model_instance.model_dump(exclude_unset=True)

model_dump(skip_defaults=True)

skip_defaults=True 是另一個(gè)選項(xiàng)，表示在轉(zhuǎn)換模型實(shí)例為字典時(shí)，排除那些使用了默認(rèn)值的字段。

• 功能：排除所有字段的值等于其默認(rèn)值的字段。
• 使用場(chǎng)景：適用于需要排除那些顯式設(shè)置為默認(rèn)值的字段，以減少輸出的冗余信息。

# 假設(shè)模型有字段 'name' 和 'age'，且 'age' 使用了默認(rèn)值
model_instance = MyModel(name='Alice', age=25)
# 如果 age 的默認(rèn)值是 0, skip_defaults=True 將只包含 'name'
serialized_data = model_instance.model_dump(skip_defaults=True)

主要區(qū)別

• 排除條件：

  • exclude_unset=True 排除那些在模型實(shí)例中未顯式設(shè)置的字段（即字段值為默認(rèn)值或未賦值）。
  • skip_defaults=True 排除那些字段值等于其默認(rèn)值的字段。

• 適用場(chǎng)景：

  • 使用 exclude_unset=True 時(shí)，目的是排除那些在實(shí)例化過(guò)程中未被顯式賦值的字段，這通常用于避免包含那些尚未配置的字段。
  • 使用 skip_defaults=True 時(shí)，目的是去掉那些顯式設(shè)置為默認(rèn)值的字段，以避免輸出不必要的信息。

4、使用**kwargs 參數(shù)，在接口中實(shí)現(xiàn)數(shù)據(jù)軟刪除的處理

例如我們?cè)趧h除接口中，如果傳遞了 kwargs 參數(shù)，則進(jìn)行軟刪除（更新記錄），否則進(jìn)行硬刪除（刪除記錄）。

async def delete_byid(self, id: PrimaryKeyType, db: AsyncSession, **kwargs) -> bool:
        """根據(jù)主鍵刪除一個(gè)對(duì)象
        :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)<strong>.values(**</strong><strong>kwargs)</strong>
            )
        await db.commit()
        return result.rowcount > 0

實(shí)例代碼如下所示。

# 示例模型
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}")
# 確保運(yùn)行主程序
import asyncio
if __name__ == "__main__":
    asyncio.run(main())

注意事項(xiàng)

• 模型定義：確保你的模型中包含 is_deleted 字段，并且字段名正確。

• 傳遞參數(shù)：在調(diào)用 delete_byid 方法時(shí)，正確傳遞 kwargs 參數(shù)。例如，如果你要進(jìn)行軟刪除，可以傳遞 is_deleted=True。

• 調(diào)試輸出：你可以添加一些調(diào)試輸出（如 print(kwargs)），以確保正確傳遞了參數(shù)。

# 示例硬刪除調(diào)用
await controller.delete_byid(1, session)
# 示例軟刪除調(diào)用
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="創(chuàng)建人")
    createtime = Column(DateTime, comment="創(chuàng)建時(shí)間")
    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, <strong>is_deleted=1</strong>)
        print(f"Soft delete successful: {result}")

注意事項(xiàng)

• 模型定義：你的 Customer 模型定義看起來(lái)是正確的，確保所有字段和注釋都符合你的要求。

• 硬刪除和軟刪除：

  • 硬刪除：直接從數(shù)據(jù)庫(kù)中刪除記錄。
  • 軟刪除：通過(guò)更新 is_deleted 字段來(lái)標(biāo)記記錄為已刪除，而不是實(shí)際刪除記錄。

• 正確傳遞參數(shù)：

  • 硬刪除時(shí)，不需要傳遞額外參數(shù)。
  • 軟刪除時(shí)，傳遞 is_deleted=1 作為參數(shù)。

通過(guò)確保正確傳遞參數(shù)并且模型包含正確的字段，你應(yīng)該能夠正確執(zhí)行軟刪除和硬刪除操作。

5、Python處理接口的時(shí)候，Iterable 和List有什么差異

在 Python 中，Iterable 和 List 是兩個(gè)不同的概念，它們有各自的特點(diǎn)和用途：

Iterable

Iterable 是一個(gè)更廣泛的概念，指的是任何可以返回一個(gè)迭代器的對(duì)象。迭代器是一個(gè)實(shí)現(xiàn)了 __iter__() 方法的對(duì)象，能夠逐個(gè)返回元素。幾乎所有的容器類型（如列表、元組、字典、集合等）都是可迭代的。要檢查一個(gè)對(duì)象是否是可迭代的，可以使用 collections.abc.Iterable 來(lái)進(jìn)行檢查。

特點(diǎn)

• 通用性：Iterable 是一個(gè)通用的接口，表示對(duì)象可以被迭代。
• 惰性：一些 Iterable 可能是惰性計(jì)算的（如生成器），即它們不會(huì)立即計(jì)算所有元素，而是按需生成元素。
• 示例：列表（List）、元組（Tuple）、字典（Dict）、集合（Set）、生成器（Generator）等都是可迭代對(duì)象。

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 中的一種具體的容器類型，表示一個(gè)有序的元素集合，可以包含重復(fù)的元素。它是最常用的可變序列類型之一，支持索引訪問(wèn)、切片操作以及其他多種方法來(lái)操作列表中的元素。

特點(diǎn)

• 具體實(shí)現(xiàn)：List 是一個(gè)具體的類型，表示一個(gè)動(dòng)態(tài)數(shù)組，可以存儲(chǔ)多個(gè)對(duì)象。
• 有序：列表保持元素的插入順序。
• 可變：可以對(duì)列表中的元素進(jìn)行修改（如添加、刪除、更新）。
• 示例：[1, 2, 3] 是一個(gè)列表。

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]

總結(jié)一下：

• Iterable：一個(gè)廣泛的概念，表示可以被迭代的對(duì)象，不一定是具體的數(shù)據(jù)結(jié)構(gòu)。例如，生成器是可迭代的但不是列表。
• List：一個(gè)具體的容器類型，是一種有序的可變集合。列表是 Iterable 的一種實(shí)現(xiàn)，但并不是所有的 Iterable 都是列表。

Iterable 是一個(gè)抽象概念，而 List 是一個(gè)具體的實(shí)現(xiàn)。你可以在 List 之上使用許多操作和方法來(lái)處理數(shù)據(jù)，而 Iterable 主要關(guān)注的是是否可以進(jìn)行迭代。

因此接收結(jié)合的處理，我們可以使用Iterable接口更加通用一些。

async def create_range(
        self, obj_in_list: Iterable[DtoType], db: AsyncSession
    ) -> bool:
        """批量創(chuàng)建對(duì)象"""
        try:
            # 將 DTO 轉(zhuǎn)換為模型實(shí)例
            db_objs = [self.model(**obj_in.model_dump()) for obj_in in obj_in_list]
            # 批量添加到數(shù)據(jù)庫(kù)
            db.add_all(db_objs)
            await db.commit()
            return True
        except SQLAlchemyError as e:
            print(e)
            await db.rollback()  # 確保在出錯(cuò)時(shí)回滾事務(wù)
            return False

以上就是在Python中使用sqlalchemy來(lái)操作數(shù)據(jù)庫(kù)的時(shí)候，對(duì)一些小問(wèn)題的總結(jié)，供大家參考。

到此這篇關(guān)于在Python中使用sqlalchemy來(lái)操作數(shù)據(jù)庫(kù)的幾個(gè)小總結(jié)的文章就介紹到這了,更多相關(guān)Python sqlalchemy操作數(shù)據(jù)庫(kù)內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論