【Python系列】只更新非空的字段

news2024/12/24 8:36:20

💝💝💝欢迎来到我的博客，很高兴能够在这里和您见面！希望您在这里可以感受到一份轻松愉快的氛围，不仅可以获得有趣的内容和知识，也可以畅所欲言、分享您的想法和见解。

推荐:kwan 的首页,持续学习,不断总结,共同进步,活到老学到老
导航
檀越剑指大厂系列:全面总结 java 核心技术,jvm,并发编程 redis,kafka,Spring,微服务等
常用开发工具系列:常用的开发工具,IDEA,Mac,Alfred,Git,typora 等
数据库系列:详细总结了常用数据库 mysql 技术点,以及工作中遇到的 mysql 问题等
新空间代码工作室:提供各种软件服务,承接各种毕业设计,毕业论文等
懒人运维系列:总结好用的命令,解放双手不香吗?能用一个命令完成绝不用两个操作
数据结构与算法系列:总结数据结构和算法,不同类型针对性训练,提升编程思维,剑指大厂

非常期待和您一起在这个小小的网络世界里共同探索、学习和成长。💝💝💝 ✨✨ 欢迎订阅本专栏 ✨✨

博客目录

- 一.基本介绍
- 二.代码实践
- - 1.筛选方法
  - 2.更新方法

一.基本介绍

在 Python 中，将对象转换为字典通常涉及到对象的属性和方法。如果你有一个类实例，并且想要将其属性转换为字典格式，你可以定义一个方法来做这件事。以下是一个简单的例子，展示了如何定义一个静态方法来将类实例的属性转换为字典：

from typing import Any, Dict, Optional

class AlchemySchemas:
    # 假设这个类有一些属性
    def __init__(self, id: int, name: str, value: Any):
        self.id = id
        self.name = name
        self.value = value

    @staticmethod
    def get_update_values(**fields: Optional[Any]) -> Dict[str, Any]:
        """
        生成更新字典，只有当字段非空时才包含该字段。
        """
        return {k: v for k, v in fields.items() if v is not None}

    @staticmethod
    def to_dict(obj: 'AlchemySchemas') -> Dict[str, Any]:
        """
        将对象的属性转换为字典。
        """
        return {attr: getattr(obj, attr) for attr in dir(obj) if not callable(getattr(obj, attr)) and not attr.startswith("__")}

# 使用示例
model = AlchemySchemas(id=1, name="Test", value="Something")
model_dict = AlchemySchemas.to_dict(model)
print(model_dict)

在这个例子中，AlchemySchemas 类有一个 to_dict 静态方法，它遍历对象的所有属性（不包括可调用的方法和特殊属性），并将它们作为键值对添加到字典中。
在这里插入图片描述

请注意，to_dict 方法使用了 dir() 函数来获取对象的所有属性和方法，然后使用 getattr() 函数来获取属性的值。它还检查了每个属性是否是可调用的（即方法），以及是否是特殊方法（以双下划线开始和结束），这些都应该被排除在字典之外。

二.代码实践

1.筛选方法

@staticmethod
def get_update_values(model: AlchemySchemas, **fields: Optional[Any]) -> Dict[str, Any]:
    """
    生成更新字典，只有当字段非空时才包含该字段。
    """
    return {k: v for k, v in fields.items() if v is not None}

2.更新方法

在这里插入图片描述

@staticmethod
async def update_schema(schemas_model: SchemasUpdateModel):
    """
    更新schema
    """
    db = get_sqlalchemy_db
    try:
        with Session(db.engine) as session:
            update_values = SchemasManager.get_update_values(
                model=AlchemySchemas,
                desc=schemas_model.desc,
              	name=schemas_model.name,
            )
            stmt = (
                update(AlchemySchemas)
                .where(AlchemySchemas.id == schemas_model.id)
                .values(update_values)
            )
            session.execute(stmt)
            session.commit()
    except SQLAlchemyError as e:
        logger.error(f"find schemas by id An error occurred: {e}")
    finally:
        db.close_session(session)