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

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

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

async def save_import(self, data: List[DtoType], db: AsyncSession) -> bool:
    """批量導入對象"""
    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

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

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

# 假設(shè)模型有字段 '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)

# 假設(shè)模型有字段 '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)

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

# 示例硬刪除調(diào)用
await controller.delete_byid(1, session)
# 示例軟刪除調(diào)用
await controller.delete_byid(2, session, is_deleted=True)

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}")

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

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]

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