在Python中使用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

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

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

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

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

# 假设模型有字段 '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)

# 假设模型有字段 '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:
        """根据主键删除一个对象

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

# 示例硬删除调用
await controller.delete_byid(1, session)

# 示例软删除调用
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="创建人")
    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}")

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:
        """批量创建对象"""
        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