00. Project Intro
1. Youtube REST API
- db 설계 (모델 구조)
- DB: Postgre SQL
- 기능:
- 유튜브 REST API ⇒ DRF
- 소켓 연결을 통한 실시간 채팅 기능 구현 (방 기능까지)
- 영상 스트리밍
- 영상을 시청 ⇒ 시청 데이터 ⇒ 추천 알고리즘 (모델링) - FastAPI
2. Youtube Dashboard
- ELK(Elasticsearch, Logstash, Kibana) Stack을 활용한 로그 데이터 시각화
3. Youtube Creator Support
- OpenAI API → FastAPI
- 유튜브 컨텐츠를 만들 수 있도록 도와주는
- OpenAI API를 활용해서 썸네일 및 스크립트 제작
01. Project Settings
1. Github Repository 생성
> git clone https://github.com/Seopftware/django-project-tweet.git2. 도커 허브 회원가입 진행
Docker hub 사이트에서 회원가입 진행 및 프로그램 설치.
3. Docker AccessToken 생성
- My Account > Security > New Access Token
원하는 이름으로 Access Token을 생성한 다음 마지막에 창이 하나 뜬다. 이 창을 닫으면 토큰 정보를 다시 볼 수 없다. 다만, 토큰 정보를 잊어버린 경우에는 토큰을 지운 다음 다시 발급을 받을 수 있다.
4. Github 프로젝트에 secret variable 등록
- Repository - settings - Screts and variables(Actions) - New repository secret
- 아래처럼 USER를 등록해준다.
NAME(KEY): DOCKERHUB_USER
SECRET(VALUE): # docker hub 유저네임- 아래처럼 TOKEN도 등록해준다.
NAME: DOCKERHUB_TOKEN
SECRET: # docker hub access token
5. requirements.txt 파일 생성, requirements.dev.txt 파일 생성
- requirements.txt 생성
: requirements.txt는 프로젝트 실행에 필요한 파이썬 패키지 목록을 포함한다.django>=5.0.1,<6.0.0 # 5.0.1 이상 6.0.0 미만 djangorestframework>=3.14.0,<4.0.0
django와djangorestframework는 프로젝트의 핵심 의존성이다. 또한 버전 제한을 명확히 지정하여 호환성 문제를 예방한다.
- requirements.dev.txt
: 개발 및 연습용 파이썬 패키지 목록을 포함한다.flake8>=7.0.0,<8.0.0
각 라이브러리의 버전을 확인하려면 Pypi site에서 해당 모듈을 검색하면 최신버전을 알려준다. 그리고 해당 버전을 어떻게 설치하는지도 알려준다.
하지만 우리는 모듈의 최신버전을 사용하는 것이 아닌 버전을 정해놓고 사용하려고 한다. (모듈 업데이트가 자주 이루어지기 때문에 최신 버전을 맞추기 위함)
6. Dockerfile 파일 생성, .dockerignore 파일 생성
-
Dockerfile
: Dockerfile은 도커 이미지를 구축하기 위한 지시사항을 포함한다. 여기서는 파이썬 환경을 설정하고 필요한 패키지를 설치한다.# Python 3.11이 설치된 Alpine Linux 3.19 # Alpine Linux는 경량화된 리눅스 배포판으로, 컨테이너 환경에 적합 FROM python:3.11-alpine3.19 # LABEL 명령어는 이미지에 메타데이터를 추가합니다. 여기서는 이미지의 유지 관리자를 "seopftware"로 지정하고 있습니다. LABEL maintainer="seopftware" # 환경 변수 PYTHONUNBUFFERED를 1로 설정합니다. # 이는 Python이 표준 입출력 버퍼링을 비활성화하게 하여, 로그가 즉시 콘솔에 출력되게 합니다. # 이는 Docker 컨테이너에서 로그를 더 쉽게 볼 수 있게 합니다. ENV PYTHONUNBUFFERED 1 # 로컬 파일 시스템의 requirements.txt 파일을 컨테이너의 /tmp/requirements.txt로 복사합니다. # 이 파일은 필요한 Python 패키지들을 명시합니다. COPY ./requirements.txt /tmp/requirements.txt COPY ./requirements.dev.txt /tmp/requirements.dev.txt COPY ./app /app
-
Alpine Linux에 Python 가상환경 설치 및 명령어 작성.
: EC2 하나를 띄워놨다고 생각하면 편하다.ARG DEV=False RUN python -m venv /py && \ /py/bin/pip install --upgrade pip && \ /py/bin/pip install -r /tmp/requirements.txt && \ if [ $DEV = "true" ]; \ then /py/bin/pip install -r /tmp/requirements.dev.txt ; \ fi && \ rm -rf /tmp && \ adduser \ --disabled-password \ --no-create-home \ django-user ENV PATH="/scripts:/py/bin:$PATH" USER django-user -
RUN python -m venv /py && \- Python의 가상 환경을
/py디렉토리에 생성한다. 가상 환경을 사용하면 시스템의 Python 환경과 독립적으로 패키지를 관리할 수 있다.
- Python의 가상 환경을
-
/py/bin/pip install --upgrade pip && \- 가상 환경 내에서 pip(파이썬 패키지 관리자)를 최신 버전으로 업그레이드한다.
-
/py/bin/pip install -r /tmp/requirements.txt && \- 필수 Python 패키지들을
requirements.txt파일에 명시된 대로 설치한다.
- 필수 Python 패키지들을
-
if [ $DEV = "true" ]; \- 여기서는 조건문을 사용하여
DEV환경 변수가true로 설정되어 있는지 확인한다. 이는 개발용 의존성이 설치될지를 결정한다.
- 여기서는 조건문을 사용하여
-
then /py/bin/pip install -r /tmp/requirements.dev.txt ; \DEV가true일 경우, 개발용 의존성을 담고 있는requirements.dev.txt파일에 명시된 추가 패키지들을 설치한다.
-
fi && \if조건문의 끝을 나타낸다.
-
rm -rf /tmp && \/tmp디렉토리를 삭제하여 빌드 중 생성된 임시 파일들을 정리한다. 이는 이미지의 크기를 줄이는 데 도움이 된다.
-
adduser \- 새로운 사용자를 추가하는 명령이다. 이는 보안상의 이유로 애플리케이션을 root 사용자가 아닌 일반 사용자 권한으로 실행하기 위함이다.
-
-disabled-password \- 이 옵션은 생성되는 사용자에게 패스워드를 설정하지 않는다.
-
-no-create-home \- 사용자의 홈 디렉토리를 생성하지 않는다.
-
django-user- 생성할 사용자의 이름을
django-user로 지정한다.
- 생성할 사용자의 이름을
-
ENV PATH="/py/bin:$PATH"- 환경 변수
PATH를 설정하여, 가상 환경의 Python과 스크립트 디렉토리(/scripts)에서 실행 가능한 파일들을 쉽게 찾을 수 있도록 한다.
- 환경 변수
이후 app 폴더 생성 필요하다.
-
.dockerignore
# Git .git .gitignore # Docker .docker # Python app/__pycache__/ app/*/__pycache__/ app/*/*/__pycache__/ app/*/*/*/__pycache__/ .env/ .venv/ venv/
- 단일 도커 이미지 빌드 명령어
> docker build .
7. docker-compose.yml 파일 생성
version: "3.11"
services:
app:
build:
context: .
args:
- DEV=true
ports:
- "8000:8000"
volumes:
- ./app:/app
command: sh -c "python manage.py runserver 0.0.0.0:8000"- 빌드 명령어
docker-compose build
8. django 설치
> docker-compose run --rm app sh -c "django-admin startproject app ."
> docker-compose up9. .flake8 파일 생성
app 폴더 안에 .flake8 파일을 생성한다.
[flake8]
exclude =
migrations,
__pycache__,
manage.py,
settings.py- flake8 실행 명령어
> docker-compose run --rm app sh -c "flake8"
10. github actions create
작업 공간에 .github이란 폴더를 만들고 그 안에 workflows라는 폴더를 만든다. 그리고 아래 파일을 생성해 준다.
-
`.github/workflows/checks.yml
--- name: Checks on: [push] jobs: test-lint: name: Test and Lint runs-on: ubuntu-20.04 steps: - name: Checkout uses: actions/checkout@v2 - name: Test run: docker-compose run --rm app sh -c "python manage.py test" - name: Lint run: docker-compose run --rm app sh -c "flake8"
11. git push
> git add .
> git commit -m "Project Settings"
> git push -u origin main
12. check git actions
github repository에 push가 잘 올라갔는지 확인한다.
02. PostgreSQL
1. PostgreSQL란
-
PostgreSQL과 MySQL 과 차이점.
2. PostgreSQL settings
(1) docker postgresql image pull
-
docker-compose.yml
version: "3.11" services: app: build: context: . args: - DEV=true ports: - "8000:8000" volumes: - ./app:/app command: sh -c "python manage.py runserver 0.0.0.0:8000" environment: - DB_HOST=db - DB_NAME=youtube - DB_USER=seopftware - DB_PASS=password123 depends_on: - db # app: 과 같은 뎁스로 db: image: postgres:16-alpine volumes: - ./data/db:/var/lib/postgresql/data environment: - POSTGRES_DB=youtube - POSTGRES_USER=seopftware - POSTGRES_PASSWORD=password123 -
docker-compose 작업 후
> docker-compose up --build
(2) docker postgresql connector install
-
Dockerfile
FROM python:3.11-alpine3.19 LABEL maintainer="seopftware" ENV PYTHONUNBUFFERED 1 COPY ./requirements.txt /tmp/requirements.txt COPY ./requirements.dev.txt /tmp/requirements.dev.txt COPY ./app /app WORKDIR /app EXPOSE 8000 ARG DEV=false RUN python -m venv /py && \ /py/bin/pip install --upgrade pip && \ apk add --update --no-cache postgresql-client && \ apk add --update --no-cache --virtual .tmp-build-deps \ build-base postgresql-dev musl-dev && \ /py/bin/pip install -r /tmp/requirements.txt && \ if [ $DEV = "true" ]; \ then /py/bin/pip install -r /tmp/requirements.dev.txt ; \ fi && \ rm -rf /tmp && \ apk del .tmp-build-deps && \ adduser \ --disabled-password \ --no-create-home \ django-user ENV PATH="/py/bin:$PATH" USER django-user -
requirements.txt
django>=5.0.1,<6.0.0 # 5.0.1 이상 6.0.0 미만 djangorestframework>=3.14.0,<4.0.0 psycopg2>=2.9.9,<3.0 -
docker
> docker-compose down > docker-compose build > docker-compose up
(3) PostgreSQL variable settings
-
app/settings.py
import os DATABASES = { 'default': { 'ENGINE': 'django.db.backends.postgresql', 'HOST': os.environ.get('DB_HOST'), 'NAME': os.environ.get('DB_NAME'), 'USER': os.environ.get('DB_USER'), 'PASSWORD': os.environ.get('DB_PASS'), } }
(4) creating app core
-
터미널에서 실행
docker-compose run --rm app sh -c "python manage.py startapp core" -
app/settings.py
INSTALLED_APPS = [ 'django.contrib.admin', 'django.contrib.auth', 'django.contrib.contenttypes', 'django.contrib.sessions', 'django.contrib.messages', 'django.contrib.staticfiles', 'core' ]
(5) 사용자 정의 django명령어 만들기
Django의 기본 기능 외에 특정 작업을 자동화하거나 특정한 프로세스를 구현하기 위해 추가적인 커맨드가 필요할 수 있다. 예를 들어, 특정 조건을 만족하는 데이터베이스 레코드를 대량으로 업데이트하거나, 복잡한 데이터 처리 작업을 수행하고 결과를 파일로 저장하는 등의 작업이다.
사용자 정의 커맨드를 만들기 위해서는 애플리케이션 디렉토리 안에 management/commands라는 서브디렉토리를 만들고, 그 안에 커맨드 모듈을 포함시킨다. 각 모듈은 BaseCommand를 확장하는 Command 클래스를 정의해야 하며, 이 클래스는 커맨드의 실행 로직을 포함한다. app폴더 내에 아래 파일 구조와 같은 폴더와 파일들을 생성한다.
-
파일 구조도
core/ │ ├── management/ ├── commands/ │ ├── __init__.py │ └── wait_for_db.py └── __init__.py이 구조도는 프로젝트 루트 폴더(project_root) 내부에 위치한 management와 tests 두 개의 서브디렉토리를 보여준다. management 디렉토리 내에는 또 다른 서브디렉토리인 commands가 있으며, 각 디렉토리에는 init.py 파일이 포함되어 있다. commands 디렉토리에는 wait_for_db.py라는 스크립트 파일도 포함되어 있다. tests 디렉토리에는 init.py 파일과 test_commands.py 테스트 스크립트 파일이 포함되어 있다.
__init__.py파일의 역할- 패키지 초기화
: Python에서 디렉토리가 패키지로 인식되기 위해서는 그 디렉토리 안에__init__.py파일이 있어야 한다. 이 파일이 있으면 Python은 해당 디렉토리를 패키지로 간주하고, 그 안에 있는 모듈들을 가져올(import) 수 있다. 파일 자체는 비어 있을 수도 있지만, 패키지를 초기화하는 코드를 포함할 수도 있다.
- 패키지 초기화
Python 3.3 이상에서는 PEP 420에 의해
__init__.py파일이 없는 디렉토리도 네임스페이스 패키지로 인식될 수 있다. 그러나__init__.py파일을 사용하는 것은 이전 버전의 Python과의 호환성을 위해서 중요하다.
(6) core/management/commands/wait_for_db.py
Docker와 같은 컨테이너화된 환경에서는 데이터베이스가 애플리케이션 서비스보다 늦게 준비될 수 있다. 이 경우, 애플리케이션이 데이터베이스 연결을 시도할 때 실패할 수 있으므로, 데이터베이스가 준비될 때까지 대기하는 것에 대한 처리가 필요하다.
-
core/management/commands/wait_for_db.py
from django.db import connections from django.core.management.base import BaseCommand import time from django.db.utils import OperationalError from psycopg2 import OperationalError as Psycopg2OpError class Command(BaseCommand): def handle(self, *args, **options): self.stdout.write("Wating for DB Connections...") is_db_connected = None while not is_db_connected: try: is_db_connected = connections['default'] except (Psycopg2OpError, OperationalError): self.stdout.write("Connection Trying ...") time.sleep(1) self.stdout.write(self.style.SUCCESS('PostgreSQL Connections Success')) -
docker-compose.yml
version: "3.11" services: # 첫 번째 컨테이너 : Django app: build: context: . args: - DEV=true ports: - "8000:8000" volumes: - ./app:/app command: > # 추가. sh -c 'python manage.py wait_for_db && python manage.py migrate && python manage.py runserver 0.0.0.0:8000' -
DB 관련 docker 설정 후 build & up
> docker-compose run --rm app sh -c 'python manage.py makemigrations' > docker-compose up --build -
superuser 생성 및 admin 페이지 접속
> docker-compose run --rm app sh -c 'python manage.py createsuperuser'(7) test code 작성
-
test_commands.py
""" Test custom Django management commands. """ from unittest.mock import patch from psycopg2 import OperationalError as Psycopg2OpError from django.core.management import call_command from django.db.utils import OperationalError from django.test import SimpleTestCase @patch('django.db.utils.ConnectionHandler.__getitem__') class CommandTests(SimpleTestCase): """Test commands.""" def test_wait_for_db_ready(self, patched_getitem): """Test waiting for database if database ready.""" patched_getitem.return_value = True call_command('wait_for_db') self.assertEqual(patched_getitem.call_count, 1) @patch('time.sleep') def test_wait_for_db_delay(self, patched_sleep, patched_getitem): """Test waiting for database when getting OperationalError.""" patched_getitem.side_effect = [Psycopg2OpError] + \ [OperationalError] * 5 + [True] call_command('wait_for_db') self.assertEqual(patched_getitem.call_count, 7)
(8) git CI/CD 설정
-
.github/workflows/checks.yml
--- name: Checks on: [push] jobs: test-lint: name: Test and Lint runs-on: ubuntu-20.04 steps: - name: Checkout uses: actions/checkout@v2 - name: Test run: docker-compose run --rm app sh -c "python manage.py wait_for_db && python manage.py test" - name: Lint run: docker-compose run --rm app sh -c "flake8" -
docker-compose.yml
version: "3.11" services: app: build: context: . args: - DEV=true ports: - "8000:8000" volumes: - ./app:/app command: > sh -c "python manage.py wait_for_db && python manage.py migrate && python manage.py runserver 0.0.0.0:8000" environment: - DB_HOST=db - DB_NAME=youtube - DB_USER=seopftware - DB_PASS=password123 depends_on: - db # app: 과 같은 뎁스로 db: image: postgres:16 volumes: - ./data/db:/var/lib/postgresql/data environment: - POSTGRES_DB=youtube - POSTGRES_USER=seopftware - POSTGRES_PASSWORD=password123
03. Model Create
1. Youtube REST API
- 모델 구조
1. User (Custom)
- email
- password
- nickname
- is_business(boolean): personal, business
2. Video
- title
- link
- description
- category
- views_count
- thumbnail
- video_uploaded_url (S3)
- video_file(FileField)
- User:FK
3. Reaction
- User:FK
- Video:FK
- reaction
4. Comment
- User:FK
- Video:FK
- like
- dislike
- content
5. Subcription (채널 구독)
- User:FK => subscriber (구독한) -> 내가 구독한 사람
- User:FK => subscribed_to (구독을 당한) -> 나를 구독한 사람
6. Notification
- User:FK
- message
- is_read
7. Common
- created_at
- updated_at
8. Chatting (예정)
- User:FK (nickname)- Table Create
# users, videos, reactions, comments, subscriptions, common
- docker-compose run --rm app sh -c 'python manage.py startapp users'
- docker-compose run --rm app sh -c 'python manage.py startapp videos'
- docker-compose run --rm app sh -c 'python manage.py startapp reactions'
- docker-compose run --rm app sh -c 'python manage.py startapp comments'
- docker-compose run --rm app sh -c 'python manage.py startapp subscriptions'
- docker-compose run --rm app sh -c 'python manage.py startapp common'2. Create Custom UserModel
-
users 앱 생성
: 위에서 이미 생성했으므로 생략 가능한 과정이다.> docker-compose run --rm app sh -c "django-admin startapp users" -
app/settings.py
AUTH_USER_MODEL = 'users.User' -
users/models.py
from django.db import models from django.contrib.auth.models import ( AbstractBaseUser, BaseUserManager, PermissionsMixin, ) class UserManager(BaseUserManager): """사용자 모델 관리자로, 사용자 생성 및 관리를 처리합니다.""" def create_user(self, email, password=None, **extra_fields): """이메일, 비밀번호 및 추가 필드를 사용하여 새로운 사용자를 생성하고 반환합니다. 이메일이 제공되지 않을 경우 ValueError를 발생시킵니다. """ if not email: raise ValueError('사용자는 이메일 주소를 가져야 합니다') user = self.model(email=self.normalize_email(email), **extra_fields) user.set_password(password) user.save(using=self._db) return user def create_superuser(self, email, password): """Create and return a new superuser.""" user = self.create_user(email, password) user.is_staff = True user.is_superuser = True user.save(using=self._db) return user class User(AbstractBaseUser, PermissionsMixin): """시스템 내 개별 사용자를 나타내는 사용자 모델입니다.""" email = models.EmailField(max_length=255, unique=True) nickname = models.CharField(max_length=255) is_business = models.BooleanField(default=False) is_active = models.BooleanField(default=True) is_staff = models.BooleanField(default=False) objects = UserManager() USERNAME_FIELD = 'email' def __str__(self): """사용자의 문자열 표현을 반환합니다.""" return self.email -
users/tests.py
from django.test import TestCase from django.contrib.auth import get_user_model class UserTestCase(TestCase): # 회원가입을 가정하고 => 회원가입 함수 테스트 코드를 작성하려고 합니다. # 이메일과 패스워드를 입력받고, 회원가입이 정상적으로 잘 이뤄졌는지 체크 def test_create_user(self): email = 'inseop@gmail.com' password = 'password123' user = get_user_model().objects.create_user( email=email, password=password ) self.assertEqual(user.email, email) self.assertTrue(user.check_password(password)) self.assertFalse(user.is_superuser) def test_create_superuser(self): email = 'inseop_super@gmail.com' password = 'password123' super_user = get_user_model().objects.create_superuser( email=email, password=password ) self.assertTrue(super_user.is_superuser) self.assertTrue(super_user.is_staff) -
실행
> docker-compose run --rm app sh -c 'python manage.py test users' > docker-compose run --rm app sh -c 'python manage.py createsuperuser'
3. Setup DRF(Django Restframework) & DRF-Spectacular
-
requirements.txt
# DRF-Spectacular drf-spectacular>=0.27.1,<0.28.0 -
관련 공식 사이트
-
실행
> docker-compose build -
app/settings.py
INSTALLED_APPS = [ 'django.contrib.admin', 'django.contrib.auth', 'django.contrib.contenttypes', 'django.contrib.sessions', 'django.contrib.messages', 'django.contrib.staticfiles', 'core' # 여기서 부터 CUSTOM_USER_APPS 'users.apps.UsersConfig', 'rest_framework', 'drf_spectacular', ] # 나는 Django_SYSTEM_APPS와 CUSTOM_USER_APPS로 나누어 INSTALLED_APPS = Django_SYSTEM_APPS + CUSTOM_USER_APPS # 로 추가했다. # 맨 마지막에 추가. REST_FRAMEWORK = { 'DEFAULT_SCHEMA_CLASS': 'drf_spectacular.openapi.AutoSchema', } -
app/urls.py
from drf_spectacular.views import SpectacularAPIView, SpectacularRedocView, SpectacularSwaggerView urlpatterns = [ path('api/v1/schema/', SpectacularAPIView.as_view(), name='schema'), path('api/v1/schema/swagger-ui/', SpectacularSwaggerView.as_view(url_name='schema'), name='swagger-ui'), path('api/v1/schema/redoc/', SpectacularRedocView.as_view(url_name='schema'), name='redoc'), ]docker-compose up --build를 한 뒤 http://127.0.0.1:8000/api-sh 에 접속하게 되면 접속 할 수 있는 주소들을 확인 할 수 있다.
# 도커 실행
> docker-compose up
# 아래 주소에서 테스트
127.0.0.1:8000/api/schema
127.0.0.1:8000/api/schema/swagger-ui
127.0.0.1:8000/api/schema/redocgithub action을 사용하기 위해서 Docker를 연결해준다.
jobs:
test-lint:
name: Test and Lint
runs-on: ubuntu-20.04
steps:
# 도커 연결 작업 (추가)
- name: Login in to Docker Hub
uses: docker/login-action@v1
with:
username: ${{ secrets.DOCKERHUB_USERNAME }}
password: ${{ secrets.DOCKERHUB_TOKEN }}
# 여기서부터는 기존 코드
- name: Checkout
uses: actions/checkout@v2
- name: Test
run: docker-compose run --rm app sh -c "python manage.py test"
- name: Lint
run: docker-compose run --rm app sh -c "flake8"4. Create App & Model Create
python manage.py startapp videos
python manage.py startapp reactions
python manage.py startapp notifications
python manage.py startapp subscriptions
python manage.py startapp comments
python manage.py startapp common기존에 해당 app들을 만들었다면 생략 가능하다.
-
common/models.py
from django.db import models class CommonModel(models.Model): # 생성된 시간 - 고정 값. created_at = models.DateTimeField(auto_now_add=True) # 업데이트된 시간 - 유동 값. updated_at = models.DateTimeField(auto_now=True) class Meta: # DB에 Table 추가 X abstract = True -
users/models.py
from django.db import models from django.contrib.auth.models import AbstractUser class User(AbstractUser): email = models.EmailField(unique=True) nickname = models.CharField(max_length=50, unique=True) is_business = models.BooleanField(default=False) -
videos/models.py
from django.db import models from users.models import User class Video(CommonModel): title = models.CharField(max_length=255) link = models.URLField() description = models.TextField(blank=True) category = models.CharField(max_length=50) views_count = models.PositiveIntegerField(default=0) thumbnail = models.URLField() video_uploaded_url = models.URLField() video_file = models.FileField(upload_to='videos/') user = models.ForeignKey(User, on_delete=models.CASCADE) -
reactions/models.py
from django.db import models from users.models import User from videos.models import Video class LikeDislike(models.Model): user = models.ForeignKey(User, on_delete=models.CASCADE) video = models.ForeignKey(Video, on_delete=models.CASCADE) like = models.BooleanField(default=True) # True for like, False for dislike -
comments/models.py
from django.db import models from users.models import User from videos.models import Video class Comment(models.Model): user = models.ForeignKey(User, on_delete=models.CASCADE) video = models.ForeignKey(Video, on_delete=models.CASCADE) content = models.TextField() like = models.PositiveIntegerField(default=0) dislike = models.PositiveIntegerField(default=0) -
subscriptions/models.py
from django.db import models from users.models import User class Subscription(models.Model): subscriber = models.ForeignKey(User, related_name='subscribing', on_delete=models.CASCADE) subscribed_to = models.ForeignKey(User, related_name='subscribers', on_delete=models.CASCADE) -
notifications/models.py
from django.db import models from users.models import User class Notification(models.Model): user = models.ForeignKey(User, on_delete=models.CASCADE) message = models.TextField() is_read = models.BooleanField(default=False) created_at = models.DateTimeField(auto_now_add=True)
예를 들어, 비디오 상세 조회 API에 댓글을 포함시키고 싶다면 다음과 같은 API 설계를 고려할 수 있다.
5. Video REST API
-
API 요약표
-
videos/views.py
from django.shortcuts import render from rest_framework import status from rest_framework.views import APIView from rest_framework.response import Response from rest_framework.exceptions import NotFound from .models import Video from .serializers import VideoSerializer # Create your views here. # Video와 관련된 REST API # 1. VideoList # api/v1/video # [GET] - 전체 비디오 목록 조회 # [POST] - 새로운 비디오 생성 # [PUT] - X # [DELETE] - X class VideoList(APIView): def get(self, request): videos = Video.objects.all() # 직렬화 (Object → Json) : Serializer serializers = VideoSerializer(videos, many=True) return Response(serializers.data, status=status.HTTP_200_OK) def post(self, request): user_data = request.data serializer = VideoSerializer(data=user_data) if serializer.is_valid(): serializer.save(user=request.user) # serializer = VideoSerializer(video) return Response(serializer.data, status=status.HTTP_201_CREATED) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) # 2. VideoDetail # api/v1/video/{video_id} # [GET] - 특정 비디오 조회 # [POST] - X # [PUT] - 특정 비디오 업데이트 # [DELETE] - 특정 비디오 삭제 class VideoDetail(APIView): def get(self, request, pk): try: video = Video.objects.get(pk=pk) except Video.DoesNotExist: raise NotFound serializer = VideoSerializer(video) return Response(serializer.data, status=status.HTTP_200_OK) def put(self, request, pk): video_obj = Video.objects.get(pk=pk) # DB에서 불러온 데이터 user_data = request.data # user가 보낸 데이터 serializer = VideoSerializer(video_obj, user_data) serializer.is_valid(raise_exception=True) serializer.save() return Response(serializer.data, status=status.HTTP_200_OK) def delete(self, request, pk): video_obj = Video.objects.get(pk=pk) video_obj.delete() return Response(status=status.HTTP_204_NO_CONTENT) -
videos/tests.py
from rest_framework.test import APITestCase from rest_framework import status from django.urls import reverse from django.core.files.uploadedfile import SimpleUploadedFile from users.models import User from .models import Video # Create your tests here. class VideoAPITestCase(APITestCase): # setup - test code가 실행되기 전 동작하는 함수 # 필요 항목 (1) 유저 생성(ORM) 및 로그인 (2) 비디오 생성 def setUp(self): self.user = User.objects.create_user( email = 'mjk@gmail.com', password = 'password123' ) self.client.login(email='mjk@gmail.com', password='password123') # self를 적으면 지역변수가 된다! 유레카! self.video = Video.objects.create( title = 'test video', link = 'https://www.test.com', user = self.user ) # Video List 관련 API def test_video_list_get(self): # url = 'http://127.0.0.1:8000/api/v1/video' # 위 처럼 url을 직접 적어서 불러와도 되지만 # 주소가 조금이라도 바뀌면 계속 수정해줘야되기 때문에 # urls.py 파일을 만들어 url만 따로 관리해준다. url = reverse('video-list') res = self.client.get(url) # 전체 비디오 조회 데이터 (응답값) self.assertEqual(res.status_code, status.HTTP_200_OK) self.assertEqual(res.headers['Content-Type'], 'application/json') self.assertTrue(len(res.data) > 0) # Video의 title column이 응답 데이터에 잘 들어가 있는지 확인. for video in res.data: self.assertIn('title', video) def test_video_list_post(self): url = reverse('video-list') data = { 'title': 'test video2', 'link': 'https://test.com', 'category' : 'test category', 'thumbnail' : 'http://test.com', 'video_file' : SimpleUploadedFile('file.mp4', b'file_content', 'video/mp4'), 'user' : self.user.pk } res = self.client.post(url, data) self.assertEqual(res.status_code, status.HTTP_201_CREATED) self.assertEqual(res.data['title'], 'test video2') # Video Detail 관련 API def test_video_detail_get(self): url = reverse('video-detail', kwargs={'pk':self.video.pk}) # url_ex) api/v1/video/1 res = self.client.get(url) self.assertEqual(res.status_code, status.HTTP_200_OK) def test_video_detail_put(self): url = reverse('video-detail', kwargs={'pk':self.video.pk}) data = { 'title': 'updated video', 'link': 'https://test.com', 'category' : 'test category', 'thumbnail' : 'http://test.com', 'video_file' : SimpleUploadedFile('file.mp4', b'file_content', 'video/mp4'), 'user' : self.user.pk } res = self.client.put(url, data) self.assertEqual(res.status_code, status.HTTP_200_OK) self.assertEqual(res.data['title'], 'updated video') def test_video_detail_delete(self): url = reverse('video-detail', kwargs={'pk':self.video.pk}) res = self.client.delete(url) self.assertEqual(res.status_code, status.HTTP_204_NO_CONTENT) res = self.client.get(url) self.assertEqual(res.status_code, status.HTTP_404_NOT_FOUND)video 파일을 업로드할 storage 폴더를 같은 app 폴더 안에 만들어준다.
-
videos/urls.py (new file)
from django.urls import path # from . import views from .views import VideoList, VideoDetail # api/v1/video urlpatterns = [ path('', VideoList.as_view(), name='video-list'), #{pk} path('<int:pk>', VideoDetail.as_view(), name='video-detail'), ] -
videos/serializers.py (new file)
from rest_framework import serializers from .models import Video from users.serializers import UserSerializer class VideoSerializer(serializers.ModelSerializer): user = UserSerializer(read_only=True) class Meta: model = Video fields = '__all__'
video 관련 REST API 작성 후 app에서도 변경사항 반영해준다.
- app/urls.py
from django.contrib import admin
from django.urls import path, include # include 추가
from drf_spectacular.views import SpectacularAPIView, SpectacularRedocView, SpectacularSwaggerView
urlpatterns = [
path('admin/', admin.site.urls),
# YOUR PATTERNS
path('api/v1/schema/', SpectacularAPIView.as_view(), name='schema'),
# Optional UI:
# Swagger-UI - 개발자가 개발할 때 사용
path('api/v1/schema/swagger-ui/', SpectacularSwaggerView.as_view(url_name='schema'), name='swagger-ui'),
# Redoc - 기획자나 비개발자분들이 결과물 확인시 사용
path('api/v1/schema/redoc/', SpectacularRedocView.as_view(url_name='schema'), name='redoc'),
# REST API
path('api/v1/video/', include('videos.urls')) # video 추가
]- app/settings.py
CUSTOM_USER_APPS = [
'users.apps.UsersConfig',
'videos.apps.VideosConfig', # video 추가
'rest_framework',
'drf_spectacular',
]admin 페이지에서 볼 수 있게 라우트 해준다.
-
videos/admin.py
from django.contrib import admin from .models import Video # Register your models here. @admin.register(Video) class VideoAdmin(admin.ModelAdmin): pass
Video에서 User에 대한 정보도 보고 싶다면 아래와 같이 users에서 serializers 코드를 작성한다.
-
users/serializers.py (new file)
from rest_framework import serializers from .models import User class UserSerializer(serializers.ModelSerializer): class Meta: model = User fields = ('id', 'email', 'nickname')
이제는 Video에서 댓글도 보고 싶다. 댓글(comments) 모델도 구현해준다.
-
comments/models.py
from django.db import models from common.models import CommonModel from users.models import User from videos.models import Video # Create your models here. class Comment(CommonModel): content = models.TextField() like = models.PositiveIntegerField(default=0) dislike = models.PositiveIntegerField(default=0) user = models.ForeignKey(User, on_delete=models.CASCADE) video = models.ForeignKey(Video, on_delete=models.CASCADE) -
comments/serializers.py (new file)
from rest_framework import serializers from .models import Comment class CommentSerializer(serializers.ModelSerializer): class Meta: model = Comment fields = '__all__'
DB에도 등록해준다.
- app/settings.py
CUSTOM_USER_APPS = [
'users.apps.UsersConfig',
'videos.apps.VideosConfig',
'comments.apps.CommentsConfig', # 추가
'rest_framework',
'drf_spectacular',
]video를 ForeignKey로 참조해줬으니 이제 comment를 video에 역참조해줘야한다.
-
videos/serializers.py
from rest_framework import serializers from .models import Video from users.serializers import UserSerializer from comments.serializers import CommentSerializer class VideoListSerializer(serializers.ModelSerializer): # Video:User - Video(FK) → User user = UserSerializer(read_only=True) class Meta: model = Video fields = '__all__' # 이 아래부분 추가. # 전체 list에서는 댓글이 보이지 않지만 동영상 한개에서는 보일 수 있도록. class VideoDetailSerializer(serializers.ModelSerializer): # Video:User - Video(FK) → User user = UserSerializer(read_only=True) # Video:Comment - Video → Comment(FK) comment_set = CommentSerializer(many=True, read_only=True) class Meta: model = Video fields = '__all__'
새로운 모델에 대한 내용이 추가되었으니 아래 명령으로 migrate를 해준다.
> docker-compose run --rm app sh -c 'python manage.py makemigrations'
> docker-compose run --rm app sh -c 'python manage.py migrate'이제 http://127.0.0.1:8000/api/v1/video/{int}에 접속하면 각 비디오별 댓글도 확인할 수 있다.
6. Subscription REST API
-
API 요약표
-
app/settings.py
CUSTOM_USER_APPS = [ 'users.apps.UsersConfig', # Config: label 변경할 일이 많다. 'videos.apps.VideosConfig', 'comments.apps.CommentsConfig', 'subscriptions.apps.SubscriptionsConfig', # 추가 ... ] -
subscriptions/views.py
from django.shortcuts import get_object_or_404 from rest_framework.views import APIView from rest_framework.response import Response from rest_framework import status from .models import Subscription, User from .serializers import SubscriptionSerializer from rest_framework.exceptions import ValidationError from django.db.models import Q class SubscriptionList(APIView): def post(self, request): subscriber = request.user subscribed_to_id = request.data.get('subscribed_to') if subscriber.id == subscribed_to_id or Subscription.objects.filter(subscriber=subscriber, subscribed_to_id=subscribed_to_id).exists(): return Response({"error": "Invalid subscription request."}, status=status.HTTP_400_BAD_REQUEST) request.data['subscriber'] = subscriber.id serializer = SubscriptionSerializer(data=request.data) try: serializer.is_valid(raise_exception=True) serializer.save() return Response(serializer.data, status=status.HTTP_201_CREATED) except ValidationError as e: return Response(e.detail, status=status.HTTP_400_BAD_REQUEST) class SubscriptionDetail(APIView): def delete(self, request, pk): subscriber = request.user subscription = get_object_or_404(Subscription, Q(subscriber=subscriber) & Q(subscribed_to=pk)) subscription.delete() return Response(status=status.HTTP_204_NO_CONTENT)이 코드는 사용자가 다른 사용자를 구독하고 구독 취소할 수 있는 기능을 제공한다. SubscriptionList 뷰에서는 사용자가 새로운 구독을 생성할 수 있으며, 자신을 구독하거나 이미 구독한 채널을 중복 구독하는 것을 방지한다. SubscriptionDetail 뷰에서는 사용자가 구독을 취소할 수 있다. 또한, 구독이 존재하지 않는 경우에는 404 오류를 반환한다.
-
subscriptions/urls.py
from django.urls import path from .views import SubscriptionList, SubscriptionDetail urlpatterns = [ path('subscriptions/', SubscriptionList.as_view(), name='subscription-list'), path('subscriptions/<int:pk>/', SubscriptionDetail.as_view(), name='subscription-detail'), ]이렇게 하면
subscriptions/URL을 통해 새로운 구독을 생성하고,subscriptions/<int:pk>/URL을 통해 특정 구독을 취소할 수 있다. 여기서 pk는 구독하려는 사용자의 ID다.
나는 URL이 너무 길어져서app/urls.py에서path('api/v1/sub/', include('subscriptions.urls'))를 추가해주고 해당 파일을 아래와 같이 코드를 작성했다.urlpatterns = [ path('', SubscriptionList.as_view(), name='sub-list'), path('<int:pk>', SubscriptionDetail.as_view(), name='sub-detail') ] -
subscriptions/serializers.py
from rest_framework import serializers from .models import Subscription class SubscriptionSerializer(serializers.ModelSerializer): class Meta: model = Subscription fields = ['id', 'subscriber', 'subscribed_to']
이제 test code를 작성해서 확인해보자.
-
subscriptions/tests.py
from django.test import TestCase from django.contrib.auth import get_user_model from rest_framework.test import APIClient from rest_framework import status from .models import Subscription User = get_user_model() class SubscriptionTestCase(TestCase): def setUp(self): self.client = APIClient() self.user1 = User.objects.create_user(username='user1', password='test123', email='user1@example.com') self.user2 = User.objects.create_user(username='user2', password='test123', email='user2@example.com') self.client.force_authenticate(user=self.user1) def test_create_subscription(self): response = self.client.post('/subscriptions/', {'subscribed_to': self.user2.id}) self.assertEqual(response.status_code, status.HTTP_201_CREATED) self.assertTrue(Subscription.objects.filter(subscriber=self.user1, subscribed_to=self.user2).exists()) def test_create_subscription_to_self(self): response = self.client.post('/subscriptions/', {'subscribed_to': self.user1.id}) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) def test_create_duplicate_subscription(self): Subscription.objects.create(subscriber=self.user1, subscribed_to=self.user2) response = self.client.post('/subscriptions/', {'subscribed_to': self.user2.id}) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) def test_delete_subscription(self): subscription = Subscription.objects.create(subscriber=self.user1, subscribed_to=self.user2) response = self.client.delete(f'/subscriptions/{self.user2.id}/') self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT) self.assertFalse(Subscription.objects.filter(id=subscription.id).exists()) def test_delete_nonexistent_subscription(self): response = self.client.delete(f'/subscriptions/{self.user2.id}/') self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)-
유저가 다른 유저를 구독할 수 있는지
:test_create_subscription -
유저가 자기 자신을 구독할 수 없는지
:test_create_subscription_to_self -
중복 구독을 방지하는지
:test_create_duplicate_subscription -
유저가 구독을 취소할 수 있는지
:test_delete_subscription -
존재하지 않는 구독을 취소할 수 없는지
:test_delete_nonexistent_subscription
-
마지막으로 migrate를 하고 test code가 작동하는지, URL에 잘 접속되어 원하는 데이터들이 보이는지 확인한다.
> docker-compose run --rm app sh -c 'python manage.py makemigrations'
> docker-compose run --rm app sh -c 'python manage.py migrage'
> docker-compose run --rm app sh -c 'python manage.py test subscriptions'
(나는 모든 table을 admin 코드를 넣어 보이도록 만들었다. 데이터 관리를 시각화 하기 위함.)
7. Reaction REST API
= 좋아요/싫어요 갯수 구현
(1) VideoManager의 annotate 방식
VideoManager에서 annotate를 사용하는 방법은 Video 객체를 쿼리할 때마다 자동으로 관련 Reaction 객체들의 수를 계산하여 추가한다.
class VideoManager(models.Manager):
def get_queryset(self):
return super().get_queryset().annotate(
likes_count=Count('reaction', filter=Q(reaction__reaction=Reaction.LIKE)),
dislikes_count=Count('reaction', filter=Q(reaction__reaction=Reaction.DISLIKE))
)-
용도
:Video객체의 쿼리셋에 자동으로 '좋아요'와 '싫어요'의 수를 포함시켜,Video객체에 대한 정보를 요청할 때마다 이 데이터를 함께 제공한다. -
쿼리 최적화
: 이 방식은Video객체를 조회할 때 '좋아요'와 '싫어요'의 수를 함께 계산하다. 이는Video리스트를 불러올 때 매Video객체마다 별도의 쿼리를 수행하는 것보다 효율적일 수 있다. 즉, 한 번의 쿼리로 여러Video객체에 대한 정보와 각각의 '좋아요'와 '싫어요' 수를 얻을 수 있다.
(2) Reaction 모델에서 집계 로직 구현
먼저, Reaction 모델에 Video에 대한 '좋아요'와 '싫어요'의 총계를 계산하는 메소드를 추가할 수 있다. 예를 들어, 다음과 같은 메소드를 Reaction 모델에 추가할 수 있다. (나는 이 방식을 채택하여 실습에 적용하였다.)
from django.db import models
from django.db.models import Count, Q
class Reaction(models.Model):
# 기존 필드들...
@staticmethod
def get_video_reactions(video):
reactions = Reaction.objects.filter(video=video).aggregate(
likes_count=Count('pk', filter=Q(reaction=Reaction.LIKE)),
dislikes_count=Count('pk', filter=Q(reaction=Reaction.DISLIKE))
)
return reactions이 메소드는 특정 Video 객체에 대한 '좋아요'와 '싫어요'의 총계를 계산한다.
+) Video 모델에서 Reaction 정보 사용
그 다음, Video 모델이나 Video의 뷰 또는 시리얼라이저에서 이 메소드를 사용하여 필요한 정보를 얻을 수 있다. 예를 들어, VideoSerializer에서 이 메소드를 사용하여 각 비디오에 대한 반응을 포함시킬 수 있다.
pythonCopy code
from rest_framework import serializers
from .models import Video, Reaction
class VideoSerializer(serializers.ModelSerializer):
reactions = serializers.SerializerMethodField()
class Meta:
model = Video
fields = ['title', 'link', 'description', 'user', 'reactions', ...]
def get_reactions(self, obj):
return Reaction.get_video_reactions(obj)
이 방법을 사용하면 VideoSerializer가 비디오 데이터를 시리얼라이즈할 때 Reaction 모델의 메소드를 호출하여 해당 비디오에 대한 '좋아요'와 '싫어요'의 총계를 가져온다.
이 방식의 장점은 Reaction 모델의 로직을 중앙집중화하여 관리할 수 있다는 것이다. 그러나, 이 방법은 Video 객체를 시리얼라이즈할 때마다 추가적인 데이터베이스 쿼리가 발생할 수 있으므로 성능 측면에서 고려해야 한다.
아래는 최종 코드이다.
-
reactions/admin.py
from django.contrib import admin from .models import Reaction # Register your models here. @admin.register(Reaction) class ReactionAdmin(admin.ModelAdmin): pass -
reactions/models.py
from django.db import models from django.db.models import Count, Q from common.models import CommonModel # Create your models here. class Reaction(CommonModel): user = models.ForeignKey('users.User', on_delete=models.CASCADE) video = models.ForeignKey('videos.Video', on_delete=models.CASCADE) LIKE = 1 DISLIKE = -1 NO_REACTION = 0 REACTION_CHOICES = ( (LIKE, 'Like'), (DISLIKE, 'Dislike'), (NO_REACTION, 'No Reaction'), ) reaction = models.IntegerField( choices=REACTION_CHOICES, default=NO_REACTION, ) @staticmethod # ORM depth 2! def get_video_reaction(video): reactions = Reaction.objects.filter(video=video).aggregate( like_count = Count('pk', filter=Q(reaction=Reaction.LIKE)), dislike_count = Count('pk', filter=Q(reaction=Reaction.DISLIKE)), ) return reactions
04. Chatting 기능 구현
- Chat API
1. Django Channels란
Websocket 통신 처리
2. WSGI와 ASGI 차이점
ASGI (Asynchronous Server Gateway Interface)와 WSGI(Web Server Gateway Interface)는 Python 웹 애플리케이션과 웹 서버 간의 통신 규약이다.
이 둘의 주요 차이점은 비동기 처리 능력에 있다.
- WSGI - HTTP
-
동기적 처리
: WSGI는 동기적 처리를 기반으로 합니다. 이는 요청이 처리되는 동안 해당 스레드나 프로세스가 차단(blocked)되어 다른 요청을 처리할 수 없음을 의미합니다. -
한정된 동시성
: 이로 인해 동시에 많은 요청을 처리하는 데 한계가 있으며, 고성능 비동기 처리를 위한 최적의 선택이 아닐 수 있다. -
전통적인 웹 애플리케이션에 적합
: 간단한 웹 애플리케이션 또는 비동기 처리가 중요하지 않은 경우에 적합하다.
- ASGI - TCP/UDP
- 비동기적 처리
: ASGI는 비동기적 처리를 지원한다. 이는 서버가 여러 요청을 동시에 처리할 수 있도록 하여, I/O 작업이 많은 애플리케이션에서 효율적이다. - 확장된 동시성
: 더 많은 요청을 동시에 처리할 수 있으며, 이는 특히 웹소켓, 롱 폴링과 같은 실시간 기능을 구현할 때 유용하다. - 현대적 웹 애플리케이션에 적합
: 실시간 통신이 필요한 애플리케이션, 예를 들어 채팅 앱, 실시간 대시보드 등에 적합하다.
3. Django Channels 설치
-
Chat 모델 생성
> docker-compose run --rm app sh -c 'python manage.py startapp chat' -
requirements.txt 수정
# 1. django 5.0.1이상 6.0.0 미만 django>=5.0.1,<6.0.0 # 2. django rest framework djangorestframework>=3.14.0,<4.0.0 # 3. psycopg2 - postgrsql adapter psycopg2>=2.9.9,<3.0.0 # 오류 난다면 psycopg2-binary로 수정. # 4. DRF-Spectacular drf-spectacular>=0.27.1,<0.28.0 # 5. channels - 추가 channels>=4.0.0,<4.0.9 -
설치 후, settings.py 파일에 Channels를 추가
INSTALLED_APPS = [ 'channels', 'chat.apps.ChatConfig' ] # 나는 CUSTOM_USER_APPS에 추가했다. # Channels를 사용하기 위한 설정 # WSGI_APPLICATON 위에 추가해준다. ASGI_APPLICATION = 'app.routes.application' # Channels 설정 CHANNEL_LAYERS = { "default": { "BACKEND": "channels.layers.InMemoryChannelLayer" }, } -
app/route.py (new file)
from channels.routing import ProtocolTypeRouter, URLRouter from channels.auth import AuthMiddlewareStack from chat.routes import websocket_urlpatterns import os from django.core.asgi import get_asgi_application os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'app.settings') application = ProtocolTypeRouter({ 'http': get_asgi_application(), 'websocket': AuthMiddlewareStack( URLRouter( websocket_urlpatterns # ws://127.0.0.1:8000/ws/room ) ) }) -
chat/routes.py
from django.urls import re_path, path from .consumers import ChatConsumer websocket_urlpatterns = [ # re_path(r'ws/chat/(?P<room_id>\d+)/'), # URL과 VIEW를 연결하는데, 정규 표현식을 사용한 경우 path('ws/chat/<int:room_id>/', ChatConsumer.as_asgi()) ]- In-Memory Channel Layer를 설정
-
chat/models.py : chat 모델 정의
# Chat 모델 # - ChatRoom: 오픈채팅방(비번O,비번X), 개인채팅방 # - ChatMessage: 메세지를 주고 받는 모델 from django.db import models from users.models import User from common.models import CommonModel class ChatRoom(CommonModel): name = models.CharField(max_length=100) class ChatMessage(CommonModel): # 채팅방이 삭제되면 => 메세지도 삭제되어야 하는가?? ㅇㅇ room = models.ForeignKey(ChatRoom, on_delete=models.CASCADE) sender = models.ForeignKey(User, on_delete=models.CASCADE) message = models.TextField() # Room:Message (1:N) # - Room => Message, Message, Message (O) # - Message => Room1, Room2, Room3 (X)- ChatRoom과 ChatMessage를 분리함으로써 얻는 이점
-
확장성(Scalability)
: 채팅방에 대한 정보를 별도로 관리함으로써, 향후 채팅방 관련 기능(예: 채팅방 설정 변경, 참가자 관리 등)을 확장하기 쉽다. 각 채팅방에 대한 설정이나 메타데이터를 추가하는 것이 더 용이해진다. -
데이터베이스 효율성(Database Efficiency)
: 채팅 메시지와 채팅방 정보를 분리함으로써, 데이터베이스 쿼리의 효율성이 향상된다. 예를 들어, 특정 채팅방의 정보를 조회할 때 채팅 메시지 전체를 로드할 필요가 없으며, 반대로 특정 메시지에 대한 쿼리도 더 간단하고 빠르게 수행할 수 있다. -
유연한 데이터 관리(Flexible Data Management)
: 채팅방마다 고유한 속성을 가질 수 있으며, 예를 들어 공개 채팅방, 비밀 채팅방 등 다양한 유형의 채팅방을 쉽게 구현할 수 있다. 또한, 채팅방의 생성, 삭제, 참가자 추가 등의 작업을 더 유연하게 관리할 수 있다. -
데이터 무결성(Data Integrity)
:ForeignKey를 사용하여ChatMessage가ChatRoom에 종속되게 함으로써, 데이터 무결성을 유지한다. 예를 들어, 채팅방이 삭제되면 해당 채팅방에 속한 모든 메시지도 함께 삭제되어야 하는 경우on_delete=models.CASCADE옵션이 이를 보장한다.
-
chat/serializers.py (new file)
from rest_framework import serializers from .models import ChatRoom, ChatMessage class ChatRoomSerializer(serializers.ModelSerializer): class Meta: model=ChatRoom fields="__all__" class ChatMessageSerializer(serializers.ModelSerializer): class Meta: model=ChatMessage fields="__all__" read_only_fields=['room', 'sender'] depth=1 -
chat/views.py
-
/api/chatrooms는 사용 가능한 채팅방 목록을 로드한다. -
/api/chatrooms/{roomId}/messages는 특정 채팅방의 메시지들을 로드한다.from rest_framework.views import APIView from .models import ChatRoom, ChatMessage from .serializers import ChatRoomSerializer, ChatMessageSerializer from rest_framework.response import Response from rest_framework import status from django.shortcuts import render def show_html(request): return render(request, 'index.html') # ChatRoom # [GET]: 전체 채팅방을 조회 # [POST]: 채팅방 생성 class ChatRoomList(APIView): def get(self, request): chatrooms = ChatRoom.objects.all() # 장고 객체 -> JSON (직렬화) serializer = ChatRoomSerializer(chatrooms, many=True) return Response(serializer.data, status=status.HTTP_200_OK) def post(self, request): user_data = request.data serializer = ChatRoomSerializer(data=user_data) # 역직렬화 (json to django objects) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=status.HTTP_201_CREATED) return Response(serializer.erros, status=status.HTTP_400_BAD_REQUEST) # ChatMessage # [GET]: 특정 채팅방의 채팅 내역 -> 카카오 채팅 서버(채팅 내역을 로컬에 저장) # [POST]: 채팅 메세지 생성 from django.shortcuts import get_object_or_404 class ChatMessageList(APIView): def get(self, request, room_id): chatroom = get_object_or_404(ChatRoom, id=room_id) messages = ChatMessage.objects.filter(room=chatroom) # django objects # 직렬화 serializer = ChatMessageSerializer(messages, many=True) return Response(serializer.data, status=status.HTTP_200_OK) def post(self, request, room_id): user_data = request.data chatroom = get_object_or_404(ChatRoom, id=room_id) serializer = ChatMessageSerializer(data=user_data) # json -> objects if serializer.is_valid(): # serializer.save(chatroom) serializer.save(room=chatroom, sender=request.user) return Response(serializer.data, status=status.HTTP_200_OK) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
4. URL 라우팅 (urls.py)
-
chat/urls.py (new file)
from django.urls import path from .views import ChatRoomList, ChatMessageList, show_html urlpatterns=[ path('room/', ChatRoomList.as_view(), name='room-list'), path('<int:room_id>/messages', ChatMessageList.as_view(), name='message-list'), path('chatting', show_html, name='chatting') ]
5. Consumer (consumers.py)
consumers.py라는 이름을 사용하는 주된 이유는 Django Channels에서 'Consumer'라는 클래스를 사용하여 WebSocket과 같은 비동기 요청을 처리하기 때문이다.
이 파일명은 Django와 Channels 커뮤니티에서 널리 받아들여진 관례를 따랐다.
-
Consumer
: Django Channels에서 Consumer 클래스는 클라이언트와 서버 간의 WebSocket 연결을 관리한다. Consumer는 클라이언트로부터 메시지를 받고 (수신), 필요에 따라 메시지를 클라이언트에게 전송 (송신)하는 역할을 한다. 이는 전통적인 Django 뷰와 유사한 역할을 하지만, 비동기적인 특성을 가지고 있다. -
consumers.py파일
: 이 파일은 일반적으로 WebSocket 요청을 처리하기 위한 Consumer 클래스들을 포함한다.views.py파일이 HTTP 요청을 처리하는 데 사용되는 것처럼,consumers.py는 WebSocket과 같은 비동기 프로토콜을 처리하는 데 사용된다.
-
chat/consumers.py
from channels.generic.websocket import AsyncWebsocketConsumer import json class ChatConsumer(AsyncWebsocketConsumer): # 소켓 연결 async def connect(self): self.room_id = self.scope['url_route']['kwargs']['room_id'] self.room_group_name = 'chat_'+str(self.room_id) await self.channel_layer.group_add(self.room_group_name, self.channel_name) await self.accept() async def receive(self, text_data): text_data_json = json.loads(text_data) msg = text_data_json.get('message') await self.channel_layer.group_send(self.room_group_name, { 'type': 'chat_message', 'message': msg }) # 연결 해제 메서드: 클라이언트의 웹소켓 연결이 끊어졌을 때 호출됩니다. # group_discard: 클라이언트를 채팅방 그룹에서 제거합니다. 이를 통해 더 이상 메시지를 받지 않게 됩니다. async def disconnect(self, close_code): await self.channel_layer.group_discard(self.room_group_name, self.channel_name) async def chat_message(self, event): msg = event['message'] email = event['email'] await self.send(text_data=json.dumps({ 'type': 'chat.message', 'message': msg, 'email': email }))
6. DB migrate
모델 변경 사항에 대해 migrate한다.
> docker-compose run --rm app sh -c 'python manage.py makemigrations'
> docker-compose run --rm app sh -c 'python manage.py migrate'7. Front-End Part
다음은 채팅방 목록과 채팅방 내부에서 메시지를 주고받을 수 있는 기본적인 프론트엔드 구조다.
-
HTML : ex) chat.html
<!DOCTYPE html> <html> <head> <title>Chat Room</title> </head> <body> <h1>Chat Rooms</h1> <div id="room-container"> <!-- 채팅방 목록이 여기에 표시됩니다 --> </div> <h2>Chat Room: <span id="room-name"></span></h2> <div id="chat-log"> <!-- 메시지 로그가 여기에 표시됩니다 --> </div> <input id="chat-message-input" type="text" size="100"> <input id="chat-message-submit" type="button" value="Send"> <script src="chat.js"></script> <!-- 외부 스크립트 파일 참조 --> </body> </html> -
JavaScript
: WebSocket 연결 및 메시지 전송을 처리하는 JavaScript 코드<script> document.addEventListener('DOMContentLoaded', function() { fetchChatRooms(); // 페이지 로드 시 채팅방 목록 로드 }); function fetchChatRooms() { fetch('/api/chatrooms') .then(response => response.json()) .then(data => { const roomContainer = document.getElementById('room-container'); roomContainer.innerHTML = ''; // 초기화 data.forEach(room => { const roomDiv = document.createElement('div'); roomDiv.innerHTML = room.name; roomDiv.onclick = function() { enterChatRoom(room.id); }; roomContainer.appendChild(roomDiv); }); }) .catch(error => { console.error('Error fetching chat rooms:', error); }); } function enterChatRoom(roomId) { // 채팅방 메시지 로드 loadChatRoom(roomId); // WebSocket 연결 const chatSocket = new WebSocket( 'ws://' + window.location.host + '/ws/chat/' + roomId + '/' ); chatSocket.onmessage = function(e) { const data = JSON.parse(e.data); const chatLog = document.getElementById('chat-log'); chatLog.innerHTML += '<div>' + data.message + '</div>'; }; chatSocket.onclose = function(e) { console.error('Chat socket closed unexpectedly'); }; document.querySelector('#chat-message-submit').onclick = function(e) { const messageInputDom = document.querySelector('#chat-message-input'); const message = messageInputDom.value; chatSocket.send(JSON.stringify({ 'message': message })); messageInputDom.value = ''; }; } function loadChatRoom(roomId) { fetch(`/api/chatrooms/${roomId}/messages`) .then(response => response.json()) .then(messages => { const chatLog = document.getElementById('chat-log'); chatLog.innerHTML = ''; // 초기화 messages.forEach(message => { const messageDiv = document.createElement('div'); messageDiv.innerHTML = message.text; chatLog.appendChild(messageDiv); }); document.getElementById('room-name').innerText = roomId; }) .catch(error => { console.error('Error loading messages:', error); }); } </script>
[Youtube REST API 실습 후기]
docker-compose down / up / build차이점docker-compose run --rm app sh -c 'python manage.py makemigrations'중요성docker-compose run --rm app sh -c 'python manage.py migrate'중요성- test code 실행 중요성
docker-compose run --rm app sh -c 'python manage.py test [원하는 table]'
[참고 자료]
-
[오즈스쿨 스타트업 웹 개발 초격차캠프 백엔드 Django 강의]
백엔드 코린이😁
강의가 정말 알찬거 같네요,,! 오즈스쿨 어떤가요?!