VberAdminPy/VapAdmin/BACKEND/utils/sqlalchemy_util.py

from sqlalchemy.engine.row import Row

from domain.models.base_model import Base


class SqlalchemyUtil:
    """
    sqlalchemy工具类
    """

    @classmethod
    def base_to_dict(cls, obj: Base):
        """
        将sqlalchemy模型对象转换为字典

        :param obj: sqlalchemy模型对象
        :return: 字典结果
        """
        base_dict = obj.__dict__.copy()
        base_dict.pop('_sa_instance_state', None)

        return base_dict

    @classmethod
    def serialize_result(cls, result):
        """
        将sqlalchemy查询结果序列化

        :param result: sqlalchemy查询结果
        :return: 序列化结果
        """
        if isinstance(result, Base):
            return cls.base_to_dict(result)
        elif isinstance(result, list):
            return [cls.serialize_result(row) for row in result]
        elif isinstance(result, Row):
            if all([isinstance(row, Base) for row in result]):
                return [cls.base_to_dict(row) for row in result]
            elif any([isinstance(row, Base) for row in result]):
                return [cls.serialize_result(row) for row in result]
            else:
                return {key: getattr(result, key) for key in result.keys()}
        return result