json.loads take a string as input and returns a dictionary as output.
json.dumps take a dictionary as input and returns a string as output.
import json
from bson import ObjectId
class JSONEncoder(json.JSONEncoder):
def default(self, o):
if isinstance(o, ObjectId):
return str(o)
if isinstance(o, datetime.datetime):
return str(o)
return json.JSONEncoder.default(self, o)
JSONEncoder().encode(qm)
load from a file,
dump to a file
users = {}
async def on_message(message):
with open('users.json', 'w') as f:
json.dump(users, f)
async def on_ready():
with open('users.json') as f:
try:
users = json.load(f)
except:
users = {}
from json import JSONEncoder, dumps
class TMCSerializer(JSONEncoder):
def default(self, value: Any) -> str:
"""JSON serialization conversion function."""
# If it's an IP, which is not normally
# serializable, convert to string.
if isinstance(value, IPv4Address):
return str(value)
# Here you can have other handling for your
# UUIDs, or datetimes, or whatever else you
# have.
# Otherwise, default to super
return super(TMCSerializer, self).default(value)
json_str = json.dumps(some_dict, cls=TMCSerializer)
class CustomEncoder(json.JSONEncoder):
"""A C{json.JSONEncoder} subclass to encode documents that have fields of
type C{bson.objectid.ObjectId}, C{datetime.datetime}
"""
def default(self, obj):
if isinstance(obj, bson.objectid.ObjectId):
return str(obj)
elif isinstance(obj, datetime.datetime):
return obj.isoformat()
return json.JSONEncoder.default(self, obj)
enc = CustomEncoder()
enc.encode(doc)
https://stackoverflow.com/questions/16586180/typeerror-objectid-is-not-json-serializable#16586277
import json
from bson import ObjectId
class JSONEncoder(json.JSONEncoder):
def default(self, o):
if isinstance(o, ObjectId):
return str(o)
return json.JSONEncoder.default(self, o)
JSONEncoder().encode(analytics)
# or
json.encode(analytics, cls=JSONEncoder)
from datetime import datetime, date
import isodate as iso
from bson import ObjectId
from flask.json import JSONEncoder
from werkzeug.routing import BaseConverter
class MongoJSONEncoder(JSONEncoder):
def default(self, o):
if isinstance(o, (datetime, date)):
return iso.datetime_isoformat(o)
if isinstance(o, ObjectId):
return str(o)
else:
return super().default(o)
class ObjectIdConverter(BaseConverter):
def to_python(self, value):
return ObjectId(value)
def to_url(self, value):
return str(value)
class JsonEncoder():
def encode(self, o):
if '_id' in o:
o['_id'] = str(o['_id'])
return o
https://pynative.com/make-python-class-json-serializable/
https://pynative.com/python-serialize-datetime-into-json/
import json
class Employee:
def __init__(self, name, salary, address):
self.name = name
self.salary = salary
self.address = address
def toJson(self):
return json.dumps(self, default=lambda o: o.__dict__)
class Address:
def __init__(self, city, street, pin):
self.city = city
self.street = street
self.pin = pin
address = Address("Alpharetta", "7258 Spring Street", "30004")
employee = Employee("John", 9000, address)
print("Encode into JSON formatted Data")
employeeJSONData = json.dumps(employee.toJson(), indent=4)
print(employeeJSONData)
# Let's load it using the load method to check if we can decode it or not.
print("Decode JSON formatted Data")
employeeJSON = json.loads(employeeJSONData)
print(employeeJSON)
import json
class MyEncoder(json.JSONEncoder):
"""
JSONEncoder subclass that leverages an object's `__json__()` method,
if available, to obtain its default JSON representation.
"""
def default(self, obj):
if hasattr(obj, '__json__'):
return obj.__json__()
return json.JSONEncoder.default(self, obj)
class MyClass(object):
name = 'John'
age = 30
def __json__(self):
return {'name': self.name, 'age': self.age}
json.dumps(MyClass(), cls=MyEncoder)
# {"age": 30, "name": "John"}
import json
from datetime import datetime, date
import uuid
from uuid import UUID
from bson import ObjectId
from typing import List, Union
from pydantic import BaseModel, Field
from utils.json import mongo_json_encoder
def mongo_json_encoder(record: [dict, list, BaseModel]):
"""
This is a json_encoder designed specially for dump mongodb records.
It can deal with both record_item and record_list type queried from mongodb.
You can extend the encoder ability in the recursive function `convert_type`.
I just covered the following datatype: datetime, date, UUID, ObjectID.
Contact me if any further support needs.
Attention: it will change the raw record, so copy it before operating this function if necessary.
Parameters
----------
**record**: a dict or a list, like the queried documents from mongodb.
Returns
-------
json formatted data.
"""
def convert_type(data):
if isinstance(data, (datetime, date)):
# ISO format: data.isoformat()
return str(data)
elif isinstance(data, (UUID, ObjectId)):
return str(data)
elif isinstance(data, list):
return list(map(convert_type, data))
elif isinstance(data, dict):
return mongo_json_encoder(data)
try:
json.dumps(data)
return data
except TypeError:
raise TypeError({
"error_msg": "暂不支持此类型序列化",
"key": key,
"value": value,
"type": type(value)
})
# add support for BaseModel
if isinstance(record, BaseModel):
return mongo_json_encoder(record.dict(by_alias=True))
elif isinstance(record, dict):
for key, value in record.items():
record[key] = convert_type(value)
return record
else:
return list(map(mongo_json_encoder, record))
def mongo_json_encoder_decorator(func):
"""
this is a decorator for converting the queried documents from mongodb
Parameters
----------
func
Returns
-------
"""
def wrapper(*args, **kwargs):
res = func(*args, **kwargs)
return mongo_json_encoder(res)
return wrapper
# tests
class FriendBase(BaseModel):
class Config:
arbitrary_types_allowed = True
allow_population_by_field_name = True
id: Union[str, UUID, ObjectId] = Field(alias='_id')
name: str
class Friend(FriendBase):
friends: List[FriendBase] = []
f_1 = Friend(id='test', name='test')
f_2 = Friend(id=uuid.uuid1(), name='test', friends=[f_1])
f_3 = Friend(id=ObjectId(), name='test', friends=[f_1, f_2])
i_1 = f_1.dict(by_alias=True)
i_2 = f_2.dict(by_alias=True)
i_3 = f_3.dict(by_alias=True)
j_1 = mongo_json_encoder(i_1.copy())
j_2 = mongo_json_encoder(i_2.copy())
j_3 = mongo_json_encoder(i_3.copy())
j_all = [f_1, f_2, f_3]
assert i_1 == j_1
assert i_2 == j_2, "this should not pass"
assert i_3 == j_3, "this should not pass"
from datetime import datetime, date
from bson import ObjectId, Decimal128, json_util
import json
# p.180
def json_serialize_date(obj):
if isinstance(obj, (date, datetime)):
return obj.strftime('%Y-%m-%dT%H:%M:%S')
elif isinstance(obj, date):
return obj.isoformat()
raise TypeError ("The type %s not serializable." % type(obj))
def json_serialize_oid(obj):
if isinstance(obj, ObjectId):
return str(obj)
if isinstance(obj, (date, datetime)):
return obj.strftime('%Y-%m-%dT%H:%M:%S')
elif isinstance(obj, float):
return float(obj)
# return Decimal128(obj).to_decimal()
elif isinstance(obj, int):
return int(obj)
raise TypeError("The type %s not serializable." % type(obj))
class JSONEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, ObjectId):
return str(obj)
elif isinstance(obj, (date, datetime)):
return obj.strftime('%Y-%m-%d')
# return obj.strftime('%Y-%m-%dT%H:%M:%S')
# return obj.isoformat()
elif isinstance(obj, float):
return float(obj)
elif isinstance(obj, int):
return int(obj)
return json.JSONEncoder.default(self, obj)
def parse_json(data):
# return json.loads(json_util.dumps(data))
return json.loads(json.dumps(data, default=str))
class ORJSONEncoder:
def default(self, obj):
if isinstance(obj, datetime):
return obj.__str__()
elif isinstance(obj, Decimal):
return str(obj)
# elif isinstance(obj, ApiQuerySet):
# return str(obj)
else:
return str(obj)
raise TypeError
def loads_encoder(obj):
if isinstance(obj, datetime):
return obj.__str__()
elif isinstance(obj, Decimal):
return str(obj)
else:
return str(obj)
raise TypeError