Last active
July 12, 2023 02:44
-
-
Save jizhang/18dafaafc5884e36e82fedc474c31edd to your computer and use it in GitHub Desktop.
SQLAlchemy result serialization
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from typing import Any, Iterable, List, Dict | |
from decimal import Decimal | |
from datetime import datetime | |
from sqlalchemy import Row | |
from sqlalchemy.orm import DeclarativeMeta | |
def row_to_dict(row) -> dict: | |
if isinstance(row, Row): | |
row_dict = dict(row) | |
elif isinstance(row.__class__, DeclarativeMeta): | |
row_dict = {c.name: getattr(row, c.name) for c in row.__table__.columns} | |
else: | |
raise ValueError('Unsupported row type.') | |
for key, value in row_dict.items(): | |
if isinstance(value, Decimal): | |
row_dict[key] = float(value) | |
elif isinstance(value, datetime): | |
row_dict[key] = value.strftime('%Y-%m-%d %H:%M:%S') | |
return row_dict | |
def row_to_dict(row: DeclarativeBase) -> Dict[str, Any]: | |
return {col.name: getattr(row, col.name) for col in row.__table__.columns} | |
def rows_to_list(rows: Iterable) -> List[dict]: | |
return [row_to_dict(i) for i in rows] |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment