def lenses(request):
if request.method == 'GET':
arguments = {}
for k, v in request.GET.items():
if v:
arguments[k] = v
l = Lens.objects.filter(**arguments)
serializer = LensSerializer(l, many=True)
return JSONResponse(serializer.data)
elif request.method == 'POST':
data = JSONParser().parse(request)
serializer = LensSerializer(data=data, many=True)
if serializer.is_valid():
serializer.save()
return JSONResponse(serializer.data, status=201)
return JSONResponse(serializer.errors, status=400)
python类JSONParser()的实例源码
def scales(request):
if request.method == 'GET':
arguments = {}
for k, v in request.GET.items():
if v:
arguments[k] = v
s = Scale.objects.filter(**arguments)
serializer = ScaleSerializer(s, many=True)
return JSONResponse(serializer.data)
elif request.method == 'POST':
data = JSONParser().parse(request)
serializer = ScaleSerializer(data=data, many=True)
if serializer.is_valid():
serializer.save()
return JSONResponse(serializer.data, status=201)
return JSONResponse(serializer.errors, status=400)
def node_list(request):
"""
List all nodes, or create a new node.
"""
if request.method == 'GET':
nodes = Node.objects.all()
serializer = NodeSerializer(nodes, many=True)
return JsonResponse(serializer.data, safe=False)
elif request.method == 'POST':
data = JSONParser().parse(request)
serializer = NodeSerializer(data=data)
if serializer.is_valid():
serializer.save()
return JsonResponse(serializer.data, status=201)
return JsonResponse(serializer.errors, status=400)
def nodes_detail(request, pk):
"""
Retrieve, update or delete a node.
"""
try:
node = Node.objects.get(intent=pk)
except Node.DoesNotExist:
return HttpResponse(status=404)
if request.method == 'GET':
serializer = NodeSerializer(node)
return JsonResponse(serializer.data)
elif request.method == 'PUT':
data = JSONParser().parse(request)
serializer = NodeSerializer(node, data=data)
if serializer.is_valid():
serializer.save()
return JsonResponse(serializer.data)
return JsonResponse(serializer.errors, status=400)
elif request.method == 'DELETE':
node.delete()
return HttpResponse(status=204)
def book_list(request):
"""
List all books, or create a new book.
"""
if request.method == 'GET':
books = Book.objects.all()
serializer = BookSerializer(books, many=True)
return JsonResponse(serializer.data, safe=False)
elif request.method == 'POST':
data = JSONParser().parse(request)
serializer = BookSerializer(data=data)
if serializer.is_valid():
serializer.save()
return JsonResponse(serializer.data, status=201)
return JsonResponse(serializer.errors, status=400)
# Returns a Json response for the book of id pk if the request is a GET request.
# Adds a new book to the database if the request is a PUT request.
# Deletes book of id pk if the request is a DELETE request.
def book_detail_by_author(request, author):
"""
Retrieve, update or delete a book.
"""
try:
book = Book.objects.filter(author=author)
except Book.DoesNotExist:
return HttpResponse(status=404)
if request.method == 'GET':
serializer = BookSerializer(book,many=True)
return JsonResponse(serializer.data,safe=False)
elif request.method == 'POST':
data = JSONParser().parse(request)
serializer = BookSerializer(book, data=data)
if serializer.is_valid():
serializer.save()
return JsonResponse(serializer.data)
return JsonResponse(serializer.errors, status=400)
elif request.method == 'DELETE':
book.delete()
return HttpResponse(status=204)
def book_year(request, year):
"""
Retrieve, update or delete a book.
"""
try:
book = Book.objects.filter(year=year)
except Book.DoesNotExist:
return HttpResponse(status=404)
if request.method == 'GET':
serializer = BookSerializer(book)
return JsonResponse(serializer.data)
elif request.method == 'PUT':
data = JSONParser().parse(request)
serializer = BookSerializer(book, data=data)
if serializer.is_valid():
serializer.save()
return JsonResponse(serializer.data)
return JsonResponse(serializer.errors, status=400)
elif request.method == 'DELETE':
book.delete()
return HttpResponse(status=204)
def author_list(request):
"""
List all authors, or create a new author.
"""
if request.method == 'GET':
authors = Author.objects.all()
serializer = AuthorSerializer(authors, many=True)
return JsonResponse(serializer.data, safe=False)
elif request.method == 'POST':
data = JSONParser().parse(request)
serializer = AuthorSerializer(data=data)
if serializer.is_valid():
serializer.save()
return JsonResponse(serializer.data, status=201)
return JsonResponse(serializer.errors, status=400)
def newhouse_detail(request, pk):
"""
Retrieve, update or delete a code newhouse.
"""
try:
newhouse = NewHouse.objects.get(pk=pk)
except NewHouse.DoesNotExist:
return HttpResponse(status=404)
if request.method == 'GET':
serializer = NewHouseDetailSerializer(newhouse)
return JsonResponse(serializer.data)
elif request.method == 'PUT':
data = JSONParser().parse(request)
serializer = NewHouseDetailSerializer(newhouse, data=data)
if serializer.is_valid():
serializer.save()
return JsonResponse(serializer.data)
return JsonResponse(serializer.errors, status=400)
elif request.method == 'DELETE':
newhouse.delete()
return HttpResponse(status=204)
def lens(request, lens_id=0):
if request.method == 'GET':
try:
l = Lens.objects.get(id=lens_id)
serializer = LensDetailSerializer(l)
return JSONResponse(serializer.data)
except:
raise Http404("Lens not found")
elif request.method == 'POST':
data = JSONParser().parse(request)
serializer = LensDetailSerializer(data=data)
if serializer.is_valid():
serializer.save()
return JSONResponse(serializer.data, status=201)
return JSONResponse(serializer.errors, status=400)
elif request.method == 'PUT':
try:
l = Lens.objects.get(id=lens_id)
data = JSONParser().parse(request)
serializer = LensDetailSerializer(l, data=data)
if serializer.is_valid():
serializer.save()
return JSONResponse(serializer.data, status=201)
return JSONResponse(serializer.errors, status=400)
except:
raise Http404("Lens not found")
elif request.method == 'DELETE':
try:
l = Lens.objects.get(id=lens_id)
l.delete()
return HttpResponse(status=200)
except:
raise Http404("Lens not found")
def test_json_to_StatusReport(self):
json = JSONRenderer().render(self.expected_dict)
stream = BytesIO(json)
data = JSONParser().parse(stream)
serializer = StatusReportSerializer(self.new_status, data=data)
self.assertTrue(serializer.is_valid())
status = serializer.save()
self.assertEqual(self.new_status, status)
self.assertEqual(self.new_status.id, status.id)
self.assertEqual(self.new_status.status, status.status)
self.assertEqual(self.new_status.when, status.when)
self.assertEqual(self.new_status.user, status.user)
def test_json_to_new_StatusReport(self):
json = JSONRenderer().render(self.expected_dict)
stream = BytesIO(json)
data = JSONParser().parse(stream)
serializer = StatusReportSerializer(data=data)
self.assertTrue(serializer.is_valid())
status = serializer.save()
self.assertEqual(self.new_status.status, status.status)
self.assertIsNotNone(status.when)
self.assertEqual(self.new_status.user, status.user)
def parse_json_response(json):
""" parse the json response """
rendered = JSONRenderer().render(json)
stream = BytesIO(rendered)
return JSONParser().parse(stream)
def book_detail(request, pk):
"""
Retrieve, update or delete a book.
"""
try:
book = Book.objects.get(pk=pk)
except Book.DoesNotExist:
return HttpResponse(status=404)
if request.method == 'GET':
serializer = BookSerializer(book)
return JsonResponse(serializer.data)
elif request.method == 'PUT':
data = JSONParser().parse(request)
serializer = BookSerializer(book, data=data)
if serializer.is_valid():
serializer.save()
return JsonResponse(serializer.data)
return JsonResponse(serializer.errors, status=400)
elif request.method == 'DELETE':
book.delete()
return HttpResponse(status=204)
# Returns a Json response for the book of title if the request is a GET request.
# Adds a new book to the database if the request is a PUT request.
# Deletes book of title if the request is a DELETE request.
def book_detail_by_title(request, title):
"""
Retrieve, update or delete a book.
"""
try:
book = Book.objects.get(title=title)
except Book.DoesNotExist:
return HttpResponse(status=404)
if request.method == 'GET':
serializer = BookSerializer(book)
return JsonResponse(serializer.data)
data = JSONParser().parse(request)
elif request.method == 'PUT':
serializer = BookSerializer(book, data=data)
if serializer.is_valid():
serializer.save()
return JsonResponse(serializer.data)
return JsonResponse(serializer.errors, status=400)
elif request.method == 'DELETE':
book.delete()
return HttpResponse(status=204)
# Returns a Json response for the book in a given language if the request is a GET request.
# Adds a new book to the database if the request is a PUT request.
# Deletes book of title if the request is a DELETE request.
def book_rate(request, rate):
"""
Retrieve, update or delete a book rating.
"""
try:
book = Book.objects.filter(rate=rate)
except Book.DoesNotExist:
return HttpResponse(status=404)
if request.method == 'GET':
serializer = BookSerializer(book)
return JsonResponse(serializer.data)
elif request.method == 'PUT':
data = JSONParser().parse(request)
serializer = BookSerializer(book, data=data)
if serializer.is_valid():
serializer.save()
return JsonResponse(serializer.data)
return JsonResponse(serializer.errors, status=400)
elif request.method == 'DELETE':
book.delete()
return HttpResponse(status=204)
# Returns a Json response containing all the authors.
def newhouse_list(request):
"""
List all code newhouses, or create a new house.
"""
try:
page_size = int(request.GET.get('page_size', 10))
page = int(request.GET.get('page', 1))
except (TypeError, ValueError):
return JsonResponse(code=status.HTTP_400_BAD_REQUEST, desc='page and page_size must be integer!')
if request.method == 'GET':
houses = NewHouse.objects.all()
print(page_size)
paginator = Paginator(houses, page_size)
total = paginator.num_pages
try:
houses = paginator.page(page)
except PageNotAnInteger:
houses = paginator.page(1)
except EmptyPage:
houses = paginator.page(paginator.num_pages)
serializer = NewHouseListSerializer(houses, many=True)
# return JsonResponse(serializer.data, safe=False)
return JsonResponse({
'data': serializer.data,
'page': page,
'total': total
}) # ??
elif request.method == 'POST':
data = JSONParser().parse(request)
serializer = NewHouseListSerializer(data=data)
if serializer.is_valid():
serializer.save()
return JsonResponse(serializer.data, status=201)
return JsonResponse(serializer.errors, status=400)
def deserialize_json(stream, username):
"""
Given a stream containing JSON, deserializers the JSON into ConfigurationModel instances.
The stream is expected to be in the following format:
{ "model": "config_models.ExampleConfigurationModel",
"data":
[
{ "enabled": True,
"color": "black"
...
},
{ "enabled": False,
"color": "yellow"
...
},
...
]
}
If the provided stream does not contain valid JSON for the ConfigurationModel specified,
an Exception will be raised.
Arguments:
stream: The stream of JSON, as described above.
username: The username of the user making the change. This must match an existing user.
Returns: the number of created entries
"""
parsed_json = JSONParser().parse(stream)
serializer_class = get_serializer_class(apps.get_model(parsed_json["model"]))
list_serializer = serializer_class(data=parsed_json["data"], context={"changed_by_username": username}, many=True)
if list_serializer.is_valid():
model_class = serializer_class.Meta.model
for data in reversed(list_serializer.validated_data):
if model_class.equal_to_current(data):
list_serializer.validated_data.remove(data)
entries_created = len(list_serializer.validated_data)
list_serializer.save()
return entries_created
else:
raise Exception(list_serializer.error_messages)
