Last active
May 5, 2023 14:04
-
-
Save toransahu/221371c981c20f0b9c645019a53b90c7 to your computer and use it in GitHub Desktop.
Writable Nested Serializers | Multiple Images Upload | DRF
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
#!/usr/bin/env python3.6.4 | |
# coding="UTF-8" | |
__author__ = 'Toran Sahu <[email protected]>' | |
__copyright__ = 'Copyright (C) 2018 Ethereal Machines Pvt. Ltd. All rights reserved' | |
from django.db import models | |
from os import path | |
from utils import directory_path_with_id | |
from django.utils.text import slugify | |
from django.db.models import Lookup, Transform | |
from django.db.models.fields import Field, CharField | |
# Create your models here. | |
class Image(models.Model): | |
# image = models.CharField(blank=True, null=True, max_length=50) | |
image = models.ImageField(blank=True, null=True, upload_to=directory_path_with_id) | |
class Post(models.Model): | |
title = models.CharField(max_length=50, unique=True) | |
content = models.TextField() | |
created_at = models.DateTimeField(auto_now_add=True) | |
updated_at = models.DateTimeField(auto_now=True) | |
thumbnail = models.ImageField(blank=True, null=True, upload_to=directory_path_with_id) | |
images = models.ManyToManyField(Image, related_name='posts') | |
def __str__(self): | |
return self.title | |
def get_slugged_title(self): | |
return slugify(self.title) |
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
#!/usr/bin/env python3.6.4 | |
# coding="UTF-8" | |
__author__ = 'Toran Sahu <[email protected]>' | |
__copyright__ = 'Copyright (C) 2018 Ethereal Machines Pvt. Ltd. All rights reserved' | |
from rest_framework import serializers | |
from .models import Post, Image | |
class ImageSerializer(serializers.ModelSerializer): | |
class Meta: | |
model = Image | |
fields = '__all__' | |
class PostSerializer(serializers.ModelSerializer): | |
images = ImageSerializer(many=True) | |
class Meta: | |
model = Post | |
fields = '__all__' | |
def create(self, validated_data): | |
""" | |
Handle writable nested serializer to create a new post. | |
:param validated_data: validated data, by serializer class's validate method | |
:return: updated Post model instance | |
""" | |
# TODO: Handle the case to avoid new Post instance creation if Image model data have any errors | |
data = validated_data.copy() | |
data.pop('images') # deleting 'images' list as it is not going to be used | |
''' | |
Fetching `images` list of image files explicitly from context. | |
Because using default way, value of `images` received at serializers from viewset was an empty list. | |
However value of `images` in viewset were OK. | |
Hence applied this workaround. | |
''' | |
images_data = self.context.get('request').data.pop('images') | |
try: | |
post = Post.objects.create(**data) | |
except TypeError: | |
msg = ( | |
'Got a `TypeError` when calling `Post.objects.create()`.' | |
) | |
raise TypeError(msg) | |
try: | |
for image_data in images_data: | |
# Image.objects.create(post=post, **image_data) | |
image, created = Image.objects.get_or_create(image=image_data) | |
post.images.add(image) | |
return post | |
except TypeError: | |
post = Post.objects.get(pk=post.id) | |
post.delete() | |
msg = ( | |
'Got a `TypeError` when calling `Image.objects.get_or_create()`.' | |
) | |
raise TypeError(msg) | |
return post | |
def update(self, instance, validated_data): | |
""" | |
Handle writable nested serializer to update the current post. | |
:param instance: current Post model instance | |
:param validated_data: validated data, by serializer class's validate method | |
:return: updated Post model instance | |
""" | |
# TODO: change the definition to make it work same as create() | |
''' | |
overwrite post instance fields with new data if not None, else assign the old value | |
''' | |
instance.title = validated_data.get('title', instance.title) | |
instance.content = validated_data.get('content', instance.content) | |
instance.thumbnail = validated_data.get('thumbnail', instance.thumbnail) | |
# instance.updated_at = validated_data.get('updated_at', instance.updated_at) # no need to update; auto_now; | |
try: | |
''' | |
Fetching `images` list of image files explicitly from context. | |
Because using default way, value of `images` received at serializers from viewset was an empty list. | |
However value of `images` in viewset were OK. | |
Hence applied this workaround. | |
''' | |
images_data = self.context.get('request').data.pop('images') | |
except: | |
images_data = None | |
if images_data is not None: | |
image_instance_list = [] | |
for image_data in images_data: | |
image, created = Image.objects.get_or_create(image=image_data) | |
image_instance_list.append(image) | |
instance.images.set(image_instance_list) | |
instance.save() # why? see base class code; need to save() to make auto_now work | |
return instance | |
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
import requests | |
import json | |
url = 'http://127.0.0.1:8000/api/v1/posts/' | |
patch_url = 'http://127.0.0.1:8000/api/v1/posts/2/' | |
# url = 'http://192.168.1.106:8000/api/v1/posts/' | |
with open('/home/toran/Desktop/FB_IMG_1475551548845.jpeg', 'rb') as f: | |
files = {'thumbnail': ('fb.jpeg', f)} | |
f_aws = open('/home/toran/Pictures/aws-https.png', 'rb') | |
f_ami = open('/home/toran/Pictures/ami.png', 'rb') | |
files = {'thumbnail': ('aws.png', f_aws)} | |
multiple_files = [ | |
('images', ('foo.png', f)), | |
('images', ('bar.png', f_ami)), | |
('thumbnail', ('aws.png', f_aws)), | |
] | |
data = dict() | |
data['title'] = '2' | |
data['content'] = 'something' | |
patch_data = dict() | |
patch_data['content'] = 'something else' | |
# res = requests.post(url, data, files=files, headers={'content_type': 'multipart/form-data'}) | |
# res = requests.post(url, data=data, files=multiple_files, headers={'content_type': 'multipart/form-data'}) | |
res = requests.patch(patch_url, data=patch_data, files=multiple_files, headers={'content_type': 'multipart/form-data'}) | |
print(res.text) | |
# blog_url = 'http://127.0.0.1:8000/api/v1/blogs/1/' | |
# files = {'thumbnail': ('aws.png', open('/home/toran/Pictures/aws-https.png', 'rb'))} | |
# res = requests.patch(blog_url, files=files, headers={'content_type': 'multipart/form-data'}) | |
# print(res.text) |
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
#!/usr/bin/env python3.6.4 | |
# coding="UTF-8" | |
__author__ = 'Toran Sahu <[email protected]>' | |
__copyright__ = 'Copyright (C) 2018 Ethereal Machines Pvt. Ltd. All rights reserved' | |
from rest_framework import viewsets, mixins, status | |
from rest_framework.settings import api_settings | |
from .models import Image, Post | |
from .serializers import PostSerializer, ImageSerializer | |
from rest_framework.parsers import MultiPartParser, FormParser | |
from backend.permissions import IsAuthenticatedOrReadOnly | |
from rest_framework.response import Response | |
# Create your views here. | |
class PostViewSet(viewsets.ModelViewSet): | |
""" | |
Post ModelViewSet. | |
""" | |
queryset = Post.objects.all() | |
serializer_class = PostSerializer | |
parser_classes = (MultiPartParser, FormParser) | |
# permission_classes = (IsAuthenticatedOrReadOnly,) | |
def create(self, request, *args, **kwargs): | |
print(request.data) | |
serializer = self.get_serializer(data=request.data) | |
serializer.is_valid(raise_exception=True) | |
self.perform_create(serializer) | |
headers = self.get_success_headers(serializer.data) | |
return Response(serializer.data, status=status.HTTP_201_CREATED, headers=headers) | |
def perform_create(self, serializer): | |
serializer.save() | |
def get_success_headers(self, data): | |
try: | |
return {'Location': str(data[api_settings.URL_FIELD_NAME])} | |
except (TypeError, KeyError): | |
return {} |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
same issue