Python中使用sqlalchemy操作數(shù)據(jù)庫的問題總結(jié)

更新時間：2024年08月02日 09:39:21 作者：伍華聰

在探索使用?FastAPI,?SQLAlchemy,?Pydantic,Redis,?JWT?構(gòu)建的項目的時候,其中數(shù)據(jù)庫訪問采用SQLAlchemy,并采用異步方式,這篇文章主要介紹了在Python中使用sqlalchemy來操作數(shù)據(jù)庫的幾個小總結(jié),需要的朋友可以參考下

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

1、SQLAlchemy事務(wù)處理

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

async def update_range(self, obj_in_list: List[DtoType], db: AsyncSession) -> bool:
    """批量更新對象"""
    try:
        async with db.begin():  # 使用事務(wù)塊確保批量操作的一致性
            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:
                    # 獲取更新數(shù)據(jù)
                    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)
        # 異常處理時，事務(wù)會自動回滾
        return False

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

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

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

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

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

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()  # 手動提交事務(wù)
        return True
    except SQLAlchemyError as e:
        print(e)
        await db.rollback()  # 確保在出錯時回滾事務(wù)
        return False

在這個手動提交事務(wù)的例子中：

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

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

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

async def create_update(
    self, obj_in: DtoType, id: PrimaryKeyType, db: AsyncSession
) -> bool:
    """創(chuàng)建或更新對象"""
    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:
            # 創(chuàng)建對象
            return await self.create(obj_in, db)
    except SQLAlchemyError as e:
        print(e)
        # 確保在出錯時回滾事務(wù)
        await db.rollback()
        return False

在這個代碼中：

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

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

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

代碼說明：

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

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

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

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

1. where(...)

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

2. filter(...)

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

主要差異

上下文：where 是 select 對象的一部分，通常用于構(gòu)建 SQL 查詢（SQLAlchemy Core）。而 filter 是 Query 對象的一部分，通常用于 ORM 查詢（SQLAlchemy ORM）。然而，在 SQLAlchemy 2.0+ 中，select 和 filter 的使用變得更加一致。
語義：在使用 SQLAlchemy Core 時，where 更加明確地表示你正在添加 SQL 語句中的 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 的使用相對較少。

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

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

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

model_dump(exclude_unset=True)

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

功能：排除所有未顯式設(shè)置（即使用默認(rèn)值）的字段。
使用場景：適用于需要忽略那些未被用戶設(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 是另一個選項，表示在轉(zhuǎn)換模型實例為字典時，排除那些使用了默認(rèn)值的字段。

功能：排除所有字段的值等于其默認(rèn)值的字段。
使用場景：適用于需要排除那些顯式設(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è)置的字段（即字段值為默認(rèn)值或未賦值）。
- skip_defaults=True 排除那些字段值等于其默認(rèn)值的字段。
適用場景：
- 使用 exclude_unset=True 時，目的是排除那些在實例化過程中未被顯式賦值的字段，這通常用于避免包含那些尚未配置的字段。
- 使用 skip_defaults=True 時，目的是去掉那些顯式設(shè)置為默認(rèn)值的字段，以避免輸出不必要的信息。

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

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

async def delete_byid(self, id: PrimaryKeyType, db: AsyncSession, **kwargs) -> bool:
        """根據(jù)主鍵刪除一個對象
        :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

實例代碼如下所示。

# 示例模型
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())

注意事項

模型定義：確保你的模型中包含 is_deleted 字段，并且字段名正確。
傳遞參數(shù)：在調(diào)用 delete_byid 方法時，正確傳遞 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)建時間")
    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}")

注意事項

模型定義：你的 Customer 模型定義看起來是正確的，確保所有字段和注釋都符合你的要求。
硬刪除和軟刪除：
- 硬刪除：直接從數(shù)據(jù)庫中刪除記錄。
- 軟刪除：通過更新 is_deleted 字段來標(biāo)記記錄為已刪除，而不是實際刪除記錄。
正確傳遞參數(shù)：
- 硬刪除時，不需要傳遞額外參數(shù)。
- 軟刪除時，傳遞 is_deleted=1 作為參數(shù)。

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

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

在 Python 中，Iterable 和 List 是兩個不同的概念，它們有各自的特點和用途：

Iterable

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

特點

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

特點

具體實現(xiàn)：List 是一個具體的類型，表示一個動態(tài)數(shù)組，可以存儲多個對象。
有序：列表保持元素的插入順序。
可變：可以對列表中的元素進(jìn)行修改（如添加、刪除、更新）。
示例：[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]

總結(jié)一下：

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

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

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

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

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

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

您可能感興趣的文章:

欧美bbbwbbbw肥妇,免费乱码人妻系列日韩,一级黄片

Python中使用sqlalchemy操作數(shù)據(jù)庫的問題總結(jié)

目錄

1、SQLAlchemy事務(wù)處理

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

1. where(...)

2. filter(...)

總結(jié)

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

model_dump(exclude_unset=True)

model_dump(skip_defaults=True)

主要區(qū)別

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

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

Iterable

List

相關(guān)文章

最新評論

大家感興趣的內(nèi)容

最近更新的內(nèi)容

常用在線小工具

欧美bbbwbbbw肥妇,免费乱码人妻系列日韩,一级黄片

Python中使用sqlalchemy操作數(shù)據(jù)庫的問題總結(jié)

目錄

1、SQLAlchemy事務(wù)處理

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

1. where(...)

2. filter(...)

總結(jié)

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

model_dump(exclude_unset=True)

model_dump(skip_defaults=True)

主要區(qū)別

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

5、Python處理接口的時候，Iterable 和List有什么差異

Iterable

List

相關(guān)文章

最新評論

大家感興趣的內(nèi)容

最近更新的內(nèi)容

常用在線小工具

1、SQLAlchemy事務(wù)處理

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

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

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

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