serializers.py 1.5 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465
  1. #!/usr/bin/env python3
  2. # -*- coding: utf-8 -*-
  3. from decimal import Decimal
  4. from typing import Any, Sequence, TypeVar
  5. import msgspec
  6. from sqlalchemy import Row, RowMapping
  7. from starlette.responses import JSONResponse
  8. RowData = Row | RowMapping | Any
  9. R = TypeVar('R', bound=RowData)
  10. def select_columns_serialize(row: R) -> dict:
  11. """
  12. Serialize SQLAlchemy select table columns, does not contain relational columns
  13. :param row:
  14. :return:
  15. """
  16. obj_dict = {}
  17. for column in row.__table__.columns.keys():
  18. val = getattr(row, column)
  19. if isinstance(val, Decimal):
  20. if val % 1 == 0:
  21. val = int(val)
  22. val = float(val)
  23. obj_dict[column] = val
  24. return obj_dict
  25. def select_list_serialize(row: Sequence[R]) -> list:
  26. """
  27. Serialize SQLAlchemy select list
  28. :param row:
  29. :return:
  30. """
  31. ret_list = [select_columns_serialize(_) for _ in row]
  32. return ret_list
  33. def select_as_dict(row: R) -> dict:
  34. """
  35. Converting SQLAlchemy select to dict, which can contain relational data,
  36. depends on the properties of the select object itself
  37. :param row:
  38. :return:
  39. """
  40. obj_dict = row.__dict__
  41. if '_sa_instance_state' in obj_dict:
  42. del obj_dict['_sa_instance_state']
  43. return obj_dict
  44. class MsgSpecJSONResponse(JSONResponse):
  45. """
  46. JSON response using the high-performance msgspec library to serialize data to JSON.
  47. """
  48. def render(self, content: Any) -> bytes:
  49. return msgspec.json.encode(content)