import json
import os
from datetime import date, datetime, timedelta
from decimal import Decimal
from typing import Any
from sqlalchemy import RowMapping
def json_serial(obj: Any) -> Any:
if isinstance(obj, datetime):
return obj.strftime('%Y-%m-%dT%H:%M:%S+00:00')
if isinstance(obj, date):
# momentjs needs timezone to parse date correctly
# todo return obj.strftime('%Y-%m-%dT%H:%M:%S.%fZ')
return obj.strftime('%Y-%m-%dT00:00:00+00:00')
if isinstance(obj, Decimal):
return '0' if obj == 0 else str(obj)
if isinstance(obj, set):
return list(obj)
if isinstance(obj, timedelta):
return str(obj)
if isinstance(obj, RowMapping):
return dict(obj)
raise TypeError("Type %s not serializable" % type(obj))
# todo remove
def openx_json_serde_serial(obj: Any) -> int | float | str | None:
"""Serializer for AWS Firehose correctly parse timestamps and save
parquet file on S3"""
if isinstance(obj, datetime):
return obj.strftime('%Y-%m-%dT%H:%M:%S.%fZ')
if isinstance(obj, Decimal):
if obj == obj.to_integral_value():
return int(obj)
else:
return float(obj)
return None
def dumps(obj: Any, default: Any = json_serial, **kwargs: Any) -> str:
return json.dumps(obj, default=default, **kwargs)
def loads(s: str, **kwargs: Any) -> Any:
return json.loads(s, **kwargs)
def load(path: str) -> Any:
if not os.path.exists(path):
raise Exception(f'File {path} not found')
with open(path, 'r') as f:
return loads(f.read())
