FastAPI를 이용한 웹 서비스 구현 연습_4

Frye 'de Bacon·2023년 10월 25일
FastAPIpythonsvelte

FastAPI를 이용한 웹 구현

목록 보기
4/12

본 시리즈는 '박응용' 님의 '점프 투 FastAPI'를 바탕으로 학습 및 실습한 내용을 정리한 것입니다.

구현 및 파인튜닝한 모델을 사용한 웹 서비스 구현을 위해 FastAPI의 학습 필요성을 느껴 학습 과정을 정리합니다. 내용의 정확성이나 이론적인 부분은 당연히 원본 페이지를 참조하시는 게 좋고, 본 시리즈에서는 구현 도중 발생하는 문제 등을 해결하는 과정을 함께 기록하여 '처음부터 끝까지 따라 할 수 있는' 시리즈를 만드는 것을 목표로 합니다(물론 제1목표는 학습 내용 기록입니다).

이번 포스트에서는 질문 목록의 조회 기능을 구현하면서 다음과 같은 FastAPI의 핵심 기능에 대해 알아본다.

  • 라우터(Router)
  • 의존성 주입(Dependency Injection)
  • Pydantic을 이용한 입출력 관리
  • CRUD 파일 작성

본격적인 실습에 아퍼 FastAPI 서버와 Svelte 서버를 실행하고 정상적으로 실행되는지를 확인한다. 잘 실행했다면 Svelte 서버에 접속했을 때 "안녕하세요, FasAPI"라는 메시지가 출력될 것이다.

1. 라우터(Router)

라우터 만들기

이제부터 할 일은 위의 화면 대신 게시판 질문 목록이 출력되도록 하는 것이다. 가장 먼저 질문 목록 API를 만들어야 한다. 앞서 시리즈 두 번째 포스트에서 언급했듯이, domain 디렉토리의 하위에 question 디렉토리를 만들어 관리도록 한다.

우선 \practice_1\domain\ 아래에 question 디렉토리를 만들고 그 아래에 다시 question_router.py 파일을 생성한다.

생성한 라우터 파일에 다음과 같이 코드를 입력한다.

from fastapi import APIRouter

from database import SessionLocal
from models import Question

router = APIRouter(
    prefix="/api/question",
)

@router.get('/list')
def question_list():
    db = SessionLocal()
    _question_list = db.query(Question).order_by(Question.create_date.desc()).all()
    db.close()
    return _question_list

라우터 파일에는 APIRouter 클래스로 생성한 router 객체가 반드시 존재해야 한다. 그리고 이 router 객체를 FastAPI 앱에 등록함으로써 라우팅 기능을 동작시킨다.
이때 라우팅이란 FastAPI가 요청받은 URL을 해석하여 그에 맞는 함수를 실행하고 그 결과를 리턴하는 행위를 말한다.

router 객체 생성 시 사용한 prefix 속성은 요청 url에 항상 포함되어야 하는 값이다. 즉, '/api/question/list'라는 url 요청이 발생하면 '/api/question'이라는 prefix가 등록된 question_router.py 파일에서 '/list'로 등록된 함수 question_list가 실행되는 것이다.

question_list는 db 세션을 생성한 뒤 해당 세션을 이용해 db에서 질문 목록을 조회하여 리턴하는 함수이다. 이때 사용한 세션은 db.close()를 이용해 커넥션 풀에 반환한다.
※ 커넥션 풀에 대한 자세한 설명은 여기에서 잘 설명해주고 있다.

이제 생성한 router 객체를 FastAPI 앱에 등록하자. 이는 main.py 파일을 통해 이루어진다.

...
from domain.question import question_router
...
# @app.get("/hello")
# def hello():
#     return {"message": "안녕하세요, FastAPI"}  <- 삭제
app.include_router(question_router.router)

위와 같이 hello API는 삭제하고, include_router 메서드를 이용해 app 객체에 question_router.py 파일의 router 객체를 등록한다.

질문 목록 API 테스트

이제 작성한 질문 목록 API를 테스트해보자. FastAPI의 docs 문서를 이용해 간단하게 테스트할 수 있다.

앞선 포스트에서 작성한 질문 데이터가 반환되는 것을 확인할 수 있다.

2. 의존성 주입(Dependency Injection)

이제 질문 목록 API를 만들었으니 프론트엔드에서 이 API를 호출하여 결괏값을 화면에 출력할 수 있다. 그런데 그 전에 question_list 함수를 다시 살펴보자.

@router.get('/list')
def question_list():
    db = SessionLocal()
    _question_list = db.query(Question).order_by(Question.create_date.desc()).all()
    db.close()
    return _question_list

보면 db 세션 객체를 생성한 후 함수 종료 직전에 다시 db.close()를 호출하여 커넥션 풀에 db 세션을 반환하는 방식을 취하고 있다. 우리가 만들 대부분의 API는 DB를 사용하며, 따라서 이러한 패턴이 반복될 것이다. 이때, FastAPI의 의존성 주입(Dependency Injection)을 사용하면 이 부분을 자동화할 수 있다.

먼저 database.py 파일에 다음과 같은 함수를 선언한다.

import contextlib
...
@contextlib.contextmanager
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

db 세션 객체를 리턴하는 제너레이터인 get_db 함수를 선언하였고, 여기에 @contextlib.contextmanager 어노테이션을 적용하여 with문과 함께 사용할 수 있도록 하였다. with문을 벗언는 순간 get_db 함수의 finally에 작성된 db.close()가 자동으로 실행되도록 하는 방식이다.
※ 제너레이터와 contextmanager에 대해서는 추가로 알아보는 것이 좋겠다.

이제 question_list 함수도 위에서 작성한 get_db 함수를 사용하도록 변경하자.

...
from database import get_db  # SessionLocal은 사용하지 않으므로 지운다.
...
@router.get('/list')
def question_list():
    with get_db() as db:
        _question_list = db.query(Question).order_by(Question.create_date.desc()).all()
    return _question_list

이렇게 하면 오류 여부와 상관없이 with 문을 벗어나는 순간 db.close()가 실행되므로 보다 안전한 코드가 된다.

Depends 사용하기

FastAPI의 Depends를 사용하면 with문을 사용하는 것보다 더 간단하게 사용할 수 있다.

from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
...
def question_list(db: Session = Depends(get_db)):
    _question_list = db.query(Question).order_by(Question.create_date.desc()).all()
    return _question_list

get_db 함수를 with문과 함께 사용하는 대신 question_list 함수의 매개변수로서 db: Session = Depends(get_db) 객체를 주입받으며, Depends는 매개변수로 전달받은 함수를 실행시킨 결과를 리턴한다. 따라서 db: Session = Depends(get_db)의 db 객체에는 get_db 제너레이터에 의해 생성된 세션 객체가 주입된다.
이때 get_db 함수에 자동으로 contextmanager가 적용되므로(Depends에서 이를 적용하도록 설계되어 있다) database.py의 get_db 함수에서 어노테이션을 제거해야 한다.

import contextlib  # 삭제
...
@contextlib.contextmanager  # 삭제
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

수정 후 질문 목록 API를 FastAPI의 docs에서 테스트하면 이전과 같이 문제 없이 작동되는 것을 확인할 수 있다.

3. 스키마

현재 질문 목록 API가 반환하는 출력값은 Question 모델의 모든 항목이다. 그런데 Question 모델에는 외부로 공개되면 안 되는 출력값이 있을 수도 있으며, 경우에 따라 출력값이 정확한지 검증할 필요가 있을 수도 있다. 그런데 현재와 같은 형태로는 이런 조건을 충족할 수 없으며, 따라서 출력 부분에 추가적인 코딩이 필요하다.
이때 사용할 수 있는 라이브러리 Pydantic이다.

Pydantic

Pydantic은 FastAPI의 입출력 스펙을 정의하고 그 값을 검증하는 데 사용하는 라이브러리로서, FastAPI와 함께 설치된다. Pydantic은 입출력 항목을 다음과 같이 정의하고 검증할 수 있다.

  • 입출력 항목의 개수와 타입 설정
  • 입출력 항목의 필수 값 체크
  • 입출력 항목의 데이터 검증

질문 목록 API의 출력 부분에 Pydantic을 적용해 보자.
이를 위해 가장 먼저 할 일은 질문 목록 API의 출력 스키마를 생성하는 것이다. 이때 스키마란 일반적으로 데이터의 구조 및 명세를 의미한다. 즉, 출력 스키마는 출력 항목이 몇 개인지, 제약 조건은 무엇이 있는지 등을 기술하는 것이다.

Pydantic schema 작성하기

질문과 관련된 스키마이므로 question 도메인 디렉토리에 question_schema.py 파일을 생성하여 관리한다. 생성한 파일에 다음과 같이 코드를 작성한다.

import datetime
from pydantic import BaseModel

class Question(BaseModel):
    id: int
    subject: str
    content: str
    create_date: datetime.datetime

위와 같이 BaseModel을 상속한 Question 클래스를 만들었다(이를 Question 스키마라 하며, models.py에 정의한 Question 클래스는 Question 모델이라 한다). Question 스키마에서는 총 4개의 출력 항목을 정의하고 그 타입을 지정하였다. 예를 들어 id: int는 id 속성에 integer 자료형만 대입 가능하다는 의미이며, 정해진 타입이 아닌 다른 타입의 자료형이 대입되면 오류가 발생한다. 또한 디폴트 값을 설정하지 않았으므로 필수 항목임을 의미한다.
만약 다음과 같이 설정하면 필수 항목이 아니게 된다.

subject: str | None = None

이는 subject 항목이 문자열 또는 None을 자료로 가질 수 있으며, 디폴트 값은 None이라는 의미이다.

라우터에 Pydantic 적용하기

이제 작성한 Question 스키마를 질문 목록 라우터 함수에 적용해 보자. question_router.py 파일을 다음과 같이 수정한다.

...
from domain.question import question_schema
...
@router.get('/list', response_model=;ist[question_schema.Question])
def question_list(db: Session = Depends(get_db)):
    _question_list = db.query(Question).order_by(Question.create_date.desc()).all()
    return _question_list

여기서 추가된 'response_model=list[question_schema.Question]'는 question_list 함수의 리턴값이 Question 스키마로 구성된 리스트임을 의미한다.

그러나 이렇게 수정한 후 FastAPI의 docs 문서에서 테스트하면 오류가 발생한다. 이는 리턴값인 _question_list의 요소값이 딕셔너리가 아닌 Question 모델이기 때문이며, 따라서 다음처럼 question_schema.py를 수정하여 orm_mode 항목을 True로 설정하면 해결된다.

import datetime
from pydantic import BaseModel

class Question(BaseModel):
    id: int
    subject: str
    content: str
    create_date: datetime.datetime

    class Config:
        orm_mode = True

...고 하는데 버전이 올라가면서 변경된 것인지 나는 오류가 나지 않았다. 그리고 orm_mode 역시 pydantic 버전 2부터는 from_attributes로 변경되었다. 따라서 다음과 같이 수정하였다.

import datetime
from pydantic import BaseModel

class Question(BaseModel):
    id: int
    subject: str
    content: str
    create_date: datetime.datetime

    class Config:
        from_attributes = True

4. CRUD

현재 질문 목록 라우터 함수에는 데이터를 조회하는 다음 부분이 포함되어 있다.

_question_list = db.query(Question).order_by(Question.create_date.desc()).all()

상기 코드가 문제가 있는 것은 아니지만, 서로 다른 라우터에서 데이터를 처리하는 부분이 동일하여 중복될 수 있으므로 이렇게 데이터를 처리하는 부분을 question_crud.py 파일에 분리하여 작성한다.
※ 이는 필수 사항은 아니며 프로젝트의 성격에 맞게 구현하면 된다.

question 디렉토리에 question_crud.py 파일을 작성하고 다음 코드를 입력한다.

from models import Question
from sqlalchemy.orm import Session

def get_question_list(db: Session):
    question_list = db.query(Question).order_by(Question.create_date.desc()).all()
    return question_list

이는 질문 목록 라우터 함수에 있던 내용을 그대로 옮긴 것이다. 이제 작성한 get_question_list를 질문 목록 라우터 함수에서 사용할 수 있도록 수정하자.

...
from domain.question import question_schema, question_crud
from models import Question  # 삭제
...
@router.get('/list', response_model=list[question_schema.Question])
def question_list(db: Session = Depends(get_db)):
    _question_list = question_crud.get_question_list(db)
    return _question_list
profile
Frye 'de Bacon
AI, NLP, Data analysis로 나아가고자 하는 개발자 지망생
이전 포스트

FastAPI를 이용한 웹 서비스 구현 연습_3

다음 포스트

FastAPI를 이용한 웹 서비스 구현 연습_5

0개의 댓글