DB 연동 / 모델 만들기

Docker-compose를 이용하여 MySQL과 Redis를 띄우고, 이를 연결하여 진행할 예정이다. redis 연결은 추후 인증 부분을 개발하면서 다룰 예정이므로, 오늘은 MySQL 연결부분을 다룬다.
사용한 docker-compsoe 파일은 아래와 같다.

version: '3'
name: ninja
services:
  redis:
    container_name: ninja-redis
    image: redis:7.0.11
    ports:
      - "6379:6379"

  mysql:
    container_name: ninja-db
    image: mysql:8.0.33
    environment:
      MYSQL_ROOT_PASSWORD: rootpassword
      MYSQL_DATABASE: ninjadb
      TZ: Asia/Seoul
    ports:
      - "3306:3306"
    depends_on:
      - redis

DB 연결

DB 연결 설정은 settings.py에서 진행한다.

1. pip install mysqlclient

위 패키지를 설치하는 중에 아래와 같은 오류가 발생했다.

위의 오류를 해결하기 위해, Github을 보고 해결을 시도했는데, Mac에서는 apt-get 대신 homebrew를 사용한다고 안내가 되어 있어 homebrew를 통해서 설정을 진행했다.

brew install mysql-client pkg-config
export PKG_CONFIG_PATH="$(brew --prefix)/opt/mysql-client/lib/pkgconfig"
python3 -m pip install mysqlclient

를 통해 pkg-config를 mysqlclient용으로 설치를 진행하고, mysqlclient를 성공적으로 설치할 수 있었다.

2. setting.py를 활용해서 Database 연결에 필요한 설정 진행

DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.mysql',
        'NAME': 'ninjadb',
        'USER' : 'root',
        'PASSWORD' : 'rootpassword',
        'HOST' : '127.0.0.1',
        'PORT' : '3306',
    }
}

위와 같이 setting.py 내의 DB 연결 정보를 sqlite에서 Mysql로 변경했다. Engine 정보도 변경해야함을 유의해야 할 것이다.

3. DB 감지

python3 manage.py inspectdb

위 과정을 통해서 DB를 감지하고 연결한다.

4. DB 테이블 생성

DB 테이블을 생성해보겠다. 나는 TODO API의 단순 CRUD를 진행할 것이기에, TODO 테이블을 생성했다.

CREATE TABLE `ninjadb`.`todo` (`todo_id` bigint NOT NULL AUTO_INCREMENT,`body` varchar(255) NOT NULL,`deadline` date,`is_finish` boolean NOT NULL, PRIMARY KEY (todo_id));

DDL이 아닌 DB Tool을 통해 생성해도 무방하다.

5. DB 감지

3번과 동일한 명령어를 입력하면 아래와 같이 모델 정보를 알려준다.

6. models.py 적용
   models.py를 적용하기 위해선 먼저 app을 만들어야 한다.

python3 manage.py startapp todo

를 통해 생성하면 아래와 같은 폴더구조를 가지게 된다.

GPT와 함께 열심히 씨름한 결과, 장고 닌자에서는 api.py를 사용하기에 아래와 같이 파일을 생성했다.

• api.py: 기존의 apps.py 대신 API 라우터로 활용
• schema.py: API 요청과 응답 간에서 사용되는 스키마 정의
• models.py: DB와 연결되는 모델 정의

앱을 새로 만들었을 경우, 앱 설정 및 api.py 관련 라우터 설정 진행
1. setting.py에 INSTALLED_APPS에 생성한 앱 정보 추가
2. api.py 기본 코드 작성

from ninja import NinjaAPI

api = NinjaAPI()

@api.get("/todo")
def get_example():
    return {"message": "Hello, Django Ninja!"}

models.py 내의 5번에서의 모델 정보 파일을 작성하면 된다.

from django.db import models

class Todo(models.Model):
    todo_id = models.BigAutoField(primary_key=True)
    body = models.CharField(max_length=255)
    deadline = models.DateField(blank=True, null=True)
    is_finish = models.IntegerField()

    class Meta:
        managed = False
        db_table = 'todo'

7. migrate 진행

python3 manage.py makemigrations
python3 manage.py migrate

DB 연결 후 최초의 migrate일 경우, Django에서 사용하는 테이블들이 자연스럽게 create되어 있을 것이다.

Django Ninja API 개발

Request

1. Query Parameter로 값 전달받기

@api.get("/hello")
def hello(request, name):
    return f"Hello {name}"

메소드 파라미터로 단순하게 선언만 해주면 된다. 단, 기본값이 없거나 형이 미일치해서 유효성 검증 문제가 발생할 수 있는데, 이를 최소한으로 대비하고자 name="world"와 같이 기본값을 설정할 수 있다.

특정 자료형으로 값이 들어오기 바란다면, name: str = "world"와 같이 지정할 수 있다. (Ex: a: int, b: int)

2. Path Variable로 값 전달받기

파이썬의 format-string과 동일한 포맷으로 값을 정의할 수 있다.

@api.get("/math/{a}and{b}")
def math(request, a: int, b: int):
    return {"add": a + b, "multiply": a * b}

위와 같이 {}내부에 변수명을 입력하고, 이를 연결할 수 있다.

3. Request Body로 값 전달받기

별도의 스키마를 정의해서 값을 바인딩해야 한다.

from ninja import NinjaAPI, Schema

api = NinjaAPI()

class HelloSchema(Schema):
    name: str = "world"

@api.post("/hello")
def hello(request, data: HelloSchema):
    return f"Hello {data.name}"

먼저, Schema 패키지를 import 해야하며, 위 코드의 HelloSchema와 같이 클래스를 정의해야 한다. 이를 전달받기 위해서는 "data: 작성한 스키마명"으로 연결시켜야 한다. 이후 접근할 때는 data.name과 같이 접근하여 사용할 수 있다.

Response

1. 응답 스키마 정의하기

class UserSchema(Schema):
    username: str
    is_authenticated: bool
    # Unauthenticated users don't have the following fields, so provide defaults.
    email: str = None
    first_name: str = None
    last_name: str = None

@api.get("/me", response=UserSchema)
def me(request):
    return request.user

@api.get에서 response 옵션에 반환할 응답 스키마를 지정함으로서, 바인딩될 수 있도록 한다.
여기서 return request.user의 의미는 장고의 user object에서 정의된 필드에 대한 값들을 가져온다는 의미이다. 인증이 되지 않은 유저의 경우 가져오지 않아야 하는 케이스는 기본값을 None으로 설정하여 구분한다.

2. 여러 종류의 응답이 가능할 경우

class UserSchema(Schema):
    username: str
    email: str
    first_name: str
    last_name: str

class Error(Schema):
    message: str

@api.get("/me", response={200: UserSchema, 403: Error})
def me(request):
    if not request.user.is_authenticated:
        return 403, {"message": "Please sign in first"}
    return request.user 

response 옵션에서 http status별로 적용되는 방식을 구분하여 표현한다.
200(성공)일 때는 1번의 방식과 같이 UserSchema를 반환하도록 하며, 403의 권한 인증 불가에는 Error Message를 반환하도록 한다.

Create

1. 요청/응답을 보내는 Dto 정의하기 (Schema.py)

from datetime import date
from ninja import Schema

# Request DTO
class TodoReq(Schema):
    body: str
    deadline: date = None
    is_finish: int = 0

# Response DTO
class TodoRes(Schema):
    todo_id: int
    body: str
    deadline: date = None
    is_finish: int

위에서 = 을 통해 정의하면 필수가 아닌 값이 되며, 미전달받을 경우 해당 값으로 대체 가능

2. 비즈니스 로직 작성 (api.py)

from ninja import Router
from .schema import TodoReq, TodoRes
from .models import Todo

router = Router()

@router.post("", response=TodoRes)
def post_todo(request, data: TodoReq):
    # Business Logic
    todo = Todo.objects.create(**data.dict())

    # Response
    return todo

JPA에 유사한 save()를 활용한 방식도 존재하지만, 나는 create방식의 저장 방법으로 작성했다. 이 둘의 차이에 대해서 간략하게 정리한 블로그 글을 첨부한다.
위에서 response부분에 스키마를 연동하면서 반환시 사용할 스키마를 연결했으며, 입력 시 받는 request body의 형태를 바인딩하기 위해서 TodoReq를 data라는 변수로 받도록 설정했다.
입력받은 요청 값을 모델로 변경해야 하기에

**data.dict()

을 사용하였다. 여기서 **는 Python에서 사용되는 unpacking 연산자로, 딕셔너리나 iterable 객체의 값을 키워드 인자로 전달할 때 사용한다고 한다. data값을 딕셔너리로 변환한 이후 키워드 인자로 반환하여 저장하게 된다.

Retrieve

위의 생성할 때 사용한 스키마를 그대로 이용할 것이다.
단, 가져올 때 사용하기 위해 아래 모듈을 import 하는 것을 추천한다.

from django.shortcuts import get_object_or_404

장고에서 제공하는 유틸리티 함수로, Object를 찾아서 가져오거나 없으면 404 에러를 반환한다.

@router.get("/{todo_id}", response=TodoRes)
def get_todo(request, todo_id: int):
    # Business Logic (특정 Object 한 개를 가져오거나 없을 경우, 404 에러 반환)
    todo = get_object_or_404(Todo, todo_id = todo_id) 

    # Response
    return todo

위와 같은 방식으로 사용할 수 있으며, todo_id = todo_id와 같은 부분에 조건이 추가하는 방식이다.
찾지 못했을 경우, 404 에러가 반환되는 것을 확인할 수 있다.

조건에 해당하는 값이 두 개 있다면 어떡하지?

위와 같이 1개의 값만 반환되지 않는다고 500 오류가 발생한다. 이를 유의해서 조건절을 추가해야 한다. 리스트 형식으로 가지고 오고 싶다면 filter를 사용하면 도니다. 이 때, filter 또한 가져온 행의 수가 0개일 경우 Http404 예외를 발생시키게 된다.

다른 방법으로는 못 가져올까?

1. Todo.objects.get(pk=1)

   • 위 방식은 사용자에게 404가 아닌 500에러를 반환한다

2. Todo.objects.filter(pk=1)

   • filter 방식은 쿼리셋으로 결과를 가져오기에 Schema로 변환하는 로직이 추가적으로 필요하다.

   # Response: Django 모델을 Pydantic 모델로 변환
   todo_res = TodoRes(
       todo_id=todo.todo_id,
       body=todo.body,
       deadline=todo.deadline,
       is_finish=todo.is_finish,
   )

Update

1. models.py에 값 업데이트 메소드 작성

def update(self, todo_req):
    self.body = todo_req.body
    self.deadline = todo_req.deadline
    self.is_finish = todo_req.is_finish

2. api.py에서 업데이트 로직 구현

@router.put("/{todo_id}", response=TodoRes)
def update_todo(request, todo_id: int, data: TodoReq):
    # Business Logic
    todo = get_object_or_404(Todo, todo_id=todo_id)
    todo.update(data)
    todo.save()

    # Response
    return todo

Delete

주어진 요청 정보에 맞는 행이 존재하는지 확인하고, delete() 메소드를 실행하면 된다.

@router.delete("/{todo_id}")
def delete_todo(request, todo_id:int):
    # Business Logic
    todo = get_object_or_404(Todo, todo_id=todo_id)
    todo.delete()

    # Response
    return JsonResponse({"meessage": "Success"})

Transaction 관리

장고는 기본적으로 auto commit을 사용하여, 별도의 명시가 없다면 쿼리에 대해서 성공 시, commit을 진행한다.
이를 메소드 단위로 묶어서 관리하고자 한다면 transaction에 대한 데코레이터를 붙여서 활용해야 한다. 이 때, try except를 통해서 rollback이 될 수 있도록 관리해주어야 한다.

@transaction.atomic()

위와 같은 데코레이터를 메소드 위에 추가해주면 된다. 추가적으로, 세부적인 작업들을 관리하고자 한다면 () 내부의 옵션들을 설정해주면 된다.

내부 옵션
1. using:

• 설명: 트랜잭션을 적용할 데이터베이스를 지정합니다.
• 예시: using='default'

2. savepoint:

   • 설명: 중첩된 트랜잭션 내에서 저장점을 사용할지 여부를 결정합니다. 기본값은 False로 중첩된 트랜잭션에서 저장점을 사용하지 않습니다.
   • 예시: savepoint=True

3. force_insert:

   • 설명: 새로운 레코드를 추가할 때, 강제로 INSERT 쿼리를 사용하도록 설정합니다.
   • 예시: force_insert=True

4. force_update:

   • 설명: 기존 레코드를 업데이트할 때, 강제로 UPDATE 쿼리를 사용하도록 설정합니다.
   • 예시: force_update=True

5. autocommit:

   • 설명: 트랜잭션을 사용할 때, autocommit 모드를 설정합니다. False로 설정하면 트랜잭션 모드가 활성화되어 수동으로 커밋 또는 롤백을 수행해야 합니다.
   • 예시: autocommit=False

6. when:

   • 설명: 트랜잭션을 언제 시작할지 정의합니다. True로 설정하면 함수 실행 전에 트랜잭션을 시작하고, False로 설정하면 함수 실행 후에 트랜잭션을 시작합니다.
   • 예시: when=True

QuerySet

Django에서 QuerySet은 데이터베이스로부터 데이터를 조회하고 조작하는데 사용되는 객체 집합을 나타낸다. QuerySet은 데이터베이스 쿼리를 생성하고 실행하여 결과를 가져온다
GPT와 함께 찾은 QuerySet에 대한 몇 가지 중요한 특징과 사용법은 아래와 같다.

1. 조회하기 (Querying):

   todos = Todo.objects.filter(is_finish=True)

   filter() 메서드는 특정 조건에 맞는 객체를 조회합니다. get() 메서드는 하나의 객체만 반환하며, 해당 조건에 맞는 객체가 없거나 여러 개라면 예외가 발생합니다.

2. 체이닝 (Chaining):

   recent_todos = Todo.objects.filter(is_finish=False).order_by('-deadline')[:5]

   filter()나 exclude()와 같은 메서드를 여러 번 체이닝하여 복잡한 쿼리를 작성할 수 있습니다.

3. 값 추출 (Values):

   todo_values = Todo.objects.filter(is_finish=False).values('id', 'body', 'deadline')

   values() 메서드를 사용하여 특정 필드만 추출할 수 있습니다.

4. 집계 함수 (Aggregation):

   from django.db.models import Count
   todo_count = Todo.objects.filter(is_finish=False).aggregate(num_todos=Count('id'))

   aggregate() 메서드를 사용하여 집계 함수를 수행할 수 있습니다.

5. 연결된 객체 검색 (Related Objects):

   author_todos = Todo.objects.filter(author__name='John')

   연결된 객체의 필드를 사용하여 검색할 수 있습니다.

6. 인덱싱 및 슬라이싱:

   first_todo = Todo.objects.filter(is_finish=False)[0]
   recent_todos = Todo.objects.filter(is_finish=False)[:5]

   QuerySet은 인덱싱과 슬라이싱이 가능하며, 이를 통해 특정 범위의 결과를 가져올 수 있습니다.

7. 업데이트 (Update):

   Todo.objects.filter(is_finish=False).update(is_finish=True)

   update() 메서드를 사용하여 데이터베이스의 레코드를 일괄적으로 업데이트할 수 있습니다.

8. 삭제 (Delete):

   Todo.objects.filter(is_finish=True).delete()

   delete() 메서드를 사용하여 데이터베이스의 레코드를 삭제할 수 있습니다.

QuerySet은 실제로 데이터베이스에 쿼리를 전송하는 것이 아니라, 필요한 순간까지 지연 평가(lazy evaluation)되므로 효율적으로 사용할 수 있습니다. 필요한 시점에만 데이터베이스에 쿼리를 전송하여 결과를 가져오게 됩니다.

최종 코드 (api.py)

from ninja import Router
from django.db import transaction
from django.shortcuts import get_object_or_404
from django.http import JsonResponse, Http404

from .schema import TodoReq, TodoRes
from .models import Todo

router = Router()

@router.post("", response=TodoRes)
@transaction.atomic()
def post_todo(request, data: TodoReq):
    try:
        # Business Logic
        todo = Todo.objects.create(**data.dict())

        # Response
        return todo
    except Exception as e:
        return JsonResponse({"message": str(e)}, status = 500)


@router.get("/{todo_id}", response=TodoRes)
@transaction.atomic()
def get_todo(request, todo_id: int):
    try:
        # Business Logic (특정 Object 한 개를 가져오거나 없을 경우, 404 에러 반환)
        todo = get_object_or_404(Todo, todo_id=todo_id) 
        # todo = Todo.objects.get(todo_id=todo_id, body="string1")
        # todo_list = Todo.objects.filter(todo_id=todo_id)[:1]

        # if not todo_list.exists():
        #     raise JsonResponse({"message": "Not found Error"})

        # todo_res = TodoRes(
        #     todo_id=todo_list[0].todo_id,
        #     body=todo_list[0].body,
        #     deadline=todo_list[0].deadline,
        #     is_finish=todo_list[0].is_finish
        # )

        # Response
        return todo
    except Http404 as e:
        return JsonResponse({"message": str(e)}, status = 404)
    except Exception as e:
        return JsonResponse({"message": str(e)}, status = 500)
    

@router.put("/{todo_id}", response=TodoRes)
@transaction.atomic()
def update_todo(request, todo_id: int, data: TodoReq):
    try:
        # Business Logic
        todo = get_object_or_404(Todo, todo_id=todo_id)
        todo.update(data)
        todo.save()

        # Response
        return todo
    except Exception as e:
        return JsonResponse({"message": str(e)}, status = 500)
    

@router.delete("/{todo_id}")
@transaction.atomic()
def delete_todo(request, todo_id:int):
    try:
        # Business Logic
        todo = get_object_or_404(Todo, todo_id=todo_id)
        todo.delete()

        # Response
        return JsonResponse({"meessage": "Success"})
    except Exception as e:
        return JsonResponse({"message": str(e)}, status = 500)
    

추가로 공부해봐야할 것

• BaseModel을 통한 논리적 삭제 방식 구현
• ManyToOne, OneToMany를 이용한 연관 관계 설정
• 전역 공통 응답 객체 정의 방법
• 에러 핸들링 (try except를 활용한 예외 처리를 진행하고 JSONResponse로 응답 객체를 계속 만들어야 되는 것인지)
• 직접 쿼리를 입력하여 값을 가져오기 위해서 어떤 식으로 구현할 수 있을지
• 장고의 단위 테스트/통합 테스트
• DTO에 대한 네이밍 컨벤션(Req/Res or In/Out)
• Pydantic과 Schema의 차이 및 데이터 유효성 검사

레퍼런스

• 장고 닌자 공식 문서
• 장고 DB 연결
• 장고 DB 연결2
• 장고 DB 연결3
• 장고 닌자 CRUD API
• 장고 닌자 기본 개발 레포 참고
• 장고 트랜잭션