Python操作数据库,只更新非空字段

一.基本介绍

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

Python操作数据库,只更新非空字段