Skip to content

Latest commit

 

History

History
92 lines (72 loc) · 2.71 KB

6.SQLAlchemy结果集转字典.md

File metadata and controls

92 lines (72 loc) · 2.71 KB

SQLAlchemy结果集转字典

1.结果集类型

  • Model(flask_sqlalchemy): from flask_sqlalchemy.model import Model
    • DeclarativeMeta(原生SQLAlchemy): from sqlalchemy.orm.decl_api import DeclarativeMeta
    • 单个模型对象
    • model.__table__: sqlalchemy.sql.schema.Table
      • model.__table__.columns: sqlalchemy.sql.base.ImmutableColumnCollection
        • model.__table__.columns.__iter__: Iterator[sqlalchemy.sql.schema.Column]
          • column: sqlalchemy.sql.schema.Column
            • column.name: str
              • 字段名
  • Row: from sqlalchemy.engine.row import Row
    • 行数据对象
    • row.keys(): List[str]
      • 字段名的列表集
  • List[Union[Model, DeclarativeMeta, Row]]
    • 上述对象的列表集
  • None
    • 结果集为空

2.扩展字典

# libs/dict.py
# coding: utf-8


class ExtDict(dict):
    """扩展字典"""

    def __setattr__(self, key, value):
        """ Implement self.key = value """
        self[key] = value

    def __getattr__(self, item):
        """ Implement self.key """
        if item not in self:
            raise AttributeError(f"'{self.__class__.__name__}' object has no attribute '{item}'")
        return self[item]

3.SQLAlchemy结果集转字典

# libs/sqlalchemy/result.py
# coding: utf-8

from typing import Union, List
from sqlalchemy.engine.row import Row
from sqlalchemy.orm.decl_api import DeclarativeMeta
try:
    from flask_sqlalchemy.model import Model
except ImportError:
    Model = DeclarativeMeta

from libs.dict import ExtDict


def model_format(model: Union[DeclarativeMeta, Model]) -> dict:
    """ 模型数据格式化 """
    return ExtDict((c.name, getattr(model, c.name, None)) for c in model.__table__.columns)


def row_format(result: Row) -> dict:
    """ 行数据格式化 """
    return ExtDict(zip(result.keys(), result))


def list_format(result: List[Union[Row, DeclarativeMeta, Model]]) -> List[dict]:
    """ 行数据列表格式化 """
    lst = []
    for items in result:
        if isinstance(items, Row):
            elem = row_format(items)
        elif isinstance(items, (DeclarativeMeta, Model)):
            elem = model_format(items)
        lst.append(elem)
    return lst


def result_format(result: Union[DeclarativeMeta, Model, Row, List, None]) -> Union[dict, List[dict]]:
    """ sqlalchemy query result 格式化 """
    if not result:
        return result
    elif isinstance(result, list):
        return list_format(result)
    elif isinstance(result, Row):
        return row_format(result)
    return model_format(result)