Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/django/core/serializers/__init__.py: 13%
106 statements
« prev ^ index » next coverage.py v6.4.4, created at 2023-07-17 14:22 -0600
1"""
2Interfaces for serializing Django objects.
4Usage::
6 from django.core import serializers
7 json = serializers.serialize("json", some_queryset)
8 objects = list(serializers.deserialize("json", json))
10To add your own serializers, use the SERIALIZATION_MODULES setting::
12 SERIALIZATION_MODULES = {
13 "csv": "path.to.csv.serializer",
14 "txt": "path.to.txt.serializer",
15 }
17"""
19import importlib
21from django.apps import apps
22from django.conf import settings
23from django.core.serializers.base import SerializerDoesNotExist
25# Built-in serializers
26BUILTIN_SERIALIZERS = {
27 "xml": "django.core.serializers.xml_serializer",
28 "python": "django.core.serializers.python",
29 "json": "django.core.serializers.json",
30 "yaml": "django.core.serializers.pyyaml",
31 "jsonl": "django.core.serializers.jsonl",
32}
34_serializers = {}
37class BadSerializer:
38 """
39 Stub serializer to hold exception raised during registration
41 This allows the serializer registration to cache serializers and if there
42 is an error raised in the process of creating a serializer it will be
43 raised and passed along to the caller when the serializer is used.
44 """
46 internal_use_only = False
48 def __init__(self, exception):
49 self.exception = exception
51 def __call__(self, *args, **kwargs):
52 raise self.exception
55def register_serializer(format, serializer_module, serializers=None):
56 """Register a new serializer.
58 ``serializer_module`` should be the fully qualified module name
59 for the serializer.
61 If ``serializers`` is provided, the registration will be added
62 to the provided dictionary.
64 If ``serializers`` is not provided, the registration will be made
65 directly into the global register of serializers. Adding serializers
66 directly is not a thread-safe operation.
67 """
68 if serializers is None and not _serializers:
69 _load_serializers()
71 try:
72 module = importlib.import_module(serializer_module)
73 except ImportError as exc:
74 bad_serializer = BadSerializer(exc)
76 module = type(
77 "BadSerializerModule",
78 (),
79 {
80 "Deserializer": bad_serializer,
81 "Serializer": bad_serializer,
82 },
83 )
85 if serializers is None:
86 _serializers[format] = module
87 else:
88 serializers[format] = module
91def unregister_serializer(format):
92 "Unregister a given serializer. This is not a thread-safe operation."
93 if not _serializers:
94 _load_serializers()
95 if format not in _serializers:
96 raise SerializerDoesNotExist(format)
97 del _serializers[format]
100def get_serializer(format):
101 if not _serializers:
102 _load_serializers()
103 if format not in _serializers:
104 raise SerializerDoesNotExist(format)
105 return _serializers[format].Serializer
108def get_serializer_formats():
109 if not _serializers:
110 _load_serializers()
111 return list(_serializers)
114def get_public_serializer_formats():
115 if not _serializers:
116 _load_serializers()
117 return [k for k, v in _serializers.items() if not v.Serializer.internal_use_only]
120def get_deserializer(format):
121 if not _serializers:
122 _load_serializers()
123 if format not in _serializers:
124 raise SerializerDoesNotExist(format)
125 return _serializers[format].Deserializer
128def serialize(format, queryset, **options):
129 """
130 Serialize a queryset (or any iterator that returns database objects) using
131 a certain serializer.
132 """
133 s = get_serializer(format)()
134 s.serialize(queryset, **options)
135 return s.getvalue()
138def deserialize(format, stream_or_string, **options):
139 """
140 Deserialize a stream or a string. Return an iterator that yields ``(obj,
141 m2m_relation_dict)``, where ``obj`` is an instantiated -- but *unsaved* --
142 object, and ``m2m_relation_dict`` is a dictionary of ``{m2m_field_name :
143 list_of_related_objects}``.
144 """
145 d = get_deserializer(format)
146 return d(stream_or_string, **options)
149def _load_serializers():
150 """
151 Register built-in and settings-defined serializers. This is done lazily so
152 that user code has a chance to (e.g.) set up custom settings without
153 needing to be careful of import order.
154 """
155 global _serializers
156 serializers = {}
157 for format in BUILTIN_SERIALIZERS:
158 register_serializer(format, BUILTIN_SERIALIZERS[format], serializers)
159 if hasattr(settings, "SERIALIZATION_MODULES"):
160 for format in settings.SERIALIZATION_MODULES:
161 register_serializer(
162 format, settings.SERIALIZATION_MODULES[format], serializers
163 )
164 _serializers = serializers
167def sort_dependencies(app_list, allow_cycles=False):
168 """Sort a list of (app_config, models) pairs into a single list of models.
170 The single list of models is sorted so that any model with a natural key
171 is serialized before a normal model, and any model with a natural key
172 dependency has it's dependencies serialized first.
174 If allow_cycles is True, return the best-effort ordering that will respect
175 most of dependencies but ignore some of them to break the cycles.
176 """
177 # Process the list of models, and get the list of dependencies
178 model_dependencies = []
179 models = set()
180 for app_config, model_list in app_list:
181 if model_list is None:
182 model_list = app_config.get_models()
184 for model in model_list:
185 models.add(model)
186 # Add any explicitly defined dependencies
187 if hasattr(model, "natural_key"):
188 deps = getattr(model.natural_key, "dependencies", [])
189 if deps:
190 deps = [apps.get_model(dep) for dep in deps]
191 else:
192 deps = []
194 # Now add a dependency for any FK relation with a model that
195 # defines a natural key
196 for field in model._meta.fields:
197 if field.remote_field:
198 rel_model = field.remote_field.model
199 if hasattr(rel_model, "natural_key") and rel_model != model:
200 deps.append(rel_model)
201 # Also add a dependency for any simple M2M relation with a model
202 # that defines a natural key. M2M relations with explicit through
203 # models don't count as dependencies.
204 for field in model._meta.many_to_many:
205 if field.remote_field.through._meta.auto_created:
206 rel_model = field.remote_field.model
207 if hasattr(rel_model, "natural_key") and rel_model != model:
208 deps.append(rel_model)
209 model_dependencies.append((model, deps))
211 model_dependencies.reverse()
212 # Now sort the models to ensure that dependencies are met. This
213 # is done by repeatedly iterating over the input list of models.
214 # If all the dependencies of a given model are in the final list,
215 # that model is promoted to the end of the final list. This process
216 # continues until the input list is empty, or we do a full iteration
217 # over the input models without promoting a model to the final list.
218 # If we do a full iteration without a promotion, that means there are
219 # circular dependencies in the list.
220 model_list = []
221 while model_dependencies:
222 skipped = []
223 changed = False
224 while model_dependencies:
225 model, deps = model_dependencies.pop()
227 # If all of the models in the dependency list are either already
228 # on the final model list, or not on the original serialization list,
229 # then we've found another model with all it's dependencies satisfied.
230 if all(d not in models or d in model_list for d in deps):
231 model_list.append(model)
232 changed = True
233 else:
234 skipped.append((model, deps))
235 if not changed:
236 if allow_cycles:
237 # If cycles are allowed, add the last skipped model and ignore
238 # its dependencies. This could be improved by some graph
239 # analysis to ignore as few dependencies as possible.
240 model, _ = skipped.pop()
241 model_list.append(model)
242 else:
243 raise RuntimeError(
244 "Can't resolve dependencies for %s in serialized app list."
245 % ", ".join(
246 model._meta.label
247 for model, deps in sorted(
248 skipped, key=lambda obj: obj[0].__name__
249 )
250 ),
251 )
252 model_dependencies = skipped
254 return model_list