Created
May 11, 2022 07:55
-
-
Save inscapist/10439785a75c89627f085f2c80716daf to your computer and use it in GitHub Desktop.
Better json.dumps supporting set, datetime, timedelta, decimal, object, and sqlalchemy objects
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
__all__ = ["better_dumps", "better_dump", "json_debug"] | |
import dataclasses | |
import datetime | |
import decimal | |
import json | |
from inspect import ismethod | |
import jsonpickle | |
from loguru import logger | |
from sqlalchemy.ext.declarative import DeclarativeMeta | |
class BasicJSONEncoder(json.JSONEncoder): | |
def default(_, o): | |
if isinstance(o, (datetime.datetime, datetime.date, datetime.time)): | |
return o.isoformat() | |
if isinstance(o, datetime.timedelta): | |
return (datetime.datetime.min + o).time().isoformat() | |
if dataclasses.is_dataclass(o): | |
return dataclasses.asdict(o) | |
if isinstance(o, decimal.Decimal): | |
return float(o) | |
if isinstance(o, bytes): | |
try: | |
return o.decode() | |
except: | |
return "failed-to-encode(1)" | |
if isinstance(o, set): | |
return list(o) | |
if isinstance(o, object): | |
return o.__dict__ | |
return super().default(o) | |
class BetterJSONEncoder(json.JSONEncoder): | |
def default(_, o): | |
if isinstance(o.__class__, DeclarativeMeta): | |
# an SQLAlchemy class | |
fields = {} | |
for field in [ | |
x for x in dir(o) if not x.startswith("_") and x != "metadata" | |
]: | |
data = o.__getattribute__(field) | |
if not ismethod(data): | |
try: | |
json.dumps(data, cls=BasicJSONEncoder) | |
fields[field] = data | |
except (TypeError, AttributeError) as e: | |
logger.debug(e) | |
fields[field] = "failed-to-encode(2)" | |
return fields | |
if isinstance(o, (datetime.datetime, datetime.date, datetime.time)): | |
return o.isoformat() | |
if isinstance(o, datetime.timedelta): | |
return (datetime.datetime.min + o).time().isoformat() | |
if dataclasses.is_dataclass(o): | |
return dataclasses.asdict(o) | |
if isinstance(o, decimal.Decimal): | |
return float(o) | |
if isinstance(o, bytes): | |
try: | |
return o.decode() | |
except: | |
return "failed-to-encode(1)" | |
if isinstance(o, set): | |
return list(o) | |
if isinstance(o, object): | |
return o.__dict__ | |
return super().default(o) | |
def better_dumps(obj, doprint=True, indent=True): | |
if indent: | |
s = json.dumps(obj, cls=BetterJSONEncoder, indent=indent) | |
else: | |
s = json.dumps(obj, cls=BetterJSONEncoder) | |
if doprint: | |
print(s) | |
else: | |
return s | |
def simplify_object(obj): | |
s = json.dumps(obj, cls=BetterJSONEncoder) | |
return json.loads(s) | |
def better_dump(obj, file_location: str): | |
with open(file_location, "w") as outfile: | |
json.dump(obj, outfile, cls=BetterJSONEncoder, indent=2) | |
def json_debug(obj): | |
""" | |
use this only as last resort. when BetterJSONEncoder fails | |
""" | |
data = json.dumps(json.loads(str(jsonpickle.encode(obj))), indent=2, sort_keys=True) | |
print(data) | |
def sqlalchemy_to_dict(result_proxy): | |
""" | |
Convert the result of fetch(), fetchall() into dict from result proxy | |
""" | |
return [ | |
{column: value for column, value in rowproxy.items()} | |
for rowproxy in result_proxy | |
] |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Usage: