Created
February 6, 2022 01:34
-
-
Save MarlonJD/c1d6610c12a22e4618e2873ea3034a60 to your computer and use it in GitHub Desktop.
Huge amount of loaddata
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
""" | |
Serialize data to/from JSON | |
""" | |
import datetime | |
import decimal | |
import json | |
import uuid | |
from django.core.serializers.base import DeserializationError | |
from django.core.serializers.python import ( | |
Deserializer as PythonDeserializer, Serializer as PythonSerializer, | |
) | |
from django.utils.duration import duration_iso_string | |
from django.utils.functional import Promise | |
from django.utils.timezone import is_aware | |
import ijson | |
import six | |
import sys | |
class Serializer(PythonSerializer): | |
"""Convert a queryset to JSON.""" | |
internal_use_only = False | |
def _init_options(self): | |
self._current = None | |
self.json_kwargs = self.options.copy() | |
self.json_kwargs.pop('stream', None) | |
self.json_kwargs.pop('fields', None) | |
if self.options.get('indent'): | |
# Prevent trailing spaces | |
self.json_kwargs['separators'] = (',', ': ') | |
self.json_kwargs.setdefault('cls', DjangoJSONEncoder) | |
self.json_kwargs.setdefault('ensure_ascii', False) | |
def start_serialization(self): | |
self._init_options() | |
self.stream.write("[") | |
def end_serialization(self): | |
if self.options.get("indent"): | |
self.stream.write("\n") | |
self.stream.write("]") | |
if self.options.get("indent"): | |
self.stream.write("\n") | |
def end_object(self, obj): | |
# self._current has the field data | |
indent = self.options.get("indent") | |
if not self.first: | |
self.stream.write(",") | |
if not indent: | |
self.stream.write(" ") | |
if indent: | |
self.stream.write("\n") | |
json.dump(self.get_dump_object(obj), self.stream, **self.json_kwargs) | |
self._current = None | |
def getvalue(self): | |
# Grandparent super | |
return super(PythonSerializer, self).getvalue() | |
import ijson | |
def Deserializer(stream_or_string, **options): | |
if isinstance(stream_or_string, six.string_types): | |
stream_or_string = six.BytesIO(stream_or_string.encode('utf-8')) | |
try: | |
objects = ijson.items(stream_or_string, 'item') | |
for obj in PythonDeserializer(objects, **options): | |
yield obj | |
except GeneratorExit: | |
raise | |
except Exception as e: | |
# Map to deserializer error | |
six.reraise(DeserializationError, DeserializationError(e), sys.exc_info()[2]) | |
class DjangoJSONEncoder(json.JSONEncoder): | |
""" | |
JSONEncoder subclass that knows how to encode date/time, decimal types, and | |
UUIDs. | |
""" | |
def default(self, o): | |
# See "Date Time String Format" in the ECMA-262 specification. | |
if isinstance(o, datetime.datetime): | |
r = o.isoformat() | |
if o.microsecond: | |
r = r[:23] + r[26:] | |
if r.endswith('+00:00'): | |
r = r[:-6] + 'Z' | |
return r | |
elif isinstance(o, datetime.date): | |
return o.isoformat() | |
elif isinstance(o, datetime.time): | |
if is_aware(o): | |
raise ValueError("JSON can't represent timezone-aware times.") | |
r = o.isoformat() | |
if o.microsecond: | |
r = r[:12] | |
return r | |
elif isinstance(o, datetime.timedelta): | |
return duration_iso_string(o) | |
elif isinstance(o, (decimal.Decimal, uuid.UUID, Promise)): | |
return str(o) | |
else: | |
return super().default(o) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment