FastAPI 를 사용한 파이썬 웹 개발

이 글은 압둘라지즈 압둘라지즈 아데시나가 지은 FastAPI 를 사용한 파이썬 웹 개발 의 책을 보고 공부한 내용입니다.

우선 Go 를 해보고 난 후, 고민을 많이했다. 많이 쓰이는 노드나 파이썬 중에 하나를 제대로 파보고 싶었다.
이미 패캠에 노드강의를 끊어놓은게 하나 있지만, 노드는 자바스크립트에 익숙하지 않은 내게.. 타입스크립트, ES6, 7, 8 까지 진입장벽이 너무 높아보였다..

그리고 저번에 주섬 (필자가 최근에 한 프로젝트) 회식 때 넷이 언어에 대한 얘기를 했는데, 그때 xx 님이 생각하는 자바스크립트의 방향성에 대한 얘기를 듣고, 집가는 길에 꽤 많이 생각하게 되었다. (다른 언어를 많이 따라하는 경향에 대한 이야기.. 정체성이라고 해야하나..)

또, 해커톤에 참여해 빠르게 개발해보고 싶었다. Go 문법은 모두 알지만, 수요 (Go 로 하려는 분들) 도 잘 없고, 레퍼런스도 잘 없고.. (구글에 영문으로 찾아야함.. Swift 처럼..)
알고리즘을 Python 으로 하고 있으니까 뭔가 파이썬으로 하면 빠르게 개발할 수 있을 것 같았다. (사람 구하기도 쉽고!)

그리고 빅데이터를 활용한 앱 개발쪽으로도 배워보고 싶었다.

그래서 결국 Python 을 공부하기로 결정했고, Django, Flask, FastAPI 중에 골라야했다.
우선 가볍게 FastAPI 로 입문하기로 결정, 교보문고에서 책을 사오는 길이다.

목표는 한달정도로 잡았다. 꽤 얇았다..

시작해봅시다.

Ch1. FastAPI 소개

처음엔 git 에 대한 내용을 설명해준다. 가볍게 읽어본다.

그리고 .gitignore.io 사이트를 추천해준다.

.gitignore.io: https://www.toptal.com/developers/gitignore/

책에서는 vertualenv 환경에서 개발하지만 이는 OS 에 독립적이진 않다고 한다.

우선 책대로 진행하고 나중에 도커환경에서 개발하는 방법을 적용해보자.

블로그: https://mingrammer.com/setup-the-python-development-environment-with-pycharm-and-docker/

mkdir todos && cd todos
python3 -m venv venv

폴더가 생겼다.

파이썬2와 파이썬3 모두 설치되어 있으면 python3 라고 버전을 명시해야 한다. 반대로 파이썬 2가 설치되어 있지 않고 파이썬3 만 있다면 python 을 실행해도 python3 가 실행된다.

venv 모듈은 가상 환경을 설치할 폴더명을 인수로 지정한다.

생성된 가상 환경 폴더 (venv) 에는 파이썬 인터프리터가 설치된 lib 폴더와 가상 환경 내에서 상호 작용이 필요한 파일을 저장하는 bin 폴더가 있다.

가상환경 활성화/비활성화

source venv/bin/activate

활성화 시 이렇게 앞에 venv 가 붙는다.

가상환경 비활성화

deactivate

pipenv 또는 poetry 으로도 가상 환경을 생성하고 애플리케이션의 의존 라이브러리를 관리할 수 있다.

가상 환경이 만들어졌으니 pip 을 사용한 패키지 관리 방법을 알아보자.

1.3 pip 을 사용한 패키지 관리

pip 은 파이썬 패키지 관리 도구로, 자바스크립트에서 사용되는 yarn 과 유사하다.

pip 설치 확인

python3 -m pip list

기본 명령

꼭 가상환경 안에서 설치할 것!

pip install fastapi # 설치
pip uninstall fastapi
pip freeze > requirements.txt

fastapi 를 설치 후 목록에 잘 나온다.

현재 프로젝트에 설치된 모든 패키지 목록을 파일로 저장하려면 다음과 같이 freeze 명령과 > 연산자를 사용하면 된다.

> 연산자는 왼쪽 명령의 실행 결과를 오른쪽 파일에 저장한다.

pip install -r requirements.txt

다음으로 도커에 관해 알아보자.

1.4 도커 설정

도커파일

FROM python:3.8
WORKDIR /usr/src/app
ADD . /usr/src/app
CMD ["python", "hello.py"]

도커는 아니까 간단히 읽어보자.

1.5 간단한 FastAPI 애플리케이션 개발

라이브러리 2개가 필요하다.

• fastapi
• uvicorn: 애플리케이션을 실행하기 위한 비동기 방식 서버 게이트웨이 인터페이스

이제 api.py 라는 파일을 만들어 FastAPI 의 새 인스턴스를 생성한다.

api.py

from fastapi import FastAPI

app = FastAPI()

@app.get('/')
async def welcome() -> dict:
    return {
        'message': 'Hello world'
    }

uvicorn api:app --port 8000 --reload

요청이 잘 가는 것을 볼 수 있다.

uvicorn 을 실행할 때 지정하는 인수는 다음과 같다.

• file:instance: FastAPI 인스턴스가 존재하는 파이썬 파일과 FastAPI 인스턴스를 가지고 있는 변수를 지정한다.
• --port PORT: 애플리케이션에 접속할 수 있는 포트 번호를 지정한다.
• --reload: 선택적 인수로, 파일이 변경될 때 마다 애플리케이션을 재시작한다.

CHAPTER 2 에서는 FastAPI 의 라우팅을 자세히 다룬다. 먼저 pydantic 을 사용해 요청 페이로드와 응답을 검증하는 모델을 개발한다.

그런 다음 경로 매개변수와 쿼리 매개변수, 요청 바디를 학습하고 CRUD 처리가 가능한 todo 애플리케이션을 만든다.

Ch2. 라우팅

라우팅은 클라이언트가 서버로 보내는 HTTP 요청을 처리하는 프로세스다.

CHAPTER 2 에서는 APIRouter 인스턴스를 사용해 라우트를 생성하는 방법과 메인 FastAPI 애플리케이션에 접속하는 방법을 다룬다. 또한 모델이 무엇인지 소개하고 이를 사용해 요청 바디를 검증한다. 마지막으로 경로 및 쿼리 매개변수를 FastAPI 애플리케이션에 적용하는 방법도 학습한다.

2.1 FastAPI 의 라우팅

HTTP 요청 메서드: https://developer.mozilla.org/ko/docs/Web/HTTP/Methods

2.2 APIRouter 클래스를 사용한 라우팅

todo.py

from fastapi import APIRouter

todo_router = APIRouter()

todo_list = []

@todo_router.post('/todo')
async def add_todo(todo: dict) -> dict:
    todo_list.append(todo)
    return {
        'message': 'Todo added successfully.'
    }
@todo_router.get('/todo')
async def retrieve_todos() -> dict:
    return {
        'todos': todo_list
    }

api.py

from fastapi import FastAPI
from todo import todo_router

app = FastAPI()

@app.get('/')
async def welcome() -> dict:
    return {
        'message': 'Hello world'
    }
app.include_router(todo_router)

APIRouter 클래스는 FastAPI 클래스와 동일한 방식으로 작동한다. 하지만 uvicorn 은 APIRouter() 인스턴스를 사용해서 애플리케이션을 실행할 수 없다. FastAPI() 인스턴스에 추가해야 외부에서 접근 가능하다.

잘 저장되고 조회되는 것을 볼 수 있다.

2.3 pydantic 모델을 사용한 요청 바디 검증

모델은 pydantic 의 BaseModel 클래스의 하위 클래스로 생성된다.

pydantic 은 파이썬의 타입 어노테이션을 사용해서 데이터를 검증하는 파이썬 라이브러리다.

model.py

from pydantic import BaseModel

class Todo(BaseModel):
    id: int
    item: str

POST 요청의 파라미터 값을 모델로 바꿔준다.

todo.py

from model import Todo

@todo_router.post('/todo')
async def add_todo(todo: Todo) -> dict:
    todo_list.append(todo)
    return {
        'message': 'Todo added successfully.'
    }

id 와 item 이 없다면 필드가 필요하다고 에러를 던지게 된다.

모델과 일치하는 데이터를 보내자 정상적인 응답을 받게 된다.

중첩 모델

pydantic 모델은 다음과 같이 중첩해서 정의할 수 있다.

class Item(BaseModel)
    item: str
    status: str

class Todo(BaseModel)
    id: int
    item: Item

이제 경로 매개변수와 쿼리 매개변수를 알아보자.

2.4 경로 매개변수와 쿼리 매개변수

여기서는 하나의 tdoo 작업만 추출하는 새로운 라우트를 만든다. 먼저 todo 의 ID 를 경로 매개변수에 추가하자.

경로 매개변수

todo.py

@todo_router.get('/todo/{todo_id}')
async def get_syngle_todo(todo_id:int) -> dict:
    for todo in todo_list:
        if todo.id == todo_id:
            return {
                'todo': todo
            }
    return {
        'message': "Todo with supplied ID dosen't exist."
    }

다음 코드에는 Path 라는 클래스가 추가된다. Path 는 FastAPI 가 제공하는 클래스로, 라우트 함수에 있는 다른 인수와 경로 매개변수를 구분하는 역할을 한다. Path 클래스는 스웨거와 ReDoc 등으로 OpenAPI 기반 문서를 자동 생성할 때 라우트 관련 정보를 함께 문서화하도록 돕는다.

todo.py

from fastapi import APIRouter, Path

@todo_router.get('/todo/{todo_id}')
async def get_syngle_todo(todo_id:int = Path(..., title='The ID of the todo to retrieve.')) -> dict:

Path 클래스는 첫 인수로 None 또는 ... 을 받을 수 있다. 첫 번째 인수가 ... 이면 경로 매개변수를 반드시 지정해야 한다. 또한 경로 매개변수가 숫자이면 수치 검증을 위한 인수를 지정할 수 있다. 예를 들어 gt (greater than, ~보다 큰), le (less than, ~보다 작은) 와 같은 검증 기호를 사용할 수 있다. 이를 통해 경로 매개변수에 사용된 값이 특정 범위에 있는 수자인지 검증 가능하다.

쿼리 매개변수

쿼리 매개변수는 선택 사항이며 보통 URL 에서 ? (물음표) 뒤에 온다. 제공된 쿼리를 기반으로 특정한 값을 반환하거나 요청을 필터링 할 때 사용된다.

async query_route(query: str = Query(None):
    return query

쿼리 매개변수의 사용법은 이 책의 후반부에서 todo 보다 복잡한 애플리케이션을 만들 때 다시 살펴볼 것이다.

FastAPI 자동 문서화

문서는 다음 두 가지 유형으로 제공된다.

• 스웨거
• ReDoc

뒤에 /docs 로 접근하면 자동으로 생성된 스웨거 문서를 볼 수 있다.

문서: http://localhost:8000/docs

마찬가지로 /redoc 을 붙이면 ReDoc 문서를 볼 수 있다.

JSON 스키마를 올바르게 생성하기 위해 사용자가 입력해야 할 데이터의 샘플을 설정할 수 있다. 샘플 데이터는 모델 클래스 안에 Config 클래스로 정의하면 된다. 다음과 같이 Todo 모델 클래스에 샘플 데이터를 추가해보자.

model.py

class Todo(BaseModel):
    id: int
    item: str

    class Config:
        json_schema_extra = {
            'example': {
                'id': 1,
                'item': 'Example Schema!',
            }
        }

책처럼 schema_extra 로 작성하면,

schema_extra 가 json_schema_extra 로 바뀌었다고 알려준다.

샘플 데이터 값이 바뀌었다.

2.6 간단한 CRUD 애플리케이션 개발

앞서 todo 아이템을 추가하고 추출할 수 있는 라우트를 만들었다. 이번에는 기존 아이템을 변경하거나 삭제하는 라우트를 추가해볼 것이다. 다음 단계를 따라 실습해보자.

1. UPDATE 라우트 요청 바디용 모델을 model.py 에 추가

class TodoItem(BaseModel):
    item: str

    class Config:
        json_schema_extra = {
            'example': {
                'item': 'Read the next chapter of the book.',
            }
        }

2. todo 를 변경하기 위한 라우트를 todo.py 에 추가

from model import Todo, TodoItem

@todo_router.put('/todo/{todo_id}')
async def update_todo(todo_data: TodoItem, todo_id: int = Path(..., title='The ID of the todo to be updated.')) -> dict:
    for todo in todo_list:
        if todo.id == todo_id:
            todo.item = todo_data.item
            return {
                'message': "Todo updated successfully.."
            }
    return {
        'message': "Todo with supplied ID dosen't exist."
    }

3. 추가

4. PUT 요청으로 수정

5. 아이템 변경됐는지 확인

1번은 Example Schema! 라는 데이터이다.

잘 수정이 되었다.

6. todo.py 에 삭제를 위한 DELETE 라우트를 추가해보자.

@todo_router.delete('/todo/{todo_id}')
async def delete_single_todo(todo_id: int) -> dict:
    for index in range(len(todo_list)):
        todo = todo_list[index]
        if todo.id == todo_id:
            todo_list.pop(index)
            return {
                'message': "Todo deleted successfully."
            }
    return {
        'message': "Todo with supplied ID dosen't exist."
    }
@todo_router.delete('/todo')
async def delete_all_todo() -> dict:
    todo_list.clear()
    return {
        'message': 'Todos deleted successfully.'
    }

7. 추가

8. 삭제

9. GET 요청으로 확인

잘 삭제가 되었다.

Ch3. 응답 모델과 오류 처리

CHAPTER 3 에서 다루는 내용은 다음과 같다.

• FastAPI 의 응답
• 응답 모델 작성
• 오류 처리

3.1 FastAPI 의 응답

API 응답은 보통 JSON 또는 XML 형식이지만 문서 형식으로 전달되기도 하며 헤더와 바디로 구성된다.

상태코드

• 1XX: 요청을 받았다.
• 2XX: 요청을 성공적으로 처리했다.
• 3XX: 요청을 리다이렉트했다.
• 4XX: 클라이언트 측에 오류가 있다.
• 5XX: 서버 측에 오류가 있다.

HTTP 상태 코드: https://ko.wikipedia.org/wiki/HTTP

3.2 응답 모델 작성

모든 todo 를 추출해서 배열로 반환하는 라우트를 ID 없이 todo 아이템만 반환하도록 변경해보자. 먼저 model.py 에 다음과 같이 새로운 모델을 추가한다.

from typing import List

class TodoItems(BaseModel):
    todos: List[TodoItem]

    class Config:
        json_schema_extra = {
            'example': {
                'todos': [
                    {
                        'item': 'Example schema 1!'
                    },
                    {
                        'item': 'Example schema 2!'
                    },
                ]
            }
        }

todo.py

from model import Todo, TodoItem, TodoItems

@todo_router.get('/todo', response_model=TodoItems)
async def retrieve_todos() -> dict:
    return {
        'todos': todo_list
    }

이제 id 가 노출되지 않는다.

3.3 오류 처리

FastAPI 에서 오류는 FastAPI 의 HTTPException 클래스를 사용해 예외를 발생시켜 처리한다.

HTTPException 클래스는 다음 세 개의 인수를 받는다.

• status_code: 예외 처리 시 반환할 상태 코드
• detail: 클라이언트에게 전달한 메시지
• headers: 헤더를 요구하는 응답을 위한 선택적 인수

todo.py 에서 추출, 변경, 삭제 라우트를 다음과 같이 변경하자.

from fastapi import APIRouter, Path, HTTPException, status 

@todo_router.get('/todo/{todo_id}')
async def get_syngle_todo(todo_id:int = Path(..., title='The ID of the todo to retrieve.')) -> dict:
    for todo in todo_list:
        if todo.id == todo_id:
            return {
                'todo': todo
            }
    raise HTTPException(
        status_code=status.HTTP_404_NOT_FOUND,
        detail="Todo with supplied ID dosen't exist.", 
    )

@todo_router.put('/todo/{todo_id}')
async def update_todo(todo_data: TodoItem, todo_id: int = Path(..., title='The ID of the todo to be updated.')) -> dict:
    for todo in todo_list:
        if todo.id == todo_id:
            todo.item = todo_data.item
            return {
                'message': "Todo updated successfully.."
            }
    raise HTTPException(
        status_code=status.HTTP_404_NOT_FOUND,
        detail="Todo with supplied ID dosen't exist.",
    )

@todo_router.delete('/todo/{todo_id}')
async def delete_single_todo(todo_id: int) -> dict:
    for index in range(len(todo_list)):
        todo = todo_list[index]
        if todo.id == todo_id:
            todo_list.pop(index)
            return {
                'message': "Todo deleted successfully."
            }
    raise HTTPException(
        status_code=status.HTTP_404_NOT_FOUND,
        detail="Todo with supplied ID dosen't exist.",
    )

이제 응답이 404 가 떨어진다.

@todo_router.post('/todo', status_code=201)
async def add_todo(todo: Todo) -> dict:
    todo_list.append(todo)
    return {
        'message': 'Todo added successfully.'
    }

CHAPTER 4 에서는 Jinja 를 사용한 FastAPI 애플리케이션 템플릿 생성 방법을 배운다. Jinja 템플릿의 실행 및 설정 방법을 익힌 다음 템플릿 지식을 활용해 UI 를 만드는 방법을 알아보자.

Ch4. 템플릿팅

4.1 Jinja

Jinja 는 파이썬으로 작성된 템플릿팅 엔진으로, API 응답을 쉽게 렌더링할 수 있도록 한다.

Jinja 는 중괄호 {} 를 사용해서 템플릿 파일의 일반적인 HTML, 텍스트 등을 표현식 및 구문과 구분한다.

{{}} 구문을 변수 블록 (variable block) 이라고 하며 이 안에 변수를 지정한다.

{% %} 는 if/else, 반복, 매크로 같은 구조를 제어할 때 사용된다.

자주 사용되는 구문

• {% ... %}: 구조를 제어하기 위한 명령
• {{todo.item}}: 식의 값 전달
• {# 이 책은 훌륭한 API 책이다! #}: 주석을 기입할 때 사용되며 웹 페이지상에는 표시되지 않는다.

필터

필터는 다음과 같이 파이프 기호 (|) 를 사용해서 변수와 구분하며 괄호를 사용해 선택적 인수를 지정한다.

{{ variable | filter_name(*args) }}

인수가 없다면 다음과 같이 정의해도 된다.

{{ variable | filter_name}}

자주 사용되는 필터를 살펴보자.

기본 필터

전달된 값이 None 일 때 사용할 값을 지정한다.

{{ todo.item | default('이것은 기본 todo 아이템입니다.' }}

이스케이프 필터

HTML 을 변환하지 않고 그대로 렌더링한다.

{{ "<title>Todo Application</title>" | escape }}
<title>Todo Application</title>

변환 필터

데이터 유형을 변환한다.

{{ 3.142 | int}}
3

{{ 31 | float }}
31.0

병합 필터

리스트 내의 요소들을 병합해서 하나의 문자열로 만든다.

{{ ['한빛미디어는'', '훌륭한', '책을', '만든다.'] | join(' ') }}
한빛미디어는 훌륭한 책을 만든다.

길이 필터

전달된 객체의 길이를 반환한다. 파이썬의 len() 과 같은 역할을 한다.

Todo count: {{ todos | length }}
Todo count: 4

Jinja 가 제공하는 전체 필터를 보고 싶다면 웹 페이지를 참고하자.

if문

Jinja 의 if문은 파이썬과 사용법이 유사하며 {% %} 제어 블록 내에서 사용할 수 있다.

{% if todo | length < 5 %}
    할 일 목록에 할 일이 많지 않네요.
{% else %}
    바쁜 날을 보내고 있군요!
{% endif %}

반복문

Jinja 에서는 변수를 사용해서 반복 처리를 할 수 있다. 다음과 같이 리스트 또는 일반적인 함수를 사용할 수도 있다.

{% for todo in todos %}
    <b> {{ todo.item }} </b>
{% endfor %}

반복문 내에서 특수한 변수를 사용할 수도 있다. 예를 들어 loop.index 는 현재 인덱스를 반환하는 특수 변수다. [표 4-1] 에서 특수 변수를 정리했다.

변수	설명
loop.index
loop.index0
loop.revindex
loop.revindex0
loop.first
loop.last
loop.length
loop.cycle
loop.depth
loop.depth0
loop.previtem
loop.nextitem
loop.changed(*val)

매크로

Jinja 의 매크로는 하나의 함수로, HTML 문자열을 반환한다. 매크로 사용의 주요 목적은 하나의 함수를 사용해 반복적으로 작성하는 코드를 줄이는 것이다. 예를 들어 입력 (input) 매크로를 정의해서 HTML 폼에 반복적으로 정의하는 입력 태그를 줄일 수 있다.

{% macro input(name, value='', type='text', size=20 %}
    <div class="form">
        <input type="{{ type }}" name="{{ name }}"
        value="{{ value|escape }}" size="{{ size }}">
    </div>
{% endmacro %}

이 매크로를 호출해서 폼에 사용할 입력 요소를 간단하게 만들 수 있다.

{{ input('item') }}

이것은 다음과 같은 HTML 을 반환한다.

<div class="form">
    <input type="text" name="item" value="" size="20">
</div>

다음으로 템플릿 상속이 무엇인지, FastAPI 에서 어떻게 사용되는지 살펴보자.

템플릿 상속

Jinja 의 가장 강력한 기능은 템플릿 상속이다. 이 기능은 중복 배제 (Don't Repeat Yourself, DRY) 원칙에 근거한 것이며 큰 규모의 웹 애플리케이션을 개발할 때 많은 도움이 된다. 템플릿 상속 (template inheritance) 은 기본 템플릿을 정의한 다음 이 템플릿을 정의한 다음 이 템플릿을 자식 템플릿이 상속하거나 교체해서 사용할 수 있게 한다.

4.2 FastAPI 에서 Jinja 를 사용하는 방법

Jinja 를 사용하려면 Jinja2 패키지를 설치하고 기존 작업 디렉터리에 template 이라는 신규 폴더를 만들어야 한다. 이 폴더에 모든 Jinja 관련 파일 (Jinja 구문이 섞여 있는 HTML 파일) 이 저장된다. 이 책에서는 사용자 인터페이스 (user interface, UI) 디자인을 다루지 않으므로 스타일을 직접 작성하지 않고 CSS 부트스트랩 라이브러리를 활용한다.

pip install jinja2 python-multipart
mkdir templates
cd templates
touch {home,todo}.html

• 애플리케이션 홈 페이지용 home.html
• todo 페이지용 todo.html

[그림 4-1] 에서 가장 바깥쪽 사각형이 home 템플릿 (홈 페이지) 이고 안쪽의 작은 사각형이 todo 템플릿 (todo 페이지) 이다.

대충 구현한 것 같다..

Ch5. 구조화

CHAPTER 5 에서 다루는 내용은 다음과 같다.

• 애플리케이션 라우트와 모델 구조화
• 플래너 API 용 모델 구현

5.1 FastAPI 애플리케이션 구조화

이제부터 이벤트 플래너를 만들어볼 것이다.

mkdir planner && cd planner
touch main.py
mkdir database routes models

touch {database,routes,models}/__init__.py

다음 명령으로 database 폴더에 database.py 라는 빈 파일을 만든다.

이 파일은 데이터베이스 추상화와 설정에 사용되는 파일로, <CHAPTER 6 데이터베이스 연결> 에서 쓰인다.

touch database/connection.py

touch {routes,models}/{events,users}.py

각 파일의 역할은 다음과 같다.

• routes 폴더
  • events.py: 이벤트 생성, 변경, 삭제 등의 처리를 위한 라우팅
  • users.py: 사용자 등록 및 로그인 처리를 위한 라우팅
• models 폴더
  • events.py: 이벤트 처리용 모델을 정의
  • users.py: 사용자 처리용 모델을 정의

API 를 구조화해서 비슷한 파일을 기능에 따라 그룹화했다.

이벤트 플래너 애플리케이션 개발

python3 -m venv venv
source venv/bin/activate

pip install fastapi uvicorn "pydantic[email]"
pip freeze > requirements.txt

pydantic[email] 은 pydantic 과 함께 email-validator 라는 의존 라이브러리를 함께 설치하라는 의미다.

필요한 라이브러리가 포함된 개발 환경이 모두 준비됐다. 이어서 애플리케이션 모델을 구현해보자.

모델 구현

models/events.py

from pydantic import BaseModel
from typing import List

class Event(BaseModel):
    id: int
    title: str
    image: str
    description: str
    tags: List[str]
    location: str

    class Config: 
        json_schema_extra = {
            'example': {
                'title': 'FastAPI Book Launch', 
                'image': 'https://linktomyimage.com/image.png',
                'description': 'We will be discussing the contents of the FastAPI book in this event...',
                'tags': ['python','fastapi','book','launch'],
                'location': 'Google Meet',
            }
        }

• id: 자동 생성되는 고유 식별자
• title: 이벤트 타이틀
• image: 이벤트 이미지 배너의 링크
• description: 이벤트 설명
• tags: 그룹화를 위한 이벤트 태그
• location: 이벤트 위치

models/users.py

from pydantic import BaseModel, EmailStr
from typing import List, Optional
from models.events import Event 

class User(BaseModel):
    email: EmailStr
    password: str
    events: Optional[List[Event]] = None 

    class Config:
        json_schema_extra = {
            'example': {
                'email': 'fastapi@packt.com',
                'username': 'strong!!!',
                'events': [],
            }
        }

6. 사용자 로그인 모델 (UserSignIn) 을 만든다.

class UserSignIn(BaseModel):
    email: EmailStr
    password: str

    class Config:
        json_schema_extra = {
            'example': {
                'email': 'fastapi@packt.com',
                'password': 'strong!!!',
                'events': [],
            }
        }

이것으로 모델을 모두 완성했다. 다음으로 라우트를 구현해보자.

라우트 구현

API 라우트 시스템을 구현할 차례다. 먼저 사용자 라우트 시스템을 설계해보자. 사용자 라우트는 로그인, 로그아웃, 등록으로 구성된다. 인증을 완료한 사용자는 이벤트를 생성, 변경, 삭제할 수 있으며, 인증을 거치지 않은 사용자는 생성된 이벤트를 확인하는 것만 가능하다. [그림 5-2] 는 사용자 라우트와 이벤트 라우트의 관계를 나타낸다.

각 라우트를 자세히 살펴보자.

• /user
  • signup
  • signin
  • signout
• /event
  • /new
  • /, {id}
  • /{id}
  • /{id}

왠 엔드포인트에 new? 앞에 restful 에 대해 설명하지 않았었나..

사용자 라우트

1. 등록 (/signup) 라우트를 정의해보자.

routes/users.py

from fastapi import APIRouter, HTTPException, status 
from models.users import User, UserSignIn

user_router = APIRouter(
    tags=['User'],
)
users = {}

@user_router.post('/signup')
async def sign_new_user(data: User) -> dict:
    if data.email in users:
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail='User with supplied username exists'
        )
    users[data.email] = data
    return {
        'message': 'User successfully registered!'
    }

2. 로그인 (/signin) 라우트를 다음과 같이 정의한다.

@user_router.post('/signin')
async def sign_user_in(user: UserSignIn) -> dict:
    if user.email not in users:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail='User does not exist'
        )
    if users[user.email].password != user.password:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail='Wrong credentials passed'
        )
    return {
        'message': 'User signed in successfully'
    }

3. 사용자 처리용 라우트를 정의했으니 main.py 에 라우트를 등록하고 애플리케이션 실행해보자. 라이브러리와 사용자 라우트 정의를 임포트한다.

main.py

from fastapi import FastAPI
from routes.users import user_router

import uvicorn

app = FastAPI()
app.include_router(user_router, prefix='/user')

if __name__ == '__main__':
    uvicorn.run('main:app', host='127.0.0.1', port=8000,reload=True)

준비가 다 됐으면 애플리케이션을 실행해보자.

python main.py

서버가 잘 실행되었다.

6. 애플리케이션이 잘 실행됐으니 사용자 라우트를 테스트해보자. 사용자 등록부터 테스트한다.

7. 이번에느느 로그인 라우트를 테스트해보자.

비밀번호를 틀리면 다른 메시지가 전달되는지 확인해보자.

스웨거 기반 문서를 통해 라우트를 확인할 수 있다.

사용자 라우트를 모두 구현했다. 다음으로 이벤트 처리용 라우트를 구현해보자.

이벤트 라우트

from fastapi import APIRouter, Body, HTTPException, status
from models.events import Event
from typing import List

event_router = APIRouter(
    tags=['Events'],
)
events = []

@event_router.get('/', response_model=List[Event])
async def retrieve_event(id: int) -> Event:
    for event in events:
        if event.id == id:
            return event
    raise HTTPException(
        status_code=status.HTTP_404_NOT_FOUND,
        detail='Event with supplied ID does not exist'
    )
@event_router.post('/new')
async def create_event(body: Event = Body(...)) -> dict:
    events.append(body)
    return {
        'message': 'Event created successfully.'
    }

@event_router.delete('/{id}')
async def delete_event(id: int) -> dict:
    for event in events:
        if event.id == id:
            events.remove(event)
            return {
                'message': 'Event deleted successfully.'
            }
    raise HTTPException(
        status_code=status.HTTP_404_NOT_FOUND,
        detail='Event with supplied ID does not exist'
    )

@event_router.delete('/')
async def delete_all_events() -> dict:
    events.clear()
    return {
        'message': 'Event deleted successfully.'
    }

4. main.py 의 라우트 설정을 변경해서 이벤트 라우트를 추가하자.

http :8000/event/new id=1 title='FastAPI Book Launch' image=https://linktomyimage.com/image.png description='We will be discussing the contents of the FastAPI book in this event.Ensure to come with your own copy to win gifts!' location='Google Meet' tags:='["python","fastapi","book","launch"]'

• GET 라우트: 다음 명령을 실행하여 생성된 이벤트를 추출해보자.

명령을 실행하면 앞서 추가한 이벤트가 그대로 표시된다.

• DELETE 라우트: 다음 명령을 실행하여 이벤트를 삭제해보자.

정상적으로 요청이 처리되면 다음과 같은 메시지가 표시된다.

이것으로 이벤트 플래너 애프리케이션의 라우트와 모델을 성공적으로 구현했으며 테스트를 통해 제대로 실행되는지도 확인했다.

Ch6. 데이터베이스 연결

CHAPTER 6 에서 다루는 내용은 다음과 같다.

• SQLModel 설정
• SQLModel 을 사용한 SQL 데이터베이스의 CRUD 처리
• 몽고DB 설정
• beanie 를 사용한 몽고DB 의 CRUD 처리

CHAPTER 6~8 의 실습을 진행하려면 몽고DB 를 설치해야 한다. 설치 방법은 몽고 DB 공식 사이트에서 운영체제에 맞는 문서를 참고하면 된다.

6.1 SQLModel 설정

SQLModel 라이브러리는 FastAPI 개발자가 만들었으며 pydantic 과 SQLAlchemy 를 기반으로 한다. pydantic 을 통해 모델을 쉽게 정의할 수 있다는 것은 이미 <CHAPTER 3 응답 모델과 오류 처리> 에서 배웠다.

새로운 브랜치를 만든다.

git checkout -b planner-sql

source venv/bin/activate
pip install sqlmodel

SQLModel 을 사용해서 테이블을 생성하려면 테이블 모델 클래스를 먼저 정의해야 한다. pydantic 모델처럼 테이블을 정의하지만 이번에는 SQLModel 서브 클래스로 정의한다. 클래스 정의에는 table 이라는 설정 변수를 갖는다. 이 변수를 통해 해당 클래스가 SQLModel 테이블이라는 것을 인식한다.

class Event(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    title: str
    image: str
    description: str
    location: str
    tags: List[str]

로우

new_event = Event(title='Book Launch',
    image='src/fastapi.png',
    description='The book launch event will be held at Packt HQ, Packt city',
    location='Google Meet',
    tags=['packt','book'])

세션 클래스를 사용해서 데이터베이스 트랜잭션을 만든다.

with Session(engine) as session:
    session.add(new_event)
    session.commit()

세션 클래스가 무엇인지, 어떻게 작동하는지부터 알아보자.

세션

세션 객체는 코드와 데이터베이스 사이에서 이루어지는 처리를 관리하며 주로 특정 처리를 데이터베이스에 적용하기 위해 사용된다. Session 클래스는 SQL 엔진의 인스턴스를 인수로 사용한다.

여기서 사용하는 Session 클래스의 메서드는 다음과 같다.

• add(): 처리 대기 중인 데이터베이스 객체를 메모리에 추가한다. 앞서 살펴본 코드에서 new_event 객체는 세션 메모리에 추가되고 commit() 메서드에 의해 데이터베이스에 등록 (커밋) 될 때까지 대기했다.
• commit(): 현재 세션에 있는 트랜잭션을 모두 정리한다.
• get(): 데이터베이스에서 단일 로우를 추출한다. 모델과 문서 ID 를 인수로 사용한다.

6.2 데이터베이스 생성

SQLModel 에서는 SQLAlchemy 엔진을 사용해서 데이터베이스를 연결한다. SQLAlchemy 엔진은 create_engine() 메서드를 사용해서 만들며 SQLModel 라이브러리에서 임포트한다.

database_file = 'database.db'
engine = create_engine(database_file, echo=True)
SQLModel.metadata.create_all(engine)

create_all() 메서드는 데이터베이스뿐만 아니라 테이블도 생성한다. 중요한 점은 데이터베이스 연결 파일 (connection.py) 에서 테이블 파일을 임포트해야 한다는 것이다.

models/events.py

from sqlmodel import JSON, SQLModel, Field, Column
from typing import Optional, List

class Event(SQLModel, table=True):
    id: int = Field(default=None, primary_key=True)
    title: str
    image: str
    description: str
    tags: List[str] = Field(sa_column=Column(JSON))
    location: str

    class Config:
        arbitrary_types_allowed = True
        json_schema_extra = {
            'example': {
                'title': 'FastAPI Book Launch',
                'image': 'https://linktomyimage.com/image.png',
                'description': 'We will be discussing the contents of the FastAPI book in this event...',
                'tags': ['python','fastapi','book','launch'],
                'location': 'Google Meet',
            }
        }
class EventUpdate(SQLModel):
    title: Optional[str] = None
    image: Optional[str] = None
    description: Optional[str] = None
    tags: Optional[List[str]] = None
    location: Optional[str] = None

    class Config:
        json_schema_extra = {
            'example': {
                'title': 'FastAPI Book Launch',
                'image': 'https://linktomyimage.com/image.png',
                'description': 'We will be discussing the contents of the FastAPI book in this event...',
                'tags': ['python','fastapi','book','launch'],
                'location': 'Google Meet',
            }
        }

이 코드는 기존 모델 클래스를 SQL 테이블 클래스로 변경한다.

3. connection.py 에 데이터베이스 및 테이블 생성을 위한 설정을 작성한다.

database/connection.py

from sqlmodel import SQLModel, Session, create_engine
from models.events import Event

database_file = 'planner.db'
database_connection_string = f'sqlite:///{database_file}'
connect_args = {'check_same_thread': False}
engine_url = create_engine(database_connection_string, echo=True, connect_args=connect_args)

def conn():
    SQLModel.metadata.create_all(engine_url)

def get_session():
    with Session(engine_url) as session:
        yield session

conn() 함수는 SQLModel 을 사용해서 데이터베이스와 테이블을 생성하고 get_session() 을 사용해서 데이터베이스 세션을 애플리케이션 내에서 유지한다.

4. main.py 를 다음과 같이 변경하여 애플리케이션이 시작될 때 데이터베이스를 생성하도록 한다.

main.py

from fastapi import FastAPI
from routes.users import user_router
from routes.events import event_router
from database.connection import conn

import uvicorn

app = FastAPI()

# 라우트 등록
app.include_router(user_router, prefix='/user')
app.include_router(event_router, prefix='/event')

@app.on_event('startup')
def on_startup():
    conn()

if __name__ == '__main__':
    uvicorn.run('main:app', host='127.0.0.1', port=8000,reload=True)

터미널에서 애플리케이션을 실행하면 데이터베이스와 테이블이 생성됐다는 메시지가 출력된다.

event 라는 테이블을 생성했다.

db 도 만든 것을 볼 수 있다.

데이터베이스를 생성했으니 데이터베이스를 사용하도록 CRUD 처리 라우트를 변경해보자.

이벤트 생성

routes/events.py 를 변경해서 이벤트 테이블 클래스와 get_session() 함수를 임포트한다.

get_session() 함수를 통해 라우트가 세션 객체에 접근할 수 있다.

from fastapi import APIRouter, Depends, HTTPException, Request, status
from database.connection import get_session
from models.events import Event, EventUpdate

@event_router.post('/new')
async def create_event(new_event: Event,
                       session=Depends(get_session)) -> dict:
    session.add(new_event)
    session.commit()
    session.refresh(new_event)

    return {
        'message': 'Event created successfully.'
    }

Depends

Depends 클래스는 FastAPI 애플리케이션에서 의존성 주입을 담당한다. 이 클래스는 함수를 인수로 사용하거나 함수 인수를 라우트에 전달할 수 있게 해서 어떤 처리가 실행되든지 필요한 의존성을 확보해준다.

신규 이벤트 생성을 담당하는 POST 라우트를 다음과 같이 변경한다.

등록 request

http :8000/event/new id=1 title='FastAPI Book Launch' image=https://linktomyimage.com/image.png description='We will be discussing the contents of the FastAPI book in this event.Ensure to come with your own copy to win gifts!' location='Google Meet' tags:='["python","fastapi","book","launch"]'

이벤트 조회

전체 이벤트를 추출하는 GET 라우트를 변경해서 데이터베이스에서 데이터를 가져오도록 만들어보자. 다음과 같이 routes/events.py 파일을 변경하면 된다.

@event_router.get('', response_model=List[Event])
async def retrieve_all_events(session=Depends(get_session)) -> List[Event]:
    statement = select(Event)
    events = session.exec(statement).all()
    return events

@event_router.get('/{id}', response_model=Event)
async def retrieve_event(id: int, session=Depends(get_session)) -> Event:
    event = session.get(Event, id)
    if event:
        return event

    raise HTTPException(
        status_code=status.HTTP_404_NOT_FOUND,
        detail='Event with supplied ID does not exist'
    )

이벤트가 잘 응답한다.

이벤트 변경

@event_router.put('/edit/{id}', response_model=Event)
async def update_event(id:int, new_data:EventUpdate,session=Depends(get_session)) -> Event:
    event = session.get(Event, id)
    if event:
        event_data = new_data.dict(exclude_unset=True)
        for key, value in event_data.items():
            setattr(event, key, value)
        session.add(event)
        session.commit()
        session.refresh(event)

        return event

    raise HTTPException(
        status_code=status.HTTP_404_NOT_FOUND,
        detail='Event with supplied ID does not exist'
    )

title 이 잘 수정되었다.

이벤트 삭제

events.py 파일에서 기존 DELETE 라우트를 다음과 같이 변경한다.

@event_router.delete('/{id}')
async def delete_event(id: int, session=Depends(get_session)) -> dict:
    event = session.get(Event, id)
    if event:
        session.delete(event)
        session.commit()
        return {
            'message': 'Event deleted successfully.'
        }

    raise HTTPException(
        status_code=status.HTTP_404_NOT_FOUND,
        detail='Event with supplied ID does not exist'
    )

책과는 다르게 /delete/{id} 대신 /{id} 로 작성했다.

잘 삭제되었다.

지금까지 SQLModel 을 사용해서 애플리케이션과 데이터베이스를 연결하고 CRUD 처리를 구현했다. 작성한 코드를 커밋하자.

이어서 몽고 DB 를 연동하여 CRUD 처리를 구현해보자.

6.3 몽고DB 설정

FastAPI 와 몽고DB 를 연결해주는 몇 가지 도구가 존재하지만 여기서는 beanie 를 사용한다.

beanie 는 비동기 객체 문서 매퍼 (Object Document Mapper, ODM) 로, 데이터베이스 처리를 담당한다.

다음 명령을 사용해서 beanie 를 설치한다.

pip install beanie==1.13.1

데이터베이스 연동에 앞서 beanie 의 기능과 데이터베이스 테이블 생성 방법을 알아보자.

문서

from beanie import Document 

class Event(Document):
    name: str
    location: str

    class Settings:
        name = 'events'

여기서 Settings 서브 클래스는 몽고DB 데이터베이스 내에 설정한 이름으로 컬렉션을 생성한다.

문서 생성 방법을 알았으니 CRUD 처리를 위한 메서드를 살펴보자.

• insert(), create(): 문서 인스턴스에 의해 호출되며 데이터베이스 내에 새로운 레코드를 생성한다. 단일 데이터는 insert_one() 메서드를 사용해 추가하고 여러 개의 데이터는 insert_many() 메서드를 사용해 추가한다.
• find(), get(): find() 메서드는 인수로 지정한 문서를 목록에서 찾는다. get() 메서드는 지정한 ID 와 일치하는 단일 문서를 반환한다. find_one() 메서드는 다음과 같이 지정한 조건과 일치하는 단일 문서를 반환한다.
• save(), update(), upsert(): save() 메서드는 데이터를 신규 문서로 저장할 때 사용된다. update() 메서드는 기존 문서를 변경할 때 사용되고, upsert() 메서드는 조건에 부합하는 문서가 없으면 신규로 추가할 때 사용된다. 여기서는 update() 메서드를 사용하며 다음과 같이 변경용 쿼리를 지정한다.
• delete(): 데이터베이스에서 문서를 삭제한다. 사용 방법은 아래와 같다.

insert 예시

event1 = Event(name='Packt office launch', location='Hybrid')
event2 = Event(name='Hanbit office launch', location='Hybrid')
await event1.create()
await event2.create()
await Event.insert_many([event1, event2])

find 예시

event = await Event.get('74478287284ff')
event = await Event.find(Event.location == 'Hybrid').to_list()
event = await Event.find_one(Event.location == 'Hybrid')

save 예시

event = await Event.get('74478287284ff')
update_query = {'$set': {'location': 'virtual'}}
await event.update(update_query)

이 코드는 변경하고자 하는 쿼리를 추출한 후 해당 문서의 location 필드를 virtual (온라인) 로 변경한다.

delete 예시

event = await Event.get('74478287284ff')
await event.delete()

지금까지 beanie 라이브러리가 제공하는 메서드를 살펴봤다. 이제 이벤트 플래너 애플리케이션에 몽고DB 를 설정하고 문서를 정의해보자.

데이터베이스 초기화

1. database 폴더에 connection.py 라는 파일을 만든다.

pydantic 의 BaseSettings 부모 클래스를 사용해서 설정 변수를 읽을 수 있다. 웹 API 개발할 때는 설정 변수를 하나의 환경 파일에 저장하는 것이 좋다.

2. connection.py 에 다음 코드를 추가한다.

from beanie import init_beanie
from motor.motor_asyncio import AsyncIOMotorClient
from typing import Optional
from pydantic import BaseSettings

class Settings(BaseSettings):
    DATABASE_URL: Optional[str] = None

    async def initialize_database(self):
        client = AsyncIOMotorClient(self.DATABASE_URL)
        await init_beanie(database=client.get_default_database(), document_models=[])

    class Config:
        env_file = '.env'

3. models 폴더의 모델 파일을 변경하여 몽고DB 문서를 사용할 수 있도록 만들자.

models.events.py

from beanie import Document
from typing import Optional, List

class Event(Document):
    title: str
    image: str
    description: str
    tags: List[str]
    location: str

    class Config:
        json_schema_extra = {
            'example': {
                'title': 'FastAPI Book Launch',
                'image': 'https://linktomyimage.com/image.png',
                'description': 'We will be discussing the contents of the FastAPI book in this event...',
                'tags': ['python','fastapi','book','launch'],
                'location': 'Google Meet',
            }
        }
    class Settings:
        name = 'events'

models.users.py

from pydantic import BaseModel, EmailStr
from typing import Optional, List
from beanie import Document, Link
from pydantic import BaseModel, EmailStr
from models.events import Event

class User(Document):
    email: EmailStr
    password: str
    events: Optional[List[Event]] = None

    class Settings:
        name = 'users'

    class Config:
        json_schema_extra = {
            'example': {
                'email': 'fastapi@packt.com',
                'username': 'strong!!!',
                'events': [],
            }
        }

class UserSignIn(BaseModel):
    email: EmailStr
    password: str

    class Config:
        json_schema_extra = {
            'example': {
                'email': 'fastapi@packt.com',
                'password': 'strong!!!',
                'events': [],
            }
        }

connection.py

from beanie import init_beanie
from motor.motor_asyncio import AsyncIOMotorClient
from typing import Optional
from pydantic import BaseSettings
from models.users import User
from models.events import Event

class Settings(BaseSettings):
    DATABASE_URL: Optional[str] = None

    async def initialize_database(self):
        client = AsyncIOMotorClient(self.DATABASE_URL)
        await init_beanie(database=client.get_default_database(), document_models=[Event, User])

    class Config:
        env_file = '.env'

7. 환경 파일 (.env) 을 생성한 다음 데이터베이스 URL 을 추가하면 데이터베이스 초기화 과정이 끝난다.

touch .env
echo DATABASE_URL=mongodb://localhost:27017/planner >> .env

데이터베이스 초기화 작업이 모두 끝났다. 다음으로 CRUD 처리를 구현해보자.

6.4 CRUD 처리

connection.py 파일에 다음과 같이 새로운 Database 클래스를 추가한다. 이 클래스는 초기화 시 모델을 인수로 받는다.

from beanie import init_beanie, PydanticObjectId
from motor.motor_asyncio import AsyncIOMotorClient
from typing import Any, List, Optional
from pydantic import BaseSettings, BaseModel
from models.users import User
from models.events import Event

class Database:
    def __init__(self, model):
        self.model = model

생성 처리

Database 클래스 안에 다음과 같이 save() 메서드를 추가한다.

class Database:
    def __init__(self, model):
        self.model = model

    async def save(self, document) -> None:
        await document.create()
        return

조회 처리

    async def get(self, PydanticObjectId) -> Any:
        doc = await self.model.get(id)
        if doc:
            return doc
        return False

    async def get_all(self) -> List[Any]:
        docs = await self.model.find_all().to_list()
        return docs

변경 처리

    async def update(self, id: PydanticObjectId, body:BaseModel) -> Any:
        doc_id = id
        des_body = body.dict()
        des_body = {k:v for k, v in des_body.items() if v is not None}
        update_query = {'$set': {
            field: value for field, value in des_body.items()
        }}
        doc = await self.get(doc_id)
        if not doc:
            return False
        await doc.update(update_query)
        return doc

삭제 처리

    async def delete(self, id:PydanticObjectId) -> bool:
        doc = await self.get(id)
        if not doc:
            return False
        await doc.delete()
        return True

CRUD 처리용 메서드가 추가된 데이터베이스 파일을 완성했다. 이제 라우트를 변경해보자.

routes/events.py

from beanie import PydanticObjectId
from fastapi import APIRouter, Body, HTTPException, status
from database.connection import Database
from models.events import Event, EventUpdate
from typing import List

event_router = APIRouter(
    tags=['Events'],
)
events = []
event_database = Database(Event)

@event_router.get('/', response_model=List[Event])
async def retrieve_all_events() -> List[Event]:
    events = await event_database.get_all()
    return events

@event_router.get('/{id}', response_model=Event)
async def retrieve_event(id: PydanticObjectId) -> Event:
    event = await event_database.get(id)
    if not event:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail='Event with supplied ID does not exist'
        )
@event_router.post('/new')
async def create_event(body: Event) -> dict:
    await event_database.save(body)
    return {
        'message': 'Event created successfully.'
    }
@event_router.put('/{id}', response_model=Event)
async def update_event(id: PydanticObjectId, body: EventUpdate) -> Event:
    updated_event = await event_database.update(id, body)
    if not updated_event:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail='Event with supplied ID does not exist'
        )
    return updated_event

@event_router.delete('/{id}')
async def delete_event(id: PydanticObjectId) -> dict:
    event = await event_database.delete(id)
    if not event:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail='Event with supplied ID does not exist'
        )
    return {
        'message': 'Event deleted successfully.'
    }

@event_router.delete('/')
async def delete_all_events() -> dict:
    events.clear()
    return {
        'message': 'Event deleted successfully.'
    }

이벤트 라우트용 CRUD 처리를 모두 구현했다. 이어서 사용자 등록 및 로그인 라우트를 구현해보자.

dotenv 라이브러리 설치

pip install 'pydantic[dotenv]'

[ 와 ] 를 이스케이프 하기 위해 [dotenv] 에 따옴표를 붙여야 한다.

Mac 에서 homebrew 로 mongo db 설치

brew tap mongodb/brew
brew install mongodb-community

이제 몽고DB 인스턴스와 이벤트 플래너 애플리케이션을 실행해보자.

mkdir store
mongod --dbpath store
python main.py

이제 테스트를 해보자.

필자는 pydantic 과 beanie, fastapi 등 버전충돌이 있어서 새로 venv 를 판 후, 책의 소스에서 requirements.txt 를 가져와 새로 패키지를 다운받았다.

더보기

소스코드: https://github.com/hanbit/web-with-fastapi/blob/main/ch06/planner/requirements.txt

deactivate

python3 -m venv venv2 # 가상환경 구축

source venv2/bin/activate # 가상환경 연결

pip install -r requirements.txt # 패키지 다운로드

python -m pip list # 설치확인

잘 실행되었다.

 

1. 다음 명령을 사용해 이벤트를 생성한다.

http :8000/event/new title='FastAPI Book Launch' image='fastapi-book.jpeg' description='We will be discussing the contents of the FastAPI book in this event. Ensure to come with your own copy to win gifts!' location='Google Meet' tags:='["python","fastapi","book","launch"]'

 

정상적으로 요청이 처리되면 다음과 같은 응답이 표시된다. 

 

2. 다음 명령을 사용해 모든 이벤트를 조회한다. 

 

3. 다음 명령을 사용해 단일 이벤트를 조회한다. 

 

4. 위치 (location) 를 "Hybrid" 로 변경한다.

http put :8000/event/65814cef207af6b39b0a6088 location=Hybrid

잘 수정되었다.

 

5. 이번에는 생성한 이벤트를 삭제해보자. 

 

6. 사용자를 등록해보자. 

 

7. 방금 생성한 계정으로 로그인해보자.

로그인이 성공했다.

 

CHAPTER 6 에서는 SQLModel 과 beanie 를 사용해 애플리케이션을 SQL 및 NoSQL 데이터베이스와 연동하고 각각의 이벤트가 의도한 대로 실행되는지 테스트했다. 

 

CHAPTER 7 에서는 웹 애플리케이션 보안에 관해 소개한다. 기본 인증 방법을 포함하여 FastAPI 개발자가 사용할 수 있는 다양한 인증 방법을 다룬 다음 JWT 를 기반으로 인증 시스템을 구현하고 각 이벤트 라우트에 인증을 적용한다. 마지막으로 라우트를 변경하여 이벤트와 사용자를 연결한다. 

Ch7. 보안

CHAPTER 7 에서는 JWT 를 사용해 애플리케이션의 보안을 강화한다. 오직 인증된 사용자만 특정 이벤트 처리할 수 있도록 만든다. 

 

CHAPTER 7 에서 다루는 내용은 다음과 같다. 

• FastAPI 의 인증 방식
• OAuth2 와 JWT 를 사용한 애플리케이션 보안
• 의존성 주입을 사용한 라우트 보호
• 교차 출처 리소스 공유 (CORS) 설정 

학습을 마치면 해시를 사용해 패스워드를 보호하고 FastAPI 애플리케이션에 인증 계층을 추가할 수 있다. 허가되지 않은 사용자로부터 라우트를 보호하는 방법도 알 수 있다. 

 

7.1 FastAPI 의 인증 방식

 

1. 기본 HTTP 인증
2. 쿠키
3. bearer 토큰 인증

 

7.2 OAuth2 와 JWT 를 사용한 애플리케이션 보안 

 

1. 프로젝트 폴더 (planner) 에 auth 폴더를 만든다. 

mkdir auth
cd auth && touch {__init__,jwt_handler,authenticate,hash_password}.py

 

• jwt_handler.py: JWT 문자열을 인코딩, 디코딩 하는 함수가 포함된다.
• authenticate.py: 의존 라이브러리가 포함되며 인증 및 권한을 위해 라우트에 주입된다.
• hash_password.py: 패스워드를 암호화하는 함수가 포함된다.
• __init__.py: 해당 폴더에 있는 파일들이 모듈로 사용된다는 것을 명시한다. 

 

패스워드 해싱

 

여기서는 bcrypt 를 사용해 패스워드를 암호화한다.

먼저 passlib 라이브러리를 설치한다. 

 

pip install passlib[bcrypt]

 

hash_password.py

from passlib.context import CryptContext

pwd_context = CryptContext(schemes=['bcrypt'], deprecated='auto')

class HashPassword:
    def create_hash(self, password: str):
        return pwd_context.hash(password)
    
    def verify_hash(self, plain_password: str, hashed_password: str):
        return pwd_context.verify(plain_password, hashed_password)

 

 

암호화되서 잘 저장되었다.

 

지금까지 패스워드를 안전하게 저장하는 컴포넌트를 성공적으로 구현했다. 이어서 JWT 를 생성하고 검증하는 컴포넌트를 만들어보자. 

 

액세스 토큰 생성과 검증

 

jwt_handler.py

import time
from datetime import datetime

from fastapi import HTTPException, status
from jose import jwt, JWTError
from database.connection import Settings

settings = Settings()

def create_access_token(user: str):
    payload = {
        'user': user,
        'expires': time.time() + 3600
    }
    
    token = jwt.encode(payload, settings.SECRET_KEY, algorithm='HS256')
    return token

def verify_access_token(token: str):
    try:
        data = jwt.decode(token, settings.SECRET_KEY, algorithm=['HS256'])
        
        expire = data.get('expires')
        
        if expire is None:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail='No acess token supplied'
            )
        if datetime.utcnow() > datetime.utcfromtimestamp(expire):
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail='Token expired!'
            )
        return data
    
    except JWTError:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail='Invalid token'
        )

 

지금까지 애플리케이션으로 전달된 토큰을 검증하는 함수를 만들었다. 이어서 사용자 인증을 검증하고 의존 라이브러리로 사용할 함수를 만들어보자. 

 

사용자 인증

 

authenticate.py

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from auth.jwt_handler import verify_access_token

oauth2_scheme = OAuth2PasswordBearer(tokenUrl='/user/signin')

async def authenticate(token: str = Depends(oauth2_scheme)) -> str:
    if not token:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail='Sign in for access'
        )
    
    decoded_token = verify_access_token(token)
    return decoded_token['user']

 

이 코드는 다음과 같은 의존 라이브러리를 임포트한다. 

• Depends: oauth2_scheme 을 의존 라이브러리 함수에 주입한다. 
• OAuth2PasswordBearer: 보안 로직이 존재한다는 것을 애플리케이션에 알려준다.
• verify_access_token: 앞서 정의한 토큰 생성 및 검증 함수로, 토큰의 유효성을 확인한다. 

 

라우트에 보안 적용을 위한 의존 라이브러리를 만들었다. 다음으로 라우트를 수정해서 인증 처리를 적용하고 authenticate() 함수를 이벤트 라우트에 주입해보자. 

 

7.3 애플리케이션 변경 

 

이번에는 라우트를 수정해서 새롭게 작성한 인증 모델을 적용해보자. 또한 이벤트 추가용 POST 라우트를 변경해서 사용자 레코드에 이벤트 필드를 추가해보자. 

 

jose 를 설치하지 않았다면 설치하고 가자. 

pip install 'python-jose[cryptography]'

 

 

 

 

토큰을 잘 발급한다.

 

라우트가 제대로 실행되는 것을 확인했다. 이제 이벤트 라우트를 변경해서 인증된 사용자만 이벤트를 생성, 변경, 삭제할 수 있도록 해보자. 

 

 

자물쇠가 추가되었다.

 

이벤트 문서 클래스와 라우트 변경

 

이제 이벤트의 작성자가 들어간다.

 

이제 UPDATE 라우트를 변경해보자. 

 

async def update_event(id: PydanticObjectId, body: EventUpdate, user: str = Depends(authenticate)) -> Event:
    event = await event_database.get(id)
    if event.creator != user:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail='Operation not allowed'
        )

 

이제 작성자만 수정할 수 있게 되었다. 

 

마지막으로 DELETE 라우트를 변경해보자. 

 

@event_router.delete('/{id}')
async def delete_event(id: PydanticObjectId, user: str = Depends(authenticate)) -> dict:
    event = await event_database.get(id)
    if event.creator != user:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail='Operation not allowed'
        )

 

해당 이벤트의 소유자가 아니라는 응답이 반환된다.

 

지금까지 애플리케이션과 라우트에 보안을 적용했다. 마지막으로 CORS 를 설정해보자. 

 

7.4 CORS 설정

 

교차 출처 리소스 공유는 등록되지 않은 사용자가 리소스를 사용하지 못하도록 제한하는 규칙이다. 

즉, API 와 출처 (도메인) 가 동일한 경우 또는 API 가 허가한 출처만 리소스에 접근할 수 있다. 

 

FastAPI 에서는 CORSMiddleware 라는 CORS 미들웨어를 통해 API 에 접근 가능한 출처를 관리한다. 

 

from fastapi.middleware.cors import CORSMiddleware

# 출처 등록

origins = ['*']

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=['*'],
    allow_headers=['*'],
)

 

이 코드는 CORSMiddleware 클래스를 FastAPI 에서 임포트한 다음 origins 배열에 등록하고 add_middleware() 메서드를 사용해 미들웨어를 등록한다. 

 

문서:https://fastapi.tiangolo.com/tutorial/cors/

 

CHAPTER 7 에서는 OAuth2 와 JWT 를 사용해 FastAPI 애플리케이션의 보안을 강화하는 방법을 배웠다. 

 

CHAPTER 8 에서는 FastAPI 애플리케이션을 테스트하는 방법을 소개한다. 애플리케이션 테스트가 무엇인지, 왜 필요한지, 어떻게 테스트하는지 학습한다. 

 

Ch8. 테스트

CHAPTER 8 에서 다루는 내용은 다음과 같다. 

• pytest 를 사용한 단위 테스트
• 테스트 환경 구축
• REST API 라우트 테스트 작성
• 테스트 커버리지 

 

8.1 pytest 를 사용한 단위 테스트

 

파이썬은 unnittest 라는 내장 라이브러리를 제공하지만 pytest 가 더 간단한 구문을 사용할 수 있어서 인기가 많다. pytest 를 사용해 테스트 코드를 작성해보자. 

 

pip install pytest==7.1.2

 

 

테스트 파일을 만들 때는 파일명 앞에 test_ 를 붙여야 한다. 그러면 해당 파일이 테스트 파일이라는 것을 pytest 라이브러리가 인식해서 실행한다. 

 

def add(a: int, b: int) -> int:
    return a+b

def subtract(a: int, b: int) -> int:
    return b-a

def multiply(a: int, b: int) -> int:
    return a*b

def divide(a: int, b: int) -> int:
    return b//a
    
def test_add() -> None:
    assert add(1,1) == 2
def test_subtract() -> None:
    assert subtract(2,5) == 3
def test_multiply() -> None:
    assert multiply(10,10) == 100
def test_divide() -> None:
    assert divide(25,100) == 4

 

테스트는 pytest 명령을 사용해 실행한다. 

단, 이 명령은 명령을 실행하는 위치에 있는 모든 테스트 파일을 실행한다. 테스트를 하나만 실행하려면 파일명을 인수로 지정해야 한다. 다음 명령을 사용해 우리가 작성한 테스트 파일만 실행해보자. 

 

pytest test_arithmetic_operations.py

 

테스트가 잘 성공했다.

 

 

pytest 가 어떻게 실행되는지 간단히 살펴봤다. 다음으로 pytest 의 픽스처 (fixture) 를 알아보자.

 

픽스처를 사용한 반복 제거 

 

픽스처는 재사용할 수 있는 함수로, 테스트 함수에 필요한 데이터를 반환하기 위해 정의된다. 

pytest.fixture 데코레이터를 사용해 픽스처를 정의할 수 있으며 API 라우트 테스트 시 애플리케이션 인스턴스를 반환하는 경우 등에 사용된다. 

 

import pytest
from models.events import EventUpdate

@pytest.fixture
def event() -> EventUpdate:
	return EventUpdate(
    	title='FastAPI Book Launch',
        image='https://packt.com/fastapi.png',
        description='We will be discussing the contents of the FastAPI book in this event.Ensure to come with your own copy to win gifts!',
        tags=['python', 'fastapi', 'book', 'launch'],
        location='Google Meet'
    )

def test_event_name(event: EventUpdate) -> None:
	assert event.title == 'FastAPI Book Launch'

 

픽스처 데코레이터는 인수를 선택적으로 받을 수 있다. 예를 들어 scope 인수는 픽스처 함수의 유효 범위를 지정할 때 사용된다. 여기서는 두 가지 scope 를 사용한다.

• session: 테스트 전체 세션 동안 해당 함수가 유효하다.
• module: 테스트 파일이 실행된 후 특정 함수에서만 유효하다.

테스트 환경을 구축해보자.

 

8.2 테스트 환경 구축

 

우리가 만든 비동기 API 를 테스트하려면 httpx 와 pytest-asyncio 라이브러리를 설치해야 한다. 

 

pip install httpx==0.22.0 pytest-asyncio==0.18.3

 

pytest.ini

[pytest]
asyncio_mode = auto

 

pytest 가 실행될 때 이 파일의 내용을 불러온다. 이 설정은 pytest 가 모든 테스트를 비동기식으로 실행한다는 의미다. 

 

import asyncio
import httpx
import pytest

from main import app
from database.connection import Settings
from models.events import Event
from models.users import User

@pytest.fixture(scope='session')
def event_loop():
    loop = asyncio.get_event_loop()
    yield loop
    loop.close()

async def init_db():
    test_settings = Settings()
    test_settings.DATABASE_URL = 'mongodb://localhost:27017/testdb'
    
    await test_settings.initialize_database()
    
@pytest.fixture(scope='session')
async def default_client():
    await init_db()
    async with httpx.AsyncClient(app=app, base_url='http://app') as client:
        yield client
        await Event.find_all().delete()
        await User.find_all().delete()

 

지금까지 테스트 환경 구축 방법을 살펴봤다. 이어서 애플리케이션의 각 라우트를 테스트하는 코드를 작성해보자. 

 

8.3 REST API 라우트 테스트 작성 

 

사용자 등록 테스트

import httpx
import pytest

@pytest.mark.asyncio
async def test_sign_new_user(default_client: httpx.AsyncClient) -> None:
    payload = {
        'email': 'testuser@packt.com',
        'password': 'testpassword',
    }
    headers = {
        'accept': 'application/json',
        'Content-Type': 'application/json',
    }
    test_response = {
        'message': 'User created successfully.'
    }

    response = await default_client.post('/user/signup', json=payload, headers=headers)
    assert response.status_code == 200
    assert response.json() == test_response

 

 

로그인 라우트 테스트

 

두 번째로 로그인 라우트 테스트를 작성해보자.

테스트가 성공했다.

 

CRUD 라우트 테스트

 

먼저 tests 폴더에 test_routes.py 파일을 만든다.

 

조회 라우트 테스트

 

import httpx
import pytest

from auth.jwt_handler import create_access_token
from models.events import Event

@pytest.fixture(scope='module')
async def access_token() -> str:
    return create_access_token('testuser@packt.com')

@pytest.fixture(scope='module')
async def mock_event() -> Event:
    new_event = Event(
        creator='testuser@packt.com',
        title='FastAPI Book Launch',
        image='https://linktomyimage.com/image.png',
        description='We will be discussing the contents of the FastAPI...',
        tags=['python','fastapi','book','launch'],
        location='Google Meet'
    )
    await Event.insert_one(new_event)

    yield new_event

@pytest.mark.asyncio
async def test_get_events(default_client: httpx.AsyncClient, mock_event: Event) -> None:
    response = await default_client.get('/event')

    assert response.status_code == 200
    assert response.json()[0]['_id'] == str(mock_event.id)

 

이번에는 /event/{id} 라우트 테스트 함수를 작성해보자.

 

@pytest.mark.asyncio
async def test_get_event(default_client: httpx.AsyncClient, mock_event: Event) -> None:
    url = f'/event/{str(mock_event.id)}'
    response = await default_client.get(url)

    assert response.status_code == 200
    assert response.json()['creator'] == mock_event.creator
    assert response.json()['_id'] == str(mock_event.id)

 

생성 라우트 테스트

@pytest.mark.asyncio
async def test_post_event(default_client: httpx.AsyncClient, access_token: str) -> None:
    payload = {
        'title': 'FastAPI Book Launch',
        'image': 'https://linktomyimage.com/image.png',
        'description': 'We will be discussing...',
        'tags': ['python','fastapi','book','launch'],
        'location': 'Google Meet',
    }
    headers = {
        'Content-Type': 'application/json',
        'Authorization': f'Bearer {access_token}'
    }
    test_response = {
        'message': 'Event created successfully.'
    }
    response = await default_client.post('/event/new', json=payload, headers=headers)

    assert response.status_code == 200
    assert response.json() == test_response

 

4개의&nbsp; 테스트 모두 성공했다.

 

변경 라우트 테스트

@pytest.mark.asyncio
async def test_update_count(default_client: httpx.AsyncClient, mock_event: Event, access_token: str) -> None:
    test_payload = {
        'title': 'Updated FastAPI event'
    }
    headers = {
        'Content-Type': 'application/json',
        'Authorization': f'Bearer {access_token}'
    }

    url = f'/event/{str(mock_event.id)}'
    response = await default_client.put(url, json=test_payload, headers=headers)

    assert response.status_code == 200
    assert response.json()['title'] == test_payload['title']

 

삭제 라우트 테스트

@pytest.mark.asyncio
async def test_delete_count(default_client: httpx.AsyncClient, mock_event: Event, access_token: str) -> None:
    test_response = {
        'message': 'Event deleted successfully.'
    }
    headers = {
        'Content-Type': 'application/json',
        'Authorization': f'Bearer {access_token}'
    }

    url = f'/event/{mock_event.id}'
    response = await default_client.delete(url, headers=headers)

    assert response.status_code == 200
    assert response.json() == test_response

 

 

@pytest.mark.asyncio
async def test_get_event_again(default_client: httpx.AsyncClient, mock_event: Event) -> None:
    url = f'/event/{str(mock_event.id)}'
    response = await default_client.get(url)

    assert response.status_code == 404

 

 

이벤트 플래너 API 의 모든 라우트를 테스트했다. 이제 커버리지 테스트를 실행하여 테스트 대상이 되는 코드의 비율을 파악해보자. 

 

8.4 테스트 커버리지

 

테스트 커버리지 보고서는 테스트가 전체 애플리케이션 코드 중 어느 정도 비율의 코드를 테스트하는지 정량화해서 보여준다. 

coverage 모듈을 설치해서 우리가 만든 API 가 적절하게 테스트되고 있는지 확인해보자. 

 

설치

pip install coverage==6.3.3

 

coverage run -m pytest

 

이 보고서는 테스트를 통해 실행된 (테스트에 사용된) 코드의 비율을 보여준다. 

 

 

htmlcov 폴더에 생성된 index.html 파일을 브라우저로 열어보자.

실행된 코드는 초록색, 그렇지 않은 코드는 빨간색으로 표시된다.

 

CHAPTER 8 에서는 인증 및 CRUD 라우트 테스트를 작성해서 API 의 전체 기능을 테스트했다. 구체적으로는 테스트가 무엇인지, pytest 를 사용해서 테스트를 어떻게 작성하는지 살펴봤다. 또한 pytest 의 픽스처에 관해 배우고 이를 사용해서 재사용할 수 있는 접속 토큰과 데이터베이스 객체를 생성했으며 테스트 세션 동안 애플리케이션 인스턴스를 유지했다. 그리고 테스트를 사용해서 API HTTP 요청의 응답을 확인하고 API 처리를 검증했다. 마지막으로 테스트 커버리지 보고서를 생성하여 테스트가 적용된 코드와 그렇지 않은 코드를 구분했다. 

 

CHAPTER 9 에서는 애플리케이션을 컨테이너화하는 방법과 도커 및 도커 구성 도구를 사용해 로컬에 배포하는 방법을 배운다. 

 

Ch9. 배포

CHAPTER 9 에서는 다음과 같은 내용을 다룬다.

• 배포 준비
• 도커를 사용한 배포
• 도커 이미지 배포

9.1 배포 준비

 

의존 라이브러리 관리

 

앞서 beanie 와 pytest 같은 라이브러리를 설치했다. 하지만 의존성 관리자 역할을 하는 requirements.txt 파일에는 포함되어 있지 않으니 추가하도록 하자. requirements.txt 파일은 항상 최신 상태로 유지해야 한다. 

 

파이썬에서는 pip freeze 명령을 사용해 개발 환경에 사용된 패키지들을 추출할 수 있다. 

 

9.2 도커를 사용한 배포

 

도커파일 작성

 

FROM python:3.10
WORKDIR /app
COPY requirements.txt /app/requirements.txt
RUN pip install --upgrade pip && pip install -r /app/requirements.txt
EXPOSE 8000
COPY ./ /app
CMD ["python", "main.py"]

 

 

docker build -t event-planner-api .

 

로컬에 애플리케이션 배포 

 

 

docker-compose.yml

version: "3"

services:
  api:
    build: .
    image: event-planner-api:latest
    ports:
      - "8000:8000"
    env_file:
      - .env.prod

  database:
    image: mongo:5.0.15
    ports:
      - "27017"
    volumes:
      - data:/data/db

volumes:
  data:

 

애플리케이션 실행 

 

도커 구성 파일을 기반으로 애플리케이션을 배포하고 실항할 준비를 마쳤다. 

docker-compose up -d

 

명령을 입력하면 서비스가 분리 모드 (detached mode) 로 실행된다. 

 

서버와 DB 컨테이너가 잘 떠있다.

 

http :8000/event
http :8000/user/signup email=fastapi@packt.com password=strong!!!

 

 

9.3 도커 이미지 배포

 

• 구글 클라우드 런
• 아마존 EC2
• 마이크로소프트 애저 

 

저자가 제공하는 추가 실습 자료: https://www.youngest.dev/

• Okteto 를 사용한 방명록 구축 동영상 강의
• 몽고DB, JWT 인증, 리액트를 활용한 추가 예제 

 

이상 이 책을 12월 21일에 마무리 짓게 되었다. 

예상했던 기간 한달보다 10일정도 빨리 끝내게 되었다.