[Project] Threepark
[Capstone Design] 3. 백엔드 구현 - (7) DRF 개발 | VIEW
mingyung
2024. 5. 21. 15:14
편리한 API의 구성을 위해서 Mixins와 GenericViewSet을 사용한다.
기본적으로 아래와 같이 사용하게 된다.
class DiaryViewSet(GenericViewSet, # 믹스인 사용을 위해 꼭 추가
mixins.ListModelMixin,#리스트 API
mixins.CreateModelMixin,#생성 API
mixins.RetrieveModelMixin,#조회 API
mixins.UpdateModelMixin,#수정 API. 부분 수정과 전체 수정 있음
mixins.DestroyModelMixin):#삭제 API
# 아래 퍼미션~쿼리셋은 필수 작성
permission_classes = [IsOwner]
serializer_class = DiarySerializer
queryset = Diary.objects.all()
"""
여기에 적는 주석은 후에 swagger API문서를 위한 것. 어떤 뷰인지 작성.
예를 들면 다음과 같이 작성.
일기의 내용에 대한 API
"""
# 이 부분은 믹스인을 사용할 때 쿼리를 필터링하여 본인의 데이터만 볼 수 있도록 처리한것
def filter_queryset(self,queryset):
queryset = queryset.filter(user=self.request.user)
return super().filter_queryset(queryset)
# 이 아래로 부터 추가로 커스텀이 필요한 믹스인들, 함수들을 작성한다.
# 아래부분없이는 기본 믹스인에서 제공하는 기능을 사용한다
기본 믹스인의 함수들은 아래 깃헙에서 확인할 수 있다.
https://github.com/encode/django-rest-framework/blob/master/rest_framework/mixins.py
django-rest-framework/rest_framework/mixins.py at master · encode/django-rest-framework
Web APIs for Django. 🎸. Contribute to encode/django-rest-framework development by creating an account on GitHub.
github.com
역시 뷰가 엄청나게 많기 때문에 특수한 경우만 살펴보자.
Follow
팔로잉 기능을 기본 mixins들의 각 기능들을 활용하여 구현하였다. (약간 야매로 구현한 느낌.. 하지만 잘 돌아간다면?)
1. permissions의 IsOwnerOrReadOnly를 통해 팔로잉에 연관된 사용자만 수정할 수 있도록 한다.
2. filter_queryset을 통해 연관된 사용자만 접근할 수 있게 한다.
개별 함수별 역할
CREATE : 팔로우 요청
DESTROY: 팔로우 취소/삭제
UPDATE: 팔로우 허용
PARTIAL_UPDATE: 팔로우 거절
class FollowViewSet(GenericViewSet,
mixins.ListModelMixin,
mixins.CreateModelMixin,
mixins.DestroyModelMixin,
mixins.RetrieveModelMixin,
mixins.UpdateModelMixin):
permission_classes = [IsOwnerOrReadOnly]
serializer_class = FollowSerializer
queryset = Follow.objects.all()
def filter_queryset(self,queryset):
queryset = queryset.filter(Q(follower=self.request.user) | Q(following_user=self.request.user))
return super().filter_queryset(queryset)
'''
팔로우 API
---
### id : 팔로우 요청의 id
'''
@swagger_auto_schema( request_body=openapi.Schema(
type=openapi.TYPE_OBJECT,
properties={
'username': openapi.Schema(type=openapi.TYPE_STRING, description='팔로우 요청할 유저의 username')
}
))
def create(self, request, *args, **kwargs):
'''
팔로우 요청하는 API
---
### id : 팔로우 요청할 username
## 예시 request:
{
"username": "threepark"
}
## 예시 response:
201
{
"id": 11,
"status": "requested",
"follower": 3,
"following_user": 1
}
400
'''
# 클라이언트로부터 사용자 이름을 받음
username = request.data.get('username')
# 받은 사용자 이름을 사용하여 사용자를 찾음
try:
following_user = User.objects.get(username=username)
except User.DoesNotExist:
return Response({"message": f"User '{username}' does not exist"}, status=status.HTTP_404_NOT_FOUND)
# 팔로우 요청 생성에 사용할 데이터 구성
request_data = {
'follower': request.user.id,
'following_user': following_user.id,
'status': Follow.REQUESTED
}
serializer = self.get_serializer(data=request_data)
serializer.is_valid(raise_exception=True)
user = self.request.user
followee = serializer.validated_data.get('following_user')
if followee==user:
return Response({"message": f"Cannot Follow yourself, {followee.username}."}, status=status.HTTP_400_BAD_REQUEST)
if Follow.objects.filter(follower=user, following_user=followee).exists() | Follow.objects.filter(follower=user, following_user=followee).exists():
return Response({"message": f"Follow request already sent to {followee.username}."}, status=status.HTTP_400_BAD_REQUEST)
follow_request, created = Follow.objects.get_or_create(follower=request.user, following_user=followee, status=Follow.REQUESTED)
serializer = self.get_serializer(follow_request)
if created:
return Response(serializer.data, status=status.HTTP_201_CREATED)
else:
return Response({"message": f"Follow request already sent to {followee.username}"}, status=status.HTTP_400_BAD_REQUEST)
def destroy(self, request, *args, **kwargs):
'''
팔로우 요청 삭제/취소하는 API
---
### id : 팔로우 요청의 id
## 예시 response:
204
{"message": "Follow request deleted"}
'''
instance = self.get_object()
self.perform_destroy(instance)
return Response({"message": "Follow request deleted"}, status=status.HTTP_204_NO_CONTENT)
@swagger_auto_schema(
request_body=openapi.Schema(
type=openapi.TYPE_OBJECT,
properties={}
))
def update(self, request, *args, **kwargs):
'''
팔로우 요청 허용하는 API
---
### id : 팔로우 요청의 id
## 예시 response:
200
{
"id": 11,
"status": "accepted",
"follower": 3,
"following_user": 1
}
401 권한이 없습니다
'''
instance = self.get_object()
# 요청받은 사용자가 현재 로그인한 사용자와 일치하는지 확인
if instance.following_user != request.user:
raise PermissionDenied("권한이 없습니다")
instance.status = Follow.ACCEPTED
instance.save()
serializer = self.get_serializer(instance)
return Response(serializer.data, status=status.HTTP_200_OK)
@swagger_auto_schema(
request_body=openapi.Schema(
type=openapi.TYPE_OBJECT,
properties={}
))
def partial_update(self, request, *args, **kwargs):
'''
팔로우 요청 거절하는 API
---
### id : 팔로우 요청의 id
## 예시 response:
200
{
"id": 9,
"status": "rejected",
"follower": 2,
"following_user": 1
}
'''
instance = self.get_object()
instance.status = Follow.REJECTED
instance.save()
serializer = self.get_serializer(instance)
return Response(serializer.data, status=status.HTTP_200_OK)
AI 작업이 필요한 Viewset들
이 아래는 API요청이 들어가는 외부 함수를 통해 AI의 작업이 있는 부분에 대한 View이다.
동작 구조는 다음과 같다.
Music
views 상단에 AI 서버로 응악 추천 결과를 받는 함수를 작성한다.
def request_music_from_flask(content):
"""
diary content 를 ai서버에 전달, 음악 추천 받아옴
"""
flask_url = f'http://{settings.FLASK_URL}:5000/get_music'
try:
response = requests.post(flask_url, json={'content': content},verify=False, timeout=50)
if response.status_code == 200:
response_data = response.json()
time.sleep(2)
return response_data
else:
print("Failed to get music from Flask:", response.status_code)
return None
except Exception as e:
print("Error:", e)
time.sleep(10)
return None
이를 바탕으로 음악 객체를 저장하고, 일기에 음악 데이터를 연결한다.
def update(self, request,*args, **kwargs):
"""
diary_music_update 일기에 대해 음악을 추천하는 API
---
### id = 일기 ID
최대 15초 소요 가능
### 예시 request:
{
"user": 1,
}
### 예시 response:
200
{
"id": 1,
"user": 1,
"content": "너무 두근거린다! 과연 rds에 내 다이어리가 잘 올라갈까? 오늘 이것만 성공하면 너무 즐거운 마음으로 잘 수 있을것 같다!",
"music": {
"id": 1,
"music_title": "그대만 있다면 (여름날 우리 X 너드커넥션 (Nerd Connection))",
"artist": "너드커넥션 (Nerd Connection)",
"genre": "발라드"
}
}
401
400
{'detail': 'Failed to get similar music from Flask'}
"""
partial = kwargs.pop('partial', True)
instance = self.get_object()
serializer = self.get_serializer(instance, data=request.data, partial=partial)
serializer.is_valid(raise_exception=True)
print(serializer.data['content'])
response = request_music_from_flask(serializer.data['content'])
best_music = response.get('most_similar_song')
print(best_music)
similar_songs = response.get('similar_songs')
print(similar_songs)
if best_music:
music, created = Music.objects.get_or_create(music_title=best_music['title'], artist=best_music['artist'], genre=best_music['genre'])
instance.music = music
serializer = self.get_serializer(instance, data=request.data, partial=partial)
serializer.is_valid(raise_exception=True)
self.perform_update(serializer)
return Response(serializer.data, status=status.HTTP_200_OK)
else:
return Response({'detail': 'Failed to get similar music from Flask'}, status=status.HTTP_400_BAD_REQUEST)
Emotion&Chat
views 상단에 AI 서버로 emotion label 검출과 응원 문구(comment) 를 생성하도록 요청하는 함수를 작성한다.
#views.py의 상단에 AI서버로 요청을하고 응답을 받는 함수 추가
def request_emotion(content):
"""
일기 내용으로 emootion label 검출
"""
flask_url = f'http://{settings.FLASK_URL}:5000/get_sentiment'
try:
response = requests.post(flask_url, json={'content': content},verify=False, timeout=50)
if response.status_code == 200:
response_data = response.json()
emotion_label = response_data['emotion_label']
print("Received emotion_label:", emotion_label)
time.sleep(2)
return emotion_label
else:
print("Failed to get emotion from Flask:", response.status_code)
return None
except Exception as e:
print("Error:", e)
time.sleep(10)
return None
def request_comment(content):
"""
일기 내용으로 응원 문구 생성
"""
flask_url = f'http://{settings.FLASK_URL}:5000/get_comment'
try:
response = requests.post(flask_url, json={'content': content},verify=False, timeout=50)
if response.status_code == 200:
response_data = response.json()
comment = response_data['comment']
print("Received comment:", comment)
time.sleep(2)
return comment
else:
print("Failed to get comment from Flask:", response.status_code)
return None
except Exception as e:
print("Error:", e)
time.sleep(10)
return None
이를 바탕으로 ViewSet내부에 함수를 생성한다.
#ViewSet내부에 create 작성
def create(self, request, *args, **kwargs):
"""
emotion_create 일기 내용으로 감정라벨, 응원문구 생성 하는 API
---
## 예시 request:
{
'diary' : 2
}
## 예시 response:
200
{
"id": 2,
"emotion_label": "불안",
"emotion_prompt": "",
"chat": " 이별은 사실일지도 모르겠어요 ",
"diary": 2
}
"""
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
diary = serializer.validated_data.get('diary')
if diary.user != request.user:
return Response({'error': "Diary does not belong to the current user."}, status=status.HTTP_400_BAD_REQUEST)
chat = request_comment(diary.content)
label = request_emotion(diary.content)
existing_emotion = Emotion.objects.filter(diary=diary).first()
if existing_emotion:
serializer = self.get_serializer(existing_emotion, data={'chat': chat, 'emotion_label': label}, partial=True)
serializer.is_valid(raise_exception=True)
serializer.save(diary=diary, chat=chat, emotion_label = label)
else:
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
serializer.save(diary=diary, chat=chat, emotion_label = label)
return Response(serializer.data, status=status.HTTP_201_CREATED)
Image
GPT - 이미지 생성용 프롬프트 생성
GPT API와 미리 정의한 GPT 프롬프트를 합하여 GPT에 이미지 생성용 프롬프트를 작성한다.
from django.conf import settings
import openai
with open(f"{settings.BASE_DIR}/ai/genTextBase.txt", 'r', encoding='utf-8') as file:
base_text = ''.join(file.readlines())
api_key = settings.OPENAI_API_KEY
openai.api_key=api_key
def get_prompt(content):
"""
일기 내용을 입력받아 프롬프트 생성
"""
completion = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": f"{base_text} {content.strip()}"},]
)
generated_text = completion["choices"][0]["message"]["content"]
output_text=generated_text.split('\n')
pre_text = "(masterpiece,detailed), (Oil Painting:1.3), (Impressionism:1.3) ,(oil painting with brush strokes:1.2), (looking away:1.1), "
prompts = [pre_text+v for v in output_text if v]
return prompts
Flask&AI서버 - 이미지 생성 요청
이미지 프롬프트를 전달해 AI서버에서 이미지를 생성하고, 이미지 url을 응답으로 받는다.
def request_image_from_flask(prompt):
"""
생성된 prompt로 이미지 생성
"""
flask_url = f'http://{settings.FLASK_URL}:5000/get_image'
try:
# HTTP POST 요청으로 prompt를 Flask에 전송
response = requests.post(flask_url, json={'prompt': prompt},verify=False, timeout=150)
# 응답 확인
if response.status_code == 200:
# 이미지 생성 성공
response_data = response.json()
image_url = response_data['image_url']
print("Received image url:", image_url)
time.sleep(2)
return image_url
else:
# 이미지 생성 실패
print("Failed to get image from Flask:", response.status_code)
return None
except Exception as e:
print("Error:", e)
time.sleep(10)
return None
Viewset - create함수 작성
위에서 작성한 함수들을 통해 이미지를 생성하고, 전달받은 url을 통해 이미지 시리얼라이저 생성
def create(self, request, *args, **kwargs):
'''
이미지 생성 API
---
### 응답에 최대 40초 소요 가능
## 예시 request:
{
'diary' : 1
}
## 예시 response:
201
{
"id": 70,
"created_at": "2024-05-02T13:04:10.208658+09:00",
"image_url": "https://버킷주소/images/826cb58e-46a3-41fc-9699-bc2eccdc1355.jpg",
"image_prompt": "(masterpiece,detailed), (Oil Painting:1.3), (Impressionism:1.3) ,(oil painting with brush strokes:1.2), (looking away:1.1), a girl in a traditional Korean hanbok, cherry blossom background, soft pastel colors, Korean artist reference, (ethereal:1.2), (delicate details:1.3), (dreamy atmosphere:1.25)",
"diary": 1
}
400
{
'error': "Failed to get image from Flask" 이 경우 AI 서버가 꺼져있을때임
}
400
{
'error': "Error uploading image: {str(e)}"
}
401
403
'''
try:
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
diary = serializer.validated_data.get('diary')
image_prompt = get_prompt(diary.content)[0]
image_url = request_image_from_flask(image_prompt)
if not image_url:
return Response({'error': "Failed to get image from Flask"}, status=status.HTTP_400_BAD_REQUEST)
new_image = Image.objects.get_or_create(diary=diary, image_url=image_url, image_prompt=image_prompt)
serializer.validated_data['diary'] = diary
serializer.validated_data['image_url'] = image_url
serializer.validated_data['image_prompt'] = image_prompt
serializer.save()
return Response(serializer.data, status=status.HTTP_201_CREATED)
except Exception as e:
return Response({'error': f"Error uploading image: {str(e)}"}, status=status.HTTP_400_BAD_REQUEST)