django ORM을 사용해서 MySQL에 get, post

코드를 작성하다보면 항상 같은 명령어를 찾아보는 것 같아서, 미리 블로그에 포스팅하면 굳이 찾아보는 수고를 하지 않을 것 같다.

 

1. MySQL DB 생성

CREATE DATABASE 데이터베이스명 CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci;
SHOW DATABASES;
USE 데이터베이스명;

 

2. 장고 앱 생성

django-admin startapp 앱이름

 

3.앱 디렉토리에 위치한 models.py 수정

from django.db import models


class User(models.Model):
    username      = models.CharField(max_length=45)
    password      = models.CharField(max_length=256)
    email_address = models.CharField(max_length=255, unique=True)
    phone_number  = models.CharField(max_length=50, null=True)
    profile_image = models.CharField(max_length=1000, null=True)
    profile_bio   = models.CharField(max_length=500, null=True)
    created_at    = models.DateTimeField(auto_now_add=True)
    updated_at    = models.DateTimeField(auto_now=True)

    class Meta:
        db_table = "users"

 

 4. 메인 디렉토리에 있는 urls.py 수정

urlpatterns = [
    path('users', include('users.urls'))
]

 

5. 앱 디렉토리에 있는 urls.py 생성 후 아래와 같이 작성

from django.urls import path
from users.views import SignUpView

urlpatterns=[
    path('/users',SignUpView.as_view())
]

 

6. 앱 디렉토리에 있는 views.py 작성

- 비밀번호는 bcrypt를 사용하여 암호화

- jwt를 사용하여 토큰을 발행하고, 추후에 데코레이터를 사용하여 인증을 수행

import json
import datetime

import jwt
import bcrypt

from django.views           import View
from django.http            import JsonResponse, Http404
from django.db.utils        import IntegrityError
from django.conf            import settings
from django.core.exceptions import (PermissionDenied,
                                    EmptyResultSet)
from users.models           import User
from users.validator        import (validate_password,
                                   validate_email)

from westagram.checkitem    import CheckItem
class SignUpView(View):
    def post(self, request):
        try:
            data = json.loads(request.body)

            signup_key_list=['username', 'email', 'password',]
            CheckItem.check_keys_in_body(data, signup_key_list)

            username     = data.get('username')
            email        = data.get('email')
            password     = data.get('password')
            phone_number = data.get('phone_number')

            validate_email(email)
            validate_password(password)

            if User.objects.filter(email_address=email).exists():
                raise IntegrityError("EMAIL_DUPLICATED_ERROR")

            hashed_password = bcrypt.hashpw(
                    password.encode('utf-8'),
                    bcrypt.gensalt()
            )

            user = User.objects.create(
                        username      = username,
                        password      = hashed_password.decode(),
                        email_address = email,
                        phone_number  = phone_number,
            )

            return JsonResponse({"message":"SUCCESS"}, status=201)

        except KeyError as e:
            #return JsonResponse({"message":str(e.message)}, status=400)
            return JsonResponse({"message":getattr(e, 'message', str(e))}, status=400)

        except IntegrityError as e:
            return JsonResponse({"message":getattr(e, 'message', str(e))}, status=409)


class SignInView(View):
    def post(self, request):
        try:
            data = json.loads(request.body)

            signin_key_list=["email", "password"]
            CheckItem.check_keys_in_body(data, signin_key_list)

            email    = data.get("email")
            password = data.get("password")

            user = User.objects.get(email_address=email)

            #if not (user.exists()):
            #    return JsonResponse({'message' : 'INVALID_USER'}, status=401)

            valid_password = user.password.encode('utf-8')

            if not bcrypt.checkpw(password.encode('utf-8'), valid_password):
                return JsonResponse({'message' : 'INVALID_USER'}, status=401)

            exp_date = datetime.datetime.utcnow() + datetime.timedelta(hours=2)
            access_token = jwt.encode(
                    {'id': user.id, 'exp':exp_date},
                    settings.SECRET_KEY,
                    algorithm=settings.ALGORITHM,
            )

            return JsonResponse({"message":"SUCCESS", "token":access_token}, status=200)

        except KeyError as e:
            return JsonResponse({"message":str(e.message)}, status=400)

        except User.DoesNotExist:
            return JsonResponse({'message':'INVALID_USER'}, status=401)

 

7. login_required 데코레이터를 작성하여 토큰이 유효한지 확인

import jwt

from users.views import User
from django.http import JsonResponse
from django.conf import settings

SECRET_KEY = settings.SECRET_KEY
ALGORITHM  = settings.ALGORITHM

def login_required(func):
    def wrapper(self, request, *args, **kwargs):
        try:
            access_token = request.headers.get("Authorization", None)
            payload      = jwt.decode(access_token, SECRET_KEY, algorithms=ALGORITHM)
            user         = User.objects.get(id=payload.get('id'))
            request.user = user
            return func(self, request, *args, **kwargs)
        except KeyError as e:
            return JsonResponse({'message': getattr(e,'message',str(e))},status=401)

        except jwt.ExpiredSignatureError:
            return JsonResponse({'message':'TOKEN_EXPIRED'}, status=400)

        except User.DoesNotExist:
            return JsonResponse({'message':'INVALID_USER'}, status=400)

    return wrapper