[Threepark] 백엔드 서버와 핵심 기능 구현

[Project] Threepark

[Threepark] 백엔드 서버와 핵심 기능 구현

mingyung 2024. 5. 21. 15:52

이번 포스팅은 백엔드 서버 구축과정과 핵심 기능 구현에 대해 알아본다.

개발 환경과 서버 구성하기

백엔드 개발을 시작하기 위해 필요한 데베 서버를 먼저 생성하도록 한다.

RDS를 이용해 데이터베이스를 관리하기로 하였고, MySQL을 사용한다.

데이터베이스

RDS생성

AWS에 회원가입을 완료한 상태로 시작한다.

AWS 콘솔의 서비스에서 RDS로 들어간다. 이때 상단의 region을 서울로 바꿔야 한다.

이제 데이터베이스 생성을 눌러 RDS인스턴스 생성을 시작한다.

데이터베이스 생성 과정

표준 생성을 선택한다.

우리는 MySQL을 사용할 예정이므로 MySQL선택.

바로 아래 엔진 버전을 확인하자

우리는 프리티어를 사용할 것이므로 프리티어 선택.

사용할 RDS 인스턴스의 이름을 입력하고, 마스터 사용자를 설정한다. 마스터 사용자 이름과 암호는 꼭 잊지말자!!

프리티어에서 스토리지는 20GiB까지 쓸 수 있다.

아래에서 퍼블릭 액세스를 '예'로 바꿔주자.

VPC보안그룹을 생성해야한다. 만약 기존 그룹이 있다면 전자를 선택할 수 있자. 만약 처음 사용하는거라면 새로 생성을 선택한다.

새로 생성을 눌렀다면 새 VPC 보안 그룹 이름에 이름을 정해준다.

초기 데이터베이스의 이름을 정해주자

이제 기타 조건들을 잘 읽어보고, 데이터베이스 생성을 선택해 생성 완료한다.

인바운드 규칙 편집

위의 과정을 통해서 RDS생성을 완료했다. 그러나 지금 이 상태로는 사용할 수 없다.

RDS에 접근하기 위해서는 인바운드 규칙을 설정해줘야 한다. 인바운드 규칙을 통해 허용된 IP주소, 포트에서 이 RDS에 접근이 가능하도록 한다.

먼저 대시보드 아래의 데이터베이스에서 생성한 데이터베이스를 볼 수 있다.

이렇게 연결 및 보안을 보면 VPC 보안 그룹이 표시 된다. 해당 보안그룹으로 들어가 인바운드 규칙을 편집한다.

인바운드 규칙 편집을 눌러 MySQL/Aurora, Port(3306)에 Anywhere-IPv4,Anywhere-IPv6을 모두 열어준다.

여기까지 완료하면 RDS세팅은 끝난다.

이제 MySQL을 RDS와 연결하여 데이터베이스 클라우드를 사용할 수 있도록 하는 방법에 대해 알아본다.

MySQL 다운로드

https://dev.mysql.com/downloads/installer/

[MySQL :: Download MySQL Installer

Note: MySQL 8.0 is the final series with MySQL Installer. As of MySQL 8.1, use a MySQL product's MSI or Zip archive for installation. MySQL Server 8.1 and higher also bundle MySQL Configurator, a tool that helps configure MySQL Server.

dev.mysql.com](https://dev.mysql.com/downloads/installer/)

MySQL을 다운받자

버전은 8.0으로 한다.

두개의 다운로드 버튼 중 두번째를 누른다.

이후 안내를 따라 MySQL을 설치하도록 한다.

MySQL과 RDS 연결

설치 완료 후 MySQL Workbench를 연다.

이제 MySQL Connections 옆의 +버튼을 누른다

그럼 이런 창이 뜨게된다.

먼저 Connection Name은 어떤 MySQL에 대한 연결인지 인지하기 위한 이름이므로 본인이 원하는 이름으로 작성한다.

Hostname에는 RDS의 엔드포인트 정보를 입력한다.

이전에 생성한 RDS에 들어가면 엔드포인트에 대한 정보가 있으니 복사하여 붙여넣는다.

Port에는 포트 번호를 입력한다.

기록해뒀던 username을 적고, store in vault를 눌러 password를 입력한다.

Test Connection를 누르고 아래 알람이 뜬다면 Ok를 누른다.

여기까지로 데이터베이스 환경 구성이 완료된다.

스토리지 서비스

구현하려는 서비스는 튜닝된 생성 모델을 활용하여 이미지를 생성하는 것이 주요 기능이기 때문에 이미지를 저장할 곳이 필요하다.

따라서 AWS의 S3를 사용하여 이미지를 저장하기로 했다. 프리티어 만세

이번 포스팅에서는 AWS의 S3버킷을 생성하는 방법에 대해 기록한다.

AWS S3 생성

RDS생성때와 같이 AWS 회원가입/로그인 상태로 시작한다. 리전이 서울로 되어있는것을 한번 더 확인하자.

AWS콘솔의 상단 네비게이션 바에서 서비스 클릭하여 S3를 찾자.

버킷 만들기 클릭

RDS생성 때 보다 간단하다.

버킷을 구별하기 위한 이름을 입력한다.

프론트나 백엔드에서 버킷에 접근하여 이미지를 얻기 위해서는 "모든 퍼블릭 액세스 차단"을 풀어야 한다.

대신 아래 사진 처럼 선택하고, 추후에 접근 가능한 액세스 지점을 정의할것이다.

암호화 설정을 확인하고 생성 완료 한다

완료 후 좌측 사이드바의 버킷으로 이동하면 생성된 버킷을 확인할 수 있다.

그러나 이 상태로는 사용할 수 없다.

외부에서 이 버킷에 접근해 이미지를 저장하고, 읽기 위해서는 접근 권한을 설정해줘야 하기 때문이다.

이번에는 CORS설정과 버킷 정책을 설정해야 한다

버킷 정책 설정

생성된 버킷으로 들어가서 버킷의 권한을 설정할 수 있다

권한을 클릭하면 정책을 편집할 수 있다. 정책 편집을 누른다.

그러면 버킷 ARN을 볼 수 있다. 이를 복사하고, 정책 생성기를 클릭한다.

1. Select Type of Policy 에서 S3 Bucket Policy를 선택한다.

2. Principal에 * 입력

3. Actions에 Get Object, Put Object 을 체크한다.

4. Amazon Resource Name (ARN) 에 위에서 복사한 ARN을 입력한 후 /* 입력 ex)arn:aws:s3:::ARN복사한것/*

5. Add Statement 클릭한다

생성을 완료하면, policy json document 가 나오는데, 이를 복사한다.

이전의 정책 편집 페이지로 돌아사서 정책한에 json구문을 붙여넣기 하고 저장한다.

CORS

버킷 정책 아래의 CORS를 설정한다.

편집을 눌러 아래를 붙여넣는다.

[
    {
        "AllowedHeaders": [
            "*"
        ],
        "AllowedMethods": [
            "HEAD",
            "GET",
            "PUT",
            "POST"
        ],
        "AllowedOrigins": [
            "*"
        ],
        "ExposeHeaders": []
    }
]

공개 설정

위의 과정으로 버킷의 정책에 대한 설정을 완료했지만, 그렇다고 해서 버킷의 이미지를 외부에서 로드할 수 있는것은 아니다.

이번에는 이미지가 저장된 images 폴더만 외부에서 이미지 url로 이미지를 볼 수 있도록 하자.

먼저 버킷으로 들어와서 폴더 만들기를 통해서 images 폴더를 만든다. 폰더가 아래처럼 생성되었다면 다음으로 넘어가자

폴더 왼쪽의 체크박스를 클릭하고, 위의 작업을 누르면 ACL을 사용하여 퍼블릭으로 설정할 수 있다.

이제 이 폴더의 이미지 파일을 외부에서 url을 통해 읽을 수 있게 되었다.

확인을 위해 이미지를 넣고,

url창에 이미지 url을 넣으면 다음처럼 나온다.

여기까지 하면 스토리지 까지 준비가 완료된다.

백엔드 - Django Rest Framework

이제는 Django - DRF와 Flask를 이용해서 백앤드 API와 생성형, 분류 모델을 연결하도록 한다.

백엔드 API 구현을 시작하도록 한다.

우리 서비스에서는 장고와 DRF를 사용하기로 하였다. DRF 개발을 위한 환경설정을 하는 과정에 대해 알아보자.

Django Rest Framework 환경 설정

DRF 설치

일단 파이썬 사용을 전제로 한다.

파이썬 버전은 3.10을 사용했다.

pip install django
pip install djangorestframework

프로젝트 생성

프로젝트 생성을 원하는 디렉토리에서 다음을 실행한다.

django-admin startproject 프로젝트이름 .

이를 완료하면 djnago 프로젝트가 생성된다.

app생성

개별 기능을 구현하는 app늘 만들기 위해서는 아래의 명령어를 실행한다.

python manage.py startapp APP이름

예) python manage.py startapp diary

Settings.py 설정

settings.py 에 들어가면 INSTALLED_APPS가 있다. 거기에 다름을 추가한다.

추가 후 전체 저장을 한 뒤 다음을 실행한다.

python manage.py migrate

개발용 로컬 서버 runserver

현재 개발에 대한 구현을 로컬서버를 통해 확인할 수 있다.

python manage.py runserver

Django Environ으로 환경변수 관리하기

프로젝트 생성은 끝났고, 이제 환경 변수를 관리하기 위한 .env파일을 구성해야 한다.

DRF를 이용해 서비스를 개발하기 위해서 관리해야 할 환경변수들을 Django Environ을 통해 관리할 수 있다.

특히 환경변수나 API키 등 외부로 유출되면 안되는 정보는 이를 통해 관리해야 한다.

설치

pip install django-environ

.env 생성

프로젝트의 루트에 .env 파일을 먼저 생성해준다.

생성 후에는 꼭 gitignore파일에 .env를 추가하여 깃허브에 올라가지 않도록 한다.

install django-environ

다음을 터미널에 입력하여 django-environ을 설치한다.

pip install django-environ

.env 작성

환경변수로 지정해야 하는값들을 env에 정의한다.

다음과 같이 작성해야 한다.

SECRET_KEY='django-insecure-...'
DEBUG=True
# MySQL DB
DB_NAME='localdb'
...

이때 주의할 점은 키와 값 사이에 띄어쓰기를 포함해서는 안된다는 것이다.

# 틀린 예시
DB_NAME = 'localdb'

# 올바른 예시
DB_NAME='localdb'

settings.py 작성

settings.py에 다음을 추가한다.

import environ
...
env = environ.Env(DEBUG=(bool, False))
environ.Env.read_env(os.path.join(BASE_DIR, '.env'))
...


# 이제는 환경변수가 들어갈 자리를 다음처럼 바꾸어 작성해준다.
SECRET_KEY = env('SECRET_KEY')
DEBUG = env('DEBUG')
...

migrate

settings.py작성을 마쳤다면 migrate를 진행한다.

python manage.py migrate

Django에 RDS 연결하기

이번에는 django 와 AWS RDS를 연결하는 방법에 대해서 알아본다.

설치

pip install boto3
pip install mysqlclient

Settings.py

프로젝트의 settings.py에 다음과 같이 작성한다.

개별 환경변수들은 이전 포스팅을 통해서 생성한 .env파일에 정의한다.

모두 저장하기를 한번 누르고,

migration을 진행한다.

python manage.py migrate

DRF 백엔드 구현

드디어 백엔드의 구현이 시작된다!! 모든 구현과정을 설명할 수는 없기 때문에 구현 상 특수한 부분과 핵심 기능부분을 포스팅한다.

과정은 다음과 같이 진행된다.

1. Model 만들기

2. Serializer 만들기

3. permissions 커스텀하여 접근 권한 만들기

4. views 작성하기

Authentication이 궁금하다면 아래 포스팅을 살펴볼 수 있다. (하지만 이 포스팅에서는 다루지 않는다)

https://he-kate1130.tistory.com/61

[[Team 22] DRF Authentication - dj-rest-auth

소셜 로그인 기능을 추후에 쉽게 추가하기 위해서 dj-rest-auth를 사용해 회원가입, 로그인 기능을 구현해보도록 하자. 사실 원래는 simple jwt를 활용해서 토큰을 발급하고 쿠키에 저장해두는 방식으

he-kate1130.tistory.com](https://he-kate1130.tistory.com/61)

DB & MODEL

이제 DB Schema를 구성하고, django의 Model을 작성하자

DB Schema 구성

DB 스키마를 다음과 같이 구성한다.

계획의 과정에서 다음과 같은 형태로 구상했다. (이미지 자체는 결과 이미지다)

- 유저 정보의 경우 django에서 제공하는 모델을 사용하기로 함

모델이 많은 편이므로 user, follow, diary, music, image, emotion의 경우만 이야기 해 본다.

User

유저모델은 django의 auth모델을 사용할 것이다. 따라서 코드는 다음과 같이 작성한다.

from django.db import models
from django.contrib.auth.models import AbstractUser, BaseUserManager

class UserManager(BaseUserManager):
    def create_user(self, email, username, password=None, **extra_fields):
        if not email:
            raise ValueError('The Email field must be set')
        email = self.normalize_email(email)
        user = self.model(email=email, username=username, **extra_fields)
        user.set_password(password)
        user.save(using=self._db)
        return user

    def create_superuser(self, email, username, password=None, **extra_fields):
        extra_fields.setdefault('is_staff', True)
        extra_fields.setdefault('is_superuser', True)

        return self.create_user(email, username, password, **extra_fields)

class User(AbstractUser):
    REQUIRED_FIELDS = ['email'] 
    objects = UserManager()

    class Meta:
        db_table = 'user'

    def get_by_natural_key(self, username):
        return self.get(username=username)

Follow

follow모델의 경우 유저 두명과 두명의 관계가 정의된 status값이 들어간다.

이때 유저 두명은 유니크한 쌍이 되어야 한다. (유저간에 여러개의 관계가 있을 수 없다.)

from django.db import models

class Follow(models.Model):
    REQUESTED = 'requested'
    ACCEPTED = 'accepted'
    REJECTED = 'rejected'
    STATUS_CHOICES = (
        (REQUESTED, 'Requested'),
        (ACCEPTED, 'Accepted'),
        (REJECTED, 'Rejected'),
    )

    follower = models.ForeignKey(User, related_name='following', on_delete=models.CASCADE)
    following_user = models.ForeignKey(User, related_name='followers', on_delete=models.CASCADE)
    status = models.CharField(max_length=20, choices=STATUS_CHOICES, default=REQUESTED)

    class Meta:
        unique_together = ('follower', 'following_user')
        db_table = 'follow'
        managed = True

Music

음악 정보를 저장하는 Music의 경우 다음과 같이 작성한다.

from django.db import models

class Music(models.Model):
    music_title = models.CharField(max_length=100, null=True)
    artist = models.CharField(max_length=100, null=True)
    genre = models.CharField(max_length=20, null=True)

    class Meta:
        managed = True
        db_table = 'music'

Diary

음악 정보와 유저를 fk로 가지는 diary의 경우 다음과 같다.

타이틀과 내용(content), 최초 생성 시간과 마지막 업데이트 시간을 저장하고, 팔로워에게 공개여부를 정할 수 있게 한다.

class Diary(models.Model):
    user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)
    music = models.ForeignKey(Music, on_delete=models.SET_NULL, blank=True, null=True)

    title = models.CharField(max_length = 30)
    content = models.TextField(blank=True)
    registered_at = models.DateTimeField(auto_now_add=True)
    last_update_at = models.DateTimeField(auto_now=True)
    is_open = models.BooleanField(default=False)

    class Meta: 
        managed = True
        db_table = 'diary'

Image

일기를 fk로 가지고, 이미지의 url을 다루는 모델로, 다음과 같이 모델을 구성한다.

class Image(models.Model):
    diary = models.ForeignKey(Diary, on_delete=models.CASCADE)
    created_at = models.DateTimeField(auto_now_add=True)
    image_url = models.URLField(null=True)
    image_prompt = models.TextField(null=True)
    class Meta:
        managed = True
        db_table = 'image'

Emotion

일기를 fk로 가지고, emotion_label, chat을 다루는 모델

class Emotion(models.Model):
    diary = models.ForeignKey(Diary,on_delete=models.CASCADE)
    emotion_label = models.CharField(max_length=10, blank=True)
    emotion_prompt = models.TextField(blank=True)
    chat = models.TextField(blank=True)
    class Meta:
        db_table ='emotion'

migration

전체 모델을 저장하고, 다음을 순차 실행한다.

python manage.py makemigrations
python manage.py migrate

여기까지해서 모델구성을 통한 DB 스키마 구성이 끝난다.

SERIALIZER

모델 작성을 완료 후 시리얼라이저를 작성하도록 한다.

시리얼라이저의 경우 민감한 데이터가 없는경우 대체로 all로 작성하였으며, 특수한 경우인 community와 diary만 살펴본다.

Diary

일기의 경우 크게 일기 내용과, 일기내용을 바탕으로 생성된 음악, 이미지, 감정 데이터로 크게 나눠볼 수 있다.

이중 음악의 경우는 diary모델에서 fk로 참조하기 때문에, 음악 데이터를 생성하고, 일기와 연결하는 부분은 따로 구성했다.

따라서 시리얼라이저는 다음과 같다.

1. 순수하게 일기를 작성할 수 있게 하고, 다른 연관 데이터는 only read로 읽을수만 있도록 하는 DiarySerializer

2. 일기 내용을 토대로 음악을 추천하고, 음악-일기를 연결할 수 있도록 하는 DiaryMusicSerializer

DiarySerializer

class DiarySerializer(serializers.ModelSerializer):
    music = MusicSerializer(required=False)
    image_set = ImageSerializer(many=True, read_only=True)
    emotion_set = EmotionSerializer(many=True, read_only=True)

    class Meta:
        model = Diary
        fields = ['id','user','title','content','registered_at','last_update_at','music','is_open','image_set','emotion_set']

DiaryMusicSerializer

class DiaryMusicSerializer(serializers.ModelSerializer):
    music = MusicSerializer(required=False)

    class Meta:
        model = Diary
        fields =['id', 'user', 'content', 'music']

    def update(self, instance, validated_data):
        music_data = validated_data.pop('music', None)
        instance = super().update(instance, validated_data)

        if music_data:
            music, _ = Music.objects.get_or_create(**music_data)
            instance.music = music

        return instance

Community

커뮤니티의 경우 서로 팔로우가 허용된 관계의 유저 일기 중 공개된 것들을 조회할 수 있도록 한다.

CommunitySerializer

class CommunitySerializer(serializers.ModelSerializer):
    user = UserSerializer(required = False)
    music = MusicSerializer(required=False)
    image_set = ImageSerializer(many=True, read_only=True)

    class Meta:
        model = Diary
        fields = ['id','user','title','content','music','image_set','registered_at','last_update_at', 'is_open']

여기에 팔로우 정보가 없는데요? > 이부분은 후에 permissions를 통해 가능하게 한다.

PERMISSION

뷰를 작성하기에 앞서서 접근의 제어를 위한 커스텀 permissions를 작성하자.

permissions.py

먼저 permissions.py를 settings.py가 있는 폴더에 생성한다.

이 폴더에 커스텀 permissions를 작성할 것이다.

import

permissions.py의 상단에 다음을 import 해야 한다.

from rest_framework import permissions

Permission 작성

총 세가지의 permission을 작성할것이다.

1. 본인의 데이터만 접근가능하게 하는 IsOwner

2. 본인만 수정,삭제할 수 있게 하되 허용된 팔로워에게는 조회할 수 있게 하는 IsOwnerOrReadOnly

3. 팔로잉 신청 시 본인과 팔로잉 신청에 관련된 사람만 조회,편집할 수 있게 하는 IsFollowerOrOwner

IsOwner

데이터의 소유자만 접근하고 수정 삭제 할 수 있게 한다

class IsOwner(permissions.BasePermission):
    """
    본인의 data만 접근 가능하다.
    """
    def has_permission(self, request, view):
        return request.user.is_authenticated

    def has_object_permission(self, request, view, obj):
        return obj.user == request.user

IsOwnerOrReadOnly

class IsOwnerOrReadOnly(permissions.BasePermission):
    """
    객체를 만든 사용자만 수정할 수 있다.
    """
    def has_permission(self, request, view):
        return request.user.is_authenticated

    def has_object_permission(self, request, view, obj):
        # 요청한 사용자가 해당 객체의 소유자인 경우에만 쓰기 권한을 부여함
        return obj.follower == request.user or obj.following_user == request.user

IsFollowerOrOwner

class IsFollowerOrOwner(permissions.BasePermission):
    """
    Custom permission to allow reading followed items only if they are open.
    """
    def has_permission(self, request, view):
        return request.user.is_authenticated

    def has_object_permission(self, request, view, obj):
        # Check if the request method is safe (GET, HEAD, OPTIONS)
        if request.method in permissions.SAFE_METHODS:
            if Follow.objects.filter(follower=request.user, following_user=obj.user, status='accepted').exists() | Follow.objects.filter(follower=obj.user, following_user=request.user, status='accepted').exists():
                return obj.is_open

        return obj.user == request.user

위의 permissions들을 통해 이제 사용자에 따른 접근 제어를 가능하게 한다.

VIEW

편리한 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](https://github.com/encode/django-rest-framework/blob/master/rest_framework/mixins.py)

역시 뷰가 엄청나게 많기 때문에 특수한 경우만 살펴보자.

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

        '''
        # 클라이언트로부터 사용자 이름을 받음
        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
        '''

        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

        '''

        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)


    def partial_update(self, request, *args, **kwargs):
        '''
        팔로우 요청 거절하는 API

        '''
        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이다.

동작 구조는 다음과 같다. 빨간 글씨가 있는 부분이 AI서버로 요청을 넣는 부분이다.

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)

이것으로 백엔드 Django API부분의 핵심 기능은 이것으로 구현이 끝난다.

API문서화

API 구현이 끝났다면 프론트로 각 API에 대한 정보를 넘겨줘야 작업이 가능하다.

이것을 우리는 Swagger를 사용하여 구현한다. 장고 REST Framework에서 swagger를 통해 API를 문서화 해보자.

먼저 drf-ysag에 대한 문서는 아래의 링크를 통해 이동할 수 있다. 상세한 정보를 원한다면 공식 문서를 활용하자.

https://drf-yasg.readthedocs.io/en/stable/

install

가상환경을 활성화 하고 install 하자.

pip install -U drf-yasg

settings.py

settings.py에 다음이 추가되어야 한다.

INSTALLED_APPS = [
   ...
   'django.contrib.staticfiles',  # required for serving swagger ui's css/js files
   'drf_yasg',
   ...
]

urls.py

총 4가지의 엔드포인트를 추가할것이다.

A JSON view of your API specification at /swagger.json
A YAML view of your API specification at /swagger.yaml
A swagger-ui view of your API specification at /swagger/

A ReDoc view of your API specification at /redoc/

프로젝트의 urls.py에 다음을 추가한다.

...
from django.urls import re_path
from rest_framework import permissions
from drf_yasg.views import get_schema_view
from drf_yasg import openapi

...

schema_view = get_schema_view(
openapi.Info(
  title="Snippets API",
  default_version='v1',
  description="Test description",
  terms_of_service="https://www.google.com/policies/terms/",
  contact=openapi.Contact(email="contact@snippets.local"),
  license=openapi.License(name="BSD License"),
),
public=True,
permission_classes=(permissions.AllowAny,),
)

urlpatterns = [
path('swagger<format>/', schema_view.without_ui(cache_timeout=0), name='schema-json'),
path('swagger/', schema_view.with_ui('swagger', cache_timeout=0), name='schema-swagger-ui'),
path('redoc/', schema_view.with_ui('redoc', cache_timeout=0), name='schema-redoc'),
...
]

확인
runserver 해주고, /swagger/에 접근한다.

그럼 다음처럼 api를 확인할 수 있다.

Flask : AI모델 서빙

이번에는 GPU서버에 올릴 Flask 를 작성한다.

Flask에서는 모델을 로드하고, 백엔드 API에 필요한 AI작업을 수행한다.

GPU비용 문제로, GPU가 꼭 필요한 경우만 이 서버에서 동작한다.

소개

Flask에서는 다음을 수행한다.

1. generate comment : 일기 작성 내용을 바탕으로 응원 문구를 생성한다.

2. generate image : 백엔드로부터 이미지 생성 프롬프트를 받고, 튜닝된 디퓨전 모델을 로드하여 이미지를 생성한다.

3. emotion classification : 일기 작성 내용을 바탕으로 감정 분석한다.

4. recommend music : 감정 분석 결과와 크롤링을 통해 수집한 음악 데이터의 감정분석 결과를 사용해 유사도를 통한 음악 추천, 결과 총 5가지 반환

Image

이미지를 생성하기 위한 프롬프트를 받아서 이미지 생성, S3버킷에 저장하고, URL정보를 응답으로 반환한다.

이미지를 생성하는 함수를 먼저 다른 파일에 작성하자.

#generate_image.py 에 모델을 로드하고, 이미지를 생성하는 함수를 만들었다.
from diffusers import StableDiffusionPipeline
import torch

# 모델 로드 및 디바이스 설정
model_path = '모델 위치'  # FineTuning Model Path
pipe = StableDiffusionPipeline.from_pretrained("runwayml/stable-diffusion-v1-5")
device = torch.device('cuda')
pipe.unet.load_attn_procs(model_path)
pipe.to(device)

# Negative prompt 설정
neg_prompt = '''FastNegativeV2,(bad-artist:1.0), (loli:1.2),
    (worst quality, low quality:1.4), (bad_prompt_version2:0.8),
    bad-hands-5,lowres, bad anatomy, bad hands, ((text)), (watermark),
    error, missing fingers, extra digit, fewer digits, cropped,
    worst quality, low quality, normal quality, ((username)), blurry,
    (extra limbs), bad-artist-anime, badhandv4, EasyNegative,
    ng_deepnegative_v1_75t, verybadimagenegative_v1.3, BadDream,
    (three hands:1.1),(three legs:1.1),(more than two hands:1.4),
    (more than two legs,:1.2),badhandv4,EasyNegative,ng_deepnegative_v1_75t,verybadimagenegative_v1.3,(worst quality, low quality:1.4),text,words,logo,watermark,
    '''

def get_image(prompt):
    image = pipe(prompt, negative_prompt=neg_prompt,num_inference_steps=30, guidance_scale=7.5).images[0]
    return image

대체로 15초정도 소요되고, 늦어지면 30초 이내로 생성된다. 하지만 사실 꽤 긴 시간이므로 이를 잘 고려해서 응답을 주고받는 백엔드 서버에서 응답 대기 시간을 40초 정도로 늘려주자.(위의 코드에 사실 이미 적용되어있다)

#app.py

# S3에 연결
s3 = boto3.client('s3',
                  aws_access_key_id=S3_ACESS_KEY,
                  aws_secret_access_key=S3_SECRET_ACCESS_KEY)

# 이미지 생성 요청에 대한 작업
@app.route('/get_image', methods=['POST'])
async def process_image_request():
    try:
        # 받은 요청의 데이터를 확인
        request_data = request.json
        prompt = request_data.get('prompt')

        #받은 프롬프트로 이미지를 생성한다.
        image = get_image(prompt)

        # 이미지를 저장한다.
        image_key = str(uuid.uuid4())
        buffered = io.BytesIO()
        image.save(buffered, format="JPEG")
        buffered.seek(0)
        s3.upload_fileobj(buffered, Bucket=S3_BUCKET_NAME, Key=f'images/{image_key}.jpg', ExtraArgs={'ContentType':'image/jpeg'})
        image_url = f'https://{S3_BUCKET_NAME}.s3.{AWS_S3_REGION_NAME}.amazonaws.com/images/{image_key}.jpg'
        buffered.close()

        # 저장 후 이미지의 URL을 응답한다.
        return jsonify({'image_url': image_url}), 200
    except Exception as e:
        print("Exception occurred in process_request:", e)
        return jsonify({"error": str(e)}), 500

Comment

프로세스는 위와 같다. 모델을 로드하여 작업을 수행하는 함수를 하나 만들고, API요청에 대한 작업을 app.py에 작성한다.

#모델 로드
#gpt model
print('gpt_load')
gpt_device = torch.device("cuda:0")
gpt_model = GPT2LMHeadModel.from_pretrained('모델').to(gpt_device)
gpt_tokenizer = PreTrainedTokenizerFast.from_pretrained('모델')
U_TKN = '<usr>'
S_TKN = '<sys>'
BOS = '</s>'
EOS = '</s>'
MASK = '<unused0>'
SENT = '<unused1>'
PAD = '<pad>'

def get_comment(input_text): #koGPT2 모델을 활용하여 입력된 질문에 대한 대답을 생성하는 함수
    q = input_text
    a = ""
    sent = ""
    while True:
        input_ids = torch.LongTensor(gpt_tokenizer.encode(U_TKN + q + SENT + sent + S_TKN + a)).unsqueeze(dim=0).to(gpt_device)
        pred = gpt_model(input_ids)
        pred = pred.logits
        gen = gpt_tokenizer.convert_ids_to_tokens(torch.argmax(pred, dim=-1).squeeze().tolist())[-1]
        if gen == EOS:
            break
        a += gen.replace("▁", " ")
    return a

# app.py
@app.route('/get_comment', methods=['POST'])
async def process_comment_request():
    try:
        request_data = request.json
        content = request_data.get('content')
        comment = get_comment(content)

        return jsonify({'comment': comment}), 200
    except Exception as e:
        print("Exception occurred in process_request:", e)
        return jsonify({"error": str(e)}), 500

Emotion & Music

def get_emotion_label(content):
    emotion_pred = inference(content)
    max_value = max(emotion_pred)
    max_index = emotion_pred .index(max_value)
    return emotion_pred, emotion_arr[max_index]


def get_music(content):
    emotion_pred, max_index=get_emotion_label(content)
    df_user_sentiment = pd.DataFrame([emotion_pred],columns=emotion_arr)
    user_emotion_str = df_user_sentiment.apply(lambda x: ' '.join(map(str, x)), axis=1)
    music_emotion_str = final_emotion[emotion_arr].apply(lambda x: ' '.join(map(str, x)), axis=1)

    tfidf = TfidfVectorizer()
    user_tfidf_matrix = tfidf.fit_transform(user_emotion_str)
    music_tfidf_matrix = tfidf.transform(music_emotion_str)

    cosine_sim = cosine_similarity(user_tfidf_matrix, music_tfidf_matrix)

    most_similar_song_index = cosine_sim.argmax()
    most_similar_song_info = final_emotion.iloc[most_similar_song_index]

    num_additional_recommendations = 4
    similar_songs_indices = cosine_sim.argsort()[0][-num_additional_recommendations-1:-1][::-1]
    similar_songs_info = final_emotion.iloc[similar_songs_indices]

    return most_similar_song_info, similar_songs_info

@app.route('/get_sentiment', methods=['POST'])
async def process_sentiment_request():
    try:
        request_data = request.json
        content = request_data.get('content')
        _, emotion_label = get_emotion_label(content)

        return jsonify({'emotion_label': emotion_label}), 200
    except Exception as e:
        print("Exception occurred in process_request:", e)
        return jsonify({"error": str(e)}), 500

@app.route('/get_music', methods=['POST'])
async def process_music_request():
    try:
        request_data = request.json
        content = request_data.get('content')
        most_similar_song_info, similar_songs_info = get_music(content)

        response_data = {
            'most_similar_song': {
                'title': most_similar_song_info[0],
                'artist': most_similar_song_info[1],
                'genre': most_similar_song_info[2]
            },
            'similar_songs': [{
                'title': song_info[0],
                'artist': song_info[1],
                'genre': song_info[2]
            } for song_info in similar_songs_info.values]
        }
        return jsonify(response_data), 200
    except Exception as e:
        print("Exception occurred in process_request:", e)
        return jsonify({"error": str(e)}), 500

GCP-VM에 올리기

작성한 Flask는 Github에 올리고, VM에 서빙하도록 한다.