こんにちは。KOUKIです。
前回は、Blog APIを作成しました。
今回は、Media/Filter APIを作成します。
<目次>
記事まとめ
STATIC設定
settings.pyに下記の追加をします。
|
1 2 3 4 5 6 7 |
STATIC_URL = '/static/' MEDIA_URL = '/media/' # See Dockerfile MEDIA_ROOT = '/vol/web/media' STATIC_ROOT = '/vol/web/static' |
STATIC系の設定は、JavaScriptやCSSなどの静的ファイルの設定、MEDIA系は画像ファイルの設定をそれぞれ扱います。
例えば、STATIC_URLは、ブラウザを介してstaticディレクトリにアクセスするための参照先を記述します。
|
1 |
例) https://<SERVER>/static/ |
STATIC_ROOTは、staticファイルがサーバー上のどこに保存されるのかについて記述します。保存先はDockerfileにて作成していますので、そのパスを参照しています。
|
1 2 3 4 |
# Dockerfile # staticファイルの保存先を作成 RUN mkdir -p /vol/web/media RUN mkdir -p /vol/web/static |
続いて、appディレクトリ配下のurls.pyを修正します。
|
1 2 3 4 5 6 7 8 9 10 11 12 |
# app/app/urls.py from django.contrib import admin from django.urls import include, path from django.conf.urls.static import static # new from django.conf import settings # new urlpatterns = [ path('admin/', admin.site.urls), path('api/account/', include('account.urls')), path('api/blog/', include('blog.urls')), ] + static(settings.MEDIA_URL, document_root=settings.MEDIA_ROOT) # new |
urls.pyに画像ファイルへのパスを追加しました。
imageフィールドの追加とテスト
Media APIは、画像をアップロード機能を提供します。
そのため、先ほど作成したMEDIA設定にて、画像が取り扱えるようにしました。
既存のモデルにimageフィールドを追加したいと思います。
最初にテストコードを書きます。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 |
# app/blog/tests/test_models.py from unittest.mock import patch class ModelTests(TestCase): ... @patch('uuid.uuid4') def test_blog_file_name_uuid(self, mock_uuid): """Check if image is saved in the correct location""" uuid = 'test-uuid' mock_uuid.return_value = uuid file_path = blog_image_file_path(None, 'blogimage.jpg') exp_path = f'uploads/blog/{uuid}.jpg' self.assertEqual(file_path, exp_path) |
画像をアップロードしたら一意のID(uuid)を画像に付与する予定です。
|
1 2 |
例) '314a7403-cc52-4614-b0d2-8437dad36d16' |
uuidは、pythonの組み込みモジュールであるuuidを使って生成します。
これにより、画像を一意に識別できるわけです。
また、テスト時には実際にファイルをアップロードするわけにはいかないので、mockを利用しています。
「@patch(‘uuid.uuid4’)」では、mockしたい関数を指定しています。そして、「mock_uuid.return_value = uuid」にて戻り値を明示的に指定しています。
これでuuid.uuid4の実行結果が、常に「test-uuid」になります。
テストを実行します。
|
1 2 3 |
make test app=blog E NameError: name 'blog_image_file_path' is not defined |
「blog_image_file_path」が存在しないとエラーがでました。
blogアプリのmodels.pyに次の処理を追加します。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 |
# app/blog/models.py import uuid import os from django.db import models from django.conf import settings # new def blog_image_file_path(instance, filename): ext = filename.split('.')[-1] filename = f'{uuid.uuid4()}.{ext}' return os.path.join('uploads/blog/', filename) def set_user(): return models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.CASCADE, ) class Tag(models.Model): name = models.CharField(max_length=255) user = set_user() def __str__(self): return self.name class Comment(models.Model): comment = models.TextField() user = set_user() def __str__(self): return self.comment class Blog(models.Model): user = set_user() title = models.CharField(max_length=255) body = models.TextField() link = models.CharField(max_length=255, blank=True) comments = models.ManyToManyField('Comment') tags = models.ManyToManyField('Tag') image = models.ImageField(null=True, upload_to=blog_image_file_path) # new def __str__(self): return self.title |
マイグレーションを実行しましょう。
|
1 |
make migrate |
blog/tests/test_models.pyにblog_image_file_path関数をインポートします。
|
1 |
from blog.models import Tag, Comment, Blog, blog_image_file_path |
テストを実行しましょう。
|
1 2 3 |
make test app=blog == 23 passed in 4.07s == |
Upload imageテスト
画像アップロードテストを行います。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 |
# app/blog/tests/test_blog_api.py import tempfile # new import os # new from PIL import Image # new from django.contrib.auth import get_user_model from django.test import TestCase from django.urls import reverse from rest_framework import status from rest_framework.test import APIClient from blog.models import Blog, Comment, Tag from blog.serializers import BlogSerializer, BlogDetailSerializer BLOGS_URL = reverse('blog:blog-list') def image_upload_url(blog_id): return reverse('blog:blog-upload-image', args=[blog_id]) ... class BlogImageUploadTests(TestCase): def setUp(self): self.client = APIClient() self.user = get_user_model().objects.create_user( 'test@selfnote.work', 'SELF NOTE', ) self.client.force_authenticate(self.user) self.blog = sample_blog(user=self.user) def tearDown(self): self.blog.image.delete() def test_upload_image_to_blog(self): """Checkif uploading an image to blog""" url = image_upload_url(self.blog.id) with tempfile.NamedTemporaryFile(suffix='.jpg') as ntf: img = Image.new('RGB', (10, 10)) img.save(ntf, format='JPEG') ntf.seek(0) res = self.client.post(url, {'image': ntf}, format='multipart') self.blog.refresh_from_db() self.assertEqual(res.status_code, status.HTTP_200_OK) self.assertIn('image', res.data) self.assertTrue(os.path.exists(self.blog.image.path)) def test_upload_image_bad_request(self): """Check if uploading an invalid image""" url = image_upload_url(self.blog.id) res = self.client.post(url, {'image': 'nothingimage'}, format='multipart') self.assertEqual(res.status_code, status.HTTP_400_BAD_REQUEST) ... |
正しい引数とともにファイルをアップロードし、成功したら200, 失敗したら400のエラーメッセージを返す仕様にしています。
tempfileモジュールをインポートしましたが、これは便利です。
一時ファイルを作成してくれるモジュールで、プログラム実行後に自動で作成したファイルを削除してくれます。
ここでは、PILモジュールのImage関数を使ってJPEGを作成しています。
テストを実行しましょう。
|
1 2 3 |
make test app=blog E django.urls.exceptions.NoReverseMatch: Reverse for 'blog-upload-image' not found. 'blog-upload-image' is not a valid view function or pattern name. |
「blog-upload-image」がないと怒られましたね。
views.pyのBlogViewSetにupload_image関数を追加すると解決できます。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 |
# app/blog/views.py from rest_framework.decorators import action from rest_framework.response import Response from rest_framework import viewsets, mixins, status from rest_framework.authentication import TokenAuthentication from rest_framework.permissions import IsAuthenticated from blog.models import Tag, Comment, Blog from .serializers import (TagSerializer, CommentSerializer, BlogSerializer, BlogDetailSerializer, BlogImageSerializer) class BaseBlogAttrViewSet( ... class TagViewSet(BaseBlogAttrViewSet): ... class CommentViewSet(BaseBlogAttrViewSet): ... class BlogViewSet(viewsets.ModelViewSet): serializer_class = BlogSerializer queryset = Blog.objects.all() authentication_classes = (TokenAuthentication,) permission_classes = (IsAuthenticated,) def get_queryset(self): return self.queryset.filter(user=self.request.user) def get_serializer_class(self): if self.action == 'retrieve': return BlogDetailSerializer elif self.action == 'upload_image': # new return BlogImageSerializer return self.serializer_class def perform_create(self, serializer): serializer.save(user=self.request.user) # new @action(methods=['POST'], detail=True, url_path='upload-image') def upload_image(self, request, pk=None): """Upload an image to a blog""" blog = self.get_object() serializer = self.get_serializer( blog, data=request.data ) if serializer.is_valid(): serializer.save() return Response( serializer.data, status=status.HTTP_200_OK ) return Response( serializer.errors, status=status.HTTP_400_BAD_REQUEST ) |
テストを実行します。
|
1 |
E ImportError: cannot import name 'BlogImageSerializer' from 'blog.serializers' (/app/blog/serializers.py) |
シリアライズクラスを追加します。
|
1 2 3 4 5 6 7 |
# app/blog/serializers.py class BlogImageSerializer(serializers.ModelSerializer): class Meta: model = Blog fields = ('id', 'image') read_only_fields = ('id',) |
テストを実行します。
|
1 2 3 |
make test app=blog === 25 passed in 4.25s === |
テストがパスしました。
Filter機能とテスト
Filter APIは、タグやコメントを使ったフィルタリング機能を提供します。
まずはテストコードからです。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 |
# app/blog/tests/test_blog_api.py class PrivateBlogApiTests(TestCase): ... def test_filter_blos_by_tags(self): """Check if returning blogs with specific tags""" blog1 = sample_blog(user=self.user, title='Sample Blog 1') blog2 = sample_blog(user=self.user, title='Sample Blog 2') tag1 = sample_tag(user=self.user, name='Programming') tag2 = sample_tag(user=self.user, name='Blog') blog1.tags.add(tag1) blog2.tags.add(tag2) blog3 = sample_blog(user=self.user, title='Sample Blog 3') res = self.client.get( BLOGS_URL, {'tags': f'{tag1.id},{tag2.id}'}) serializer1 = BlogSerializer(blog1) serializer2 = BlogSerializer(blog2) serializer3 = BlogSerializer(blog3) self.assertIn(serializer1.data, res.data) self.assertIn(serializer2.data, res.data) self.assertNotIn(serializer3.data, res.data) def test_filter_blogs_by_comments(self): """Check if returning blogs with specific comments""" blog1 = sample_blog(user=self.user, title='Sample Blog 1') blog2 = sample_blog(user=self.user, title='Sample Blog 2') comment1 = sample_comment(user=self.user, comment='Comment Test 1') comment2 = sample_comment(user=self.user, comment='Comment Test 2') blog1.comments.add(comment1) blog2.comments.add(comment2) blog3 = sample_blog(user=self.user, title='Sample Blog 3') res = self.client.get( BLOGS_URL, {'comments': f'{comment1.id},{comment2.id}'} ) serializer1 = BlogSerializer(blog1) serializer2 = BlogSerializer(blog2) serializer3 = BlogSerializer(blog3) self.assertIn(serializer1.data, res.data) self.assertIn(serializer2.data, res.data) self.assertNotIn(serializer3.data, res.data) |
テストを実行します。
|
1 2 3 |
make test app=blog === 2 failed, 25 passed in 4.69s === |
今は、タグやコメントでフィルタをかけていないので、当然エラーがでます。
views.pyに次の処理を追加します。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 |
# app/blog/views.py class BlogViewSet(viewsets.ModelViewSet): serializer_class = BlogSerializer queryset = Blog.objects.all() authentication_classes = (TokenAuthentication,) permission_classes = (IsAuthenticated,) def _params_to_inits(self, qs): """ Convert a list of string IDs to a list of integers our_string = '1,2,3' out_string_list = ['1', '2', '3'] """ return [int(str_id) for str_id in qs.split(',')] def get_queryset(self): tags = self.request.query_params.get('tags') comments = self.request.query_params.get('comments') queryset = self.queryset if tags: tag_ids = self._params_to_inits(tags) queryset = queryset.filter(tags__id__in=tag_ids) if comments: comment_ids = self._params_to_inits(comments) queryset = queryset.filter(comments__id__in=comment_ids) return queryset.filter(user=self.request.user) |
テストを実行します。
|
1 2 3 |
make test app=blog == 27 passed in 4.48s == |
test_tags_api.py及びtest_comments_api.pyにもそれぞれテストを追加します。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 |
# app/blog/tests/test_tags_api.py from blog.models import Tag, Blog ... class PrivateTagsApiTests(TestCase): ... def test_retrieve_tags_assigned_to_blogs(self): """Checkif filtering tags by those assigned to blogs""" tag1 = Tag.objects.create(user=self.user, name='Programming') tag2 = Tag.objects.create(user=self.user, name='Blog') blog = Blog.objects.create( title='Coriander eggs on toast', body='Happy!', user=self.user ) blog.tags.add(tag1) res = self.client.get(TAGS_URL, {'assigned_only': 1}) serializer1 = TagSerializer(tag1) serializer2 = TagSerializer(tag2) self.assertIn(serializer1.data, res.data) self.assertNotIn(serializer2.data, res.data) def test_retrieve_tags_assigned_unique(self): """Check if filtering tags by assigned unique items""" tag = Tag.objects.create(user=self.user, name='Programming') Tag.objects.create(user=self.user, name='Blog') blog1 = Blog.objects.create( title='Pancakes', body='Unhappy!', user=self.user ) blog1.tags.add(tag) blog2 = Blog.objects.create( title='Porridge', body='Unhappy!', user=self.user ) blog2.tags.add(tag) res = self.client.get(TAGS_URL, {'assigned_only': 1}) self.assertEqual(len(res.data), 1) |
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 |
# app/blog/tests/test_comments_api.py from blog.models import Comment, Blog ... class PrivateCommentAPITests(TestCase): ... def test_retrieve_comments_assigned_to_blogs(self): """Test filtering ingredients by those assigned to blogs""" comment1 = Comment.objects.create( user=self.user, comment='Apples' ) comment2 = Comment.objects.create( user=self.user, comment='Turkey' ) blog = Blog.objects.create( title='Apple crumble', body='Happy!', user=self.user ) blog.comments.add(comment1) res = self.client.get(COMMENTS_URL, {'assigned_only': 1}) serializer1 = CommentSerializer(comment1) serializer2 = CommentSerializer(comment2) self.assertIn(serializer1.data, res.data) self.assertNotIn(serializer2.data, res.data) def test_retrieve_comment_assigned_unique(self): """Test filtering comments by assigned returns unique items""" comment = Comment.objects.create(user=self.user, comment='taggs') Comment.objects.create(user=self.user, comment='Cheese') blog1 = Blog.objects.create( title='Eggs benedict', body='Unhappy!', user=self.user ) blog1.comments.add(comment) blog2 = Blog.objects.create( title='Coriander eggs on toast', body='Unhappy!', user=self.user ) blog2.comments.add(comment) res = self.client.get(COMMENTS_URL, {'assigned_only': 1}) self.assertEqual(len(res.data), 1) |
テストを実行します。
|
1 2 3 |
make test app=blog === 4 failed, 27 passed in 5.33s === |
当然エラーになりますよね。
blogアプリのviews.pyのTagViewSet及びCommentViewSetを修正します。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 |
class TagViewSet(BaseBlogAttrViewSet): queryset = Tag.objects.all() serializer_class = TagSerializer def get_queryset(self): assigned_only = bool( int(self.request.query_params.get('assigned_only', 0)) ) queryset = self.queryset if assigned_only: queryset = queryset.filter(blog__isnull=False) return queryset.filter( user=self.request.user).order_by('-name').distinct() class CommentViewSet(BaseBlogAttrViewSet): queryset = Comment.objects.all() serializer_class = CommentSerializer def get_queryset(self): assigned_only = bool( int(self.request.query_params.get('assigned_only', 0)) ) queryset = self.queryset if assigned_only: queryset = queryset.filter(blog__isnull=False) return queryset.filter( user=self.request.user).order_by('-comment').distinct() |
テストを実行します。
|
1 2 3 |
make test app=blog == 31 passed in 5.09s == |
テストがパスしましたね。
おわりに
Django REST Frameworkは、機能がたくさんあり使いこなすのが難しそうですね。
関連記事
基礎編はこちらです。
にほんブログ村
コメントを残す
コメントを投稿するにはログインしてください。