def retrieveAssignmentByBranch(request):
response_data = {}
try:
C = Assignment.objects.filter(assignmentCode__contains="SE")
except Exception as e:
response_data['success'] = '0'
response_data['exception'] = str(e)
else:
response_data['success'] = '1'
global data
try:
data = serializers.serialize('json', C)
except Exception as e:
data = serializers.serialize('json', [C, ])
response_data["assignment"] = json.loads(data)
return JsonResponse(response_data)
python类serialize()的实例源码
def retrieveAssignmentResponses(request):
response_data = {}
try:
C = AssignmentResponse.objects.retrieveAssignmentResponsesByStudent(request.GET)
except Exception as e:
response_data['success'] = '0'
response_data['exception'] = str(e)
else:
response_data['success'] = '1'
global data
try:
data = serializers.serialize('json', C)
except Exception as e:
data = serializers.serialize('json', [C, ])
response_data["assignment"] = json.loads(data)
return JsonResponse(response_data)
def test_serialization_partial(self):
o1 = models.PartialVersionModel(
partial=Version('0.1.1', partial=True),
optional=Version('0.2.4-rc42', partial=True),
optional_spec=None,
)
o2 = models.PartialVersionModel(
partial=Version('0.4.3-rc3+build3', partial=True),
optional='',
optional_spec=Spec('==0.1.1,!=0.1.1-alpha'),
)
data = serializers.serialize('json', [o1, o2])
obj1, obj2 = serializers.deserialize('json', data)
self.assertEqual(o1.partial, obj1.object.partial)
self.assertEqual(o1.optional, obj1.object.optional)
self.assertEqual(o2.partial, obj2.object.partial)
self.assertEqual(o2.optional, obj2.object.optional)
def post(self, request, *args, **kwargs):
obj_list = []
pessoa = Pessoa.objects.get(pk=request.POST['pessoaId'])
cliente = Cliente.objects.get(pk=request.POST['pessoaId'])
obj_list.append(cliente)
if pessoa.endereco_padrao:
obj_list.append(pessoa.endereco_padrao)
if pessoa.email_padrao:
obj_list.append(pessoa.email_padrao)
if pessoa.telefone_padrao:
obj_list.append(pessoa.telefone_padrao)
if pessoa.tipo_pessoa == 'PJ':
obj_list.append(pessoa.pessoa_jur_info)
elif pessoa.tipo_pessoa == 'PF':
obj_list.append(pessoa.pessoa_fis_info)
data = serializers.serialize('json', obj_list, fields=('indicador_ie', 'limite_de_credito', 'cnpj', 'inscricao_estadual', 'responsavel', 'cpf', 'rg', 'id_estrangeiro', 'logradouro', 'numero', 'bairro',
'municipio', 'cmun', 'uf', 'pais', 'complemento', 'cep', 'email', 'telefone',))
return HttpResponse(data, content_type='application/json')
def post(self, request, *args, **kwargs):
obj_list = []
pessoa = Pessoa.objects.get(pk=request.POST['pessoaId'])
fornecedor = Fornecedor.objects.get(pk=request.POST['pessoaId'])
obj_list.append(fornecedor)
if pessoa.endereco_padrao:
obj_list.append(pessoa.endereco_padrao)
if pessoa.email_padrao:
obj_list.append(pessoa.email_padrao)
if pessoa.telefone_padrao:
obj_list.append(pessoa.telefone_padrao)
if pessoa.tipo_pessoa == 'PJ':
obj_list.append(pessoa.pessoa_jur_info)
elif pessoa.tipo_pessoa == 'PF':
obj_list.append(pessoa.pessoa_fis_info)
data = serializers.serialize('json', obj_list, fields=('indicador_ie', 'limite_de_credito', 'cnpj', 'inscricao_estadual', 'responsavel', 'cpf', 'rg', 'id_estrangeiro', 'logradouro', 'numero', 'bairro',
'municipio', 'cmun', 'uf', 'pais', 'complemento', 'cep', 'email', 'telefone',))
return HttpResponse(data, content_type='application/json')
def _format_records(self, queryset):
attributes = self._get_attributes()
attr_fields = [x['field'] for x in attributes['fields']]
related_fields = self._get_related_fields()
json_data = serializers.serialize('json', queryset)
records = json.loads(json_data)
data = []
for index, rec in enumerate(records):
rec['fields']['id'] = rec['pk']
for field in attr_fields:
if field not in rec['fields']:
rec['fields'][field] = []
for field in related_fields:
rel_obj = getattr(queryset[index], field)
ser_rel_obj = serializers.serialize('json', [rel_obj,])
rel_data = json.loads(ser_rel_obj)[0]
rel_data['fields']['id'] = rel_data['pk']
rec['fields'][field] = rel_data['fields']
data.append(rec['fields'])
return data
def get_geojson(data, fields):
return serialize('geojson', data, geometry_field='mpoly', fields=fields)
def testview(request):
user = User.objects.all()
userset= serializers.serialize('json',user)
return render(request,'location/list.html',{'user':user})
def sync_settings(request):
current_username=str(request.POST.get('username'))
current_username=str(current_username)
try:
user=User.objects.get(username=current_username)
except Exception as e:
return HttpResponse("username is not registered")
queryset = SaveSettings.objects.filter(username__username = current_username)
queryset=serializers.serialize('json',queryset)
return HttpResponse(queryset)
def sync_reminders(request):
current_username=str(request.POST.get('username'))
current_username=str(current_username)
try:
user=User.objects.get(username=current_username)
except Exception as e:
return HttpResponse("username is not registered")
queryset = Reminder.objects.filter(username__username = current_username)
queryset=serializers.serialize('json',queryset)
return HttpResponse(queryset)
def test_model_serialization_json(self):
obj = IntegerTupleModel.objects.create(values=(3, 1, 2))
self.assertEqual(
json.loads(serialize('json', [obj])),
json.loads(
'[{'
'"fields": {"values": "|3|1|2|"}, '
'"model": "tests.integertuplemodel", "pk": 1'
'}]'
)
)
def serialize(format, queryset, **options):
"""
Serialize a queryset (or any iterator that returns database objects) using
a certain serializer.
"""
s = get_serializer(format)()
s.serialize(queryset, **options)
return s.getvalue()
def serialize_db_to_string(self):
"""
Serializes all data in the database into a JSON string.
Designed only for test runner usage; will not handle large
amounts of data.
"""
# Build list of all apps to serialize
from django.db.migrations.loader import MigrationLoader
loader = MigrationLoader(self.connection)
app_list = []
for app_config in apps.get_app_configs():
if (
app_config.models_module is not None and
app_config.label in loader.migrated_apps and
app_config.name not in settings.TEST_NON_SERIALIZED_APPS
):
app_list.append((app_config, None))
# Make a function to iteratively return every object
def get_objects():
for model in serializers.sort_dependencies(app_list):
if (model._meta.can_migrate(self.connection) and
router.allow_migrate_model(self.connection.alias, model)):
queryset = model._default_manager.using(self.connection.alias).order_by(model._meta.pk.name)
for obj in queryset.iterator():
yield obj
# Serialize to a string
out = StringIO()
serializers.serialize("json", get_objects(), indent=None, stream=out)
return out.getvalue()
def serialize(format, queryset, **options):
"""
Serialize a queryset (or any iterator that returns database objects) using
a certain serializer.
"""
s = get_serializer(format)()
s.serialize(queryset, **options)
return s.getvalue()
def serialize_objs(self, objs):
data_objs = json.loads(serializers.serialize('json', objs))
for data in data_objs:
data.update(data["fields"])
del data["fields"]
return data_objs
def obj_as_dict(self, request, obj):
data = self.serialize_obj(obj)
# serialize model instance fields datas
for field in obj._meta.get_fields():
if field.is_relation and field.concrete:
field_value = getattr(obj, field.name)
if field_value:
if field.many_to_many:
data[field.name] = self.serialize_objs(field_value.all())
elif field.many_to_one or field.one_to_one or field.one_to_many:
data[field.name] = self.serialize_obj(field_value)
# add custom admin class field to serialized bundle
model_admin_fields = self.get_model_admin_field_names(request, obj)
for field in model_admin_fields:
if field in data:
continue
if hasattr(obj, field):
f = getattr(obj, field)
data[field] = unicode(f)
if hasattr(self, field):
field_method = getattr(self, field)
if callable(field_method):
data[field] = field_method(obj)
else:
data[field] = field_method
info = self.model._meta.app_label, self.model._meta.model_name
admin_detail_url = str(reverse_lazy("admin:%s_%s_change" % info, args=(obj.id,)))
data["admin_detail_url"] = admin_detail_url
return data
def serialize(format, queryset, **options):
"""
Serialize a queryset (or any iterator that returns database objects) using
a certain serializer.
"""
s = get_serializer(format)()
s.serialize(queryset, **options)
return s.getvalue()
def serialize_model_object(obj):
"""
Serialize model into a dict representable as JSON
Args:
obj (django.db.models.Model): An instantiated Django model
Returns:
dict:
A representation of the model
"""
# serialize works on iterables so we need to wrap object in a list, then unwrap it
data = json.loads(serialize('json', [obj]))[0]
serialized = data['fields']
serialized['id'] = data['pk']
return serialized
def messages(request):
""" REST endpoint providing basic operations. GET will return the list of
all messages created so far in JSON form, POST will add a new message to
the list of messages (guestbook).
"""
if request.method == 'GET':
data = serializers.serialize("json", Message.objects.all())
return HttpResponse(data)
elif request.method == 'POST':
Message.objects.create(text=request.body)
return HttpResponse(request.body)
else:
return HttpResponse("Unsupported HTTP Verb.")
def users_json(request):
users = Employee.objects.all()
data = serializers.serialize("json", users,
fields=('first_name', 'last_name', 'department'),
use_natural_keys=True)
return HttpResponse(data, content_type='application/json')
