FastAPI에서의 라우팅

FastAPI의 라우팅은 매우 유연하고, 번거로운 일을 처리해준다. Routing이란 클라이언트가 서버로 보내는 HTTP request들을 처리하는 과정을 말한다. HTTP request들은 정해진 routes로 전달되며, 이는 정해진 핸들러를 통해서 요청과 응답을 처리한다. 이러한 핸들러를 route handler라고 한다.

우리는 APIRouter 인스턴스를 사용하여 route를 만드는 방법에 대해서 알아보고, 메인 FastAPI 어플리케이션과 연결하도록 할 것이다. 또한, model이 어떤 것들이고, request body를 검증하는 방법에 대해서 알아볼 것이다. 또한, path를 어떻게 사용하고, query parameter를 FastAPI에서 어떻게 사용할 것인지 알아볼 것이다.

먼저 환경 설정을 하도록 하자.

python3 -m venv venv
source ./venv/bin/activate
pip3 install fastapi uvicorn
pip freeze > requirementes.txt

환경 설정이 끝났다면, FastAPI를 사용하여 라우팅을 해보도록 하자.

FastAPI에서 라우팅에 대한 이해

HTTP request method으로부터 request를 받아들이는 라우팅을 정의하고, 추가적으로 파라미터를 받아들일 수 있다. request가 route로 전달될 때 application은 route handler에서 request를 처리하기 이전에 route가 정의되어있는 지 아닌 지 확인한다.

route handler는 server에 전달된 request를 처리하는 함수이다. 가령, server로 요청이 오면 database로부터 data를 가져오는 handler도 함수인 것이다.

FastAPI에서 라우팅을 하는 예제는 다음과 같다.

api.py라는 파일을 만들어서 다음의 코드를 넣도록 하자.

• api.py

from fastapi import FastAPI

app = FastAPI()

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

라우팅은 FastAPI 인스턴스인 app 변수로 부터 다뤄진다.

uvicron을 사용하여 FastAPI 인스턴스를 지정해주고, application을 실행시켜주도록 하자.

uvicorn api:app --port 8080 --reload

다음의 url로 들어가면 우리가 전달한 응답이 보일 것이다.

curl http://127.0.0.1:8080/

{"message":"Hello World"}

예전에는 FastAPI인스턴스만이 routing operation에 사용되었다. 그러나, 이는 단일 path에 대해서만 사용되었는데, 분리된 path에 대하여 핸들러를 지정하려고 해도 uvicorn이 단일 FastAPI만을 사용하기 때문에 entry point는 하나 밖에 되지않는다.

가령, user에 대한 route처리를 하고 싶지만 이를 그룹핑할 방법이 없다는 것이다. FastAPI 인스턴스는 하나이고, 이를 통해서 그룹핑한다면 다른 route는 설정이 안되기 때문이다.

그렇다면 일련의 route들을 다른 함수들을 통해서 어떻게 사용할 수 있을까?? 이를 위해 APIRouter class가 multiple routing을 지원하게 되었다.

APIRouter class를 사용한 라우팅

APIRouter클래스는 FastAPI 패키지에 속하며, 다양한 route에 대한 path 연산들을 만든다. APIRouter 클래스는 모듈화와 application routing과 logic의 조직을 장려한다.

APIRouter 클래스는 fastapi 패키지를 통해 import되고, 인스턴스를 생성하면 된다. 그리고 해당 인스턴스를 통해서 route method이 만들어지고, 배포된다.

• api.py

from fastapi import FastAPI
from fastapi import APIRouter

app = FastAPI()
router = APIRouter()

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

@router.get('/hello')
async def say_hello() -> dict:
    return {"message": "Hello!"}

app.include_router(router=router)

다음의 코드를 실행하고 /hello로 curl을 전달하면 응답이 전달될 것이다.

uvicorn api:app --port 8080 --reload

curl localhost:8080/hello

{"message":"Hello!"}

APIRouter를 통해서 라우팅을 설정할 수 있으며, 이를 FastAPI 인스턴스에 등록해주면 끝이다. 등록은 include_router이 끝이며, 이를 사용할 때는 꼭 @router.get과 같은 등록이 먼저 실행된 뒤에 해야한다.

이제 실질적으로 APIRouter를 사용하는 방법에 대해서 알아보자. todo라는 기능에 관련된 route들을 하나의 todo.py에서 모두 처리한다고 하자. 다음과 같이 만들 수 있다.

todo.py파일을 만든다음

touch todo.py

다음의 코드를 넣어주도록 하자.

• 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}

위의 코드는 두가지 todo연산을 만들었다. 첫번째는 todo list에 todo 객체를 추가하는 POST메서드이고, 두 번째는 todo list를 가져오는 GET연산이다.

이제 이를 FastAPI에서 찾을 수 있도록, include_router()에 등록하도록 하자.

include_router() 메서드는 APIRouter클래스로 정의된 route들을 등록하는 일을 한다.

• 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(router=todo_router)

다음과 같이 설정해놓으면, todo가 import되면서 APIRouter 설정이 완료되고, app.include_router를 통해서 todo_router가 설정된다.

이제 server를 실행시켜보도록 하자.

uvicorn api:app --port 8080 --reload

서버가 실행되고, curl을 통해서 todo get요청을 전달해보도록하자.

curl localhost:8080/todo

{"todos":[]}

잘 응답이 왔을 것이다. 다음으로, post에 요청을 전달하여, todo객체를 생성해보도록 하자.

다음의 응답이 전달될 것이다.

{"message":"Todo added successfully"}

다시 get으로 todo를 요청하면 다음의 응답이 오게 된다.

curl localhost:8080/todo
{"todos":[{"id":1,"item":"First Todo is to finish this book!"}]}

APIRouter가 어떻게 동작하는 지 알아보았고, 이를 어떻게 main API instance에 포함시킬 수 있는 지 보았다. 그러나, todo model은 완벽한 모델이 아니다. 다음 챕터로 pydantic model에 대해서 알아보고 usecase를 만들어보도록 하자.

Pyadntic model을 사용하여 request body를 검증하는 방법

FastAPI에서는 오직 정의된 데이터만 전달되도록 보장할 수 있게 request body를 검증할 수 있다. 이는 매우 중요한데, 이는 request data를 깨끗히하고, 의도적인 공격을 줄이도록 한다. 이러한 과정을 vaidation이라고 한다.

FastAPI에서의 model은 구조화된 class로 어떤 데이터가 오고, parse될 것인지 표기되어 있다. model은 Pydantic의 BaseModel class의 서브클래스로 만들어진다.

pydantic이란 python library로 python-type annotation을 통해서 data validation을 수행하도록 한다.

Model을 정의할 때, request body에 대한 type hint가 사용되어, request-response object에 사용된다. 이번 챕터에서는 오직 pydantic model을 request body에만 사용하는 방법을 알아보도록 하자.

사용방법은 다음과 같다.

from pydantic import BaseModel
class Book(BaseModel):
    id: int
    Name: str
    Publishers: str
    Isbn: str

위의 코드는 pydantic의 BaseModel 서브클래스로 Book이라는 클래스를 만들었다. 여기에는 변수들이 있는데, 이들은 타입 힌트인 :로 표시되어있다. 이 4개의 field들은 오직 타입 힌트로 적힌 타입으로만 정의될 수 있다.

이제 todo model을 만들어보도록 하자.

async def add_todo(todo: dict) -> dict:
    ...

post method의 예제로 다음의 데이터가 전달된다.

{
    "id": id,
    "item": item
}

이전에 사용한 todo 모델 방법으로는 user가 빈 field를 가진 todo 객체를 전달해도 문제가 없다. user는 자신이 어떤 잘못을 했는 지 모르며, 어떠한 error response도 받지 못한다.

이를 해결해주는 것이 바로 pydantic의 BaseModel이다.

가령, 우리는 request body가 위의 id, item을 포함하게 하기 위해서는 다음과 같이 정의할 수 있다. model.py파일을 만들어서 다음의 코드를 넣어보도록 하자.

• model.py

from pydantic import BaseModel

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

이제 pydantic의 서브 클래스 모델을 만들었고, id는 int타입이어야 하고, item은 str 타입이어야 한다.

다음의 모델을 todo.py에서 가져와 request body에 쓰도록 하자.

• todo.py

from fastapi import APIRouter
from model import Todo

todo_router = APIRouter()

todo_list = []

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

@todo_router.get("/todo")
async def retrieve_todos() -> dict:
    return {"todos": todo_list}

이제 새로운 request body가 잘 검증되는 지 확인해보도록 하자. 비어있는 request body를 전달하여 잘 검증되는 지를 확인해보도록 하는 것이다.

curl -X 'POST' \
  'http://127.0.0.1:8080/todo' \
  -H 'accept: application/json' \
  -H 'Content-Type: application/json' \
  -d '{
}' -v

다음의 응답이 전달될 것이다.

< HTTP/1.1 422 Unprocessable Entity
< date: Sun, 28 May 2023 07:23:53 GMT
< server: uvicorn
< content-length: 162
< content-type: application/json
< 
* Connection #0 to host 127.0.0.1 left intact
{"detail":[{"loc":["body","id"],"msg":"field required","type":"value_error.missing"},{"loc":["body","item"],"msg":"field required","type":"value_error.missing"}]}%                    

validation 실패로 전달된 응답에는 detail로 하여 생략된 field들이 적혀있다.

다음으로 올바른 요청을 전달해보도록 하자.

curl -X 'POST' \
  'http://127.0.0.1:8080/todo' \
  -H 'accept: application/json' \
  -H 'Content-Type: application/json' \
  -d '{
  "id": 2,
  "item": "Validation models help with input types"
}' -v

다음의 응답이 전달된다.

< HTTP/1.1 200 OK
< date: Sun, 28 May 2023 07:26:07 GMT
< server: uvicorn
< content-length: 37
< content-type: application/json
< 
* Connection #0 to host 127.0.0.1 left intact
{"message":"Todo added successfully"}

Nested models(중첩된 모델)

pydantic model은 중첩이 될 수도 있다. 다음을 확인해보도록 하자.

• model.py

from pydantic import BaseModel

class Item(BaseModel):
    item: str
    status: str

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

다음과 같이 요청해보도록 하자.

curl -X 'POST' \
  'http://127.0.0.1:8080/todo' \
  -H 'accept: application/json' \
  -H 'Content-Type: application/json' \
  -d '{
  "id": 2,
  "item": {
    "item": "hello",
    "status": "world"
  }
}' -v

todo 아이템이 만들어지는 것에 성공하게 될 것이다. 이제 아이템을 불러와보도록 하자.

curl localhost:8080/todo

{"todos":[{"id":2,"item":{"item":"hello","status":"world"}}]}

중첩된 모델이 성공한 것을 볼 수 있다.

path parameter와 query parameter

경로 매개변수

경로 매개변수는 resource를 식별하기 위해서 API routing을 할 때 사용된다. 이 매개변수는 식별자 역할을 하며, 웹 어플리케이션이 추가 처리할 수 있도록 연결 고리가 되고도 한다.

이제 todo의 특정 id를 지정하여 특정 todo를 반환하는 코드를 만들도록 하자. 먼저 하나의 todo만 추출하는 새로운 라우트를 만들도록 하자. 이때 사용되는 것이 바로 경로 매개변수이며, 경로 매개변수로 id를 지정하도록 하는 것이다.

todo.py에 다음과 같이 새로운 route를 추가한다.

• todo.py

from fastapi import APIRouter
from model import Todo

todo_router = APIRouter()

todo_list = []

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

@todo_router.get("/todo")
async def retrieve_todos() -> dict:
    return {"todos": todo_list}

@todo_router.get("/todo/{todo_id}")
async def get_single_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."
    }

여기서 {todo_id}가 바로 경로 매개변수이다. 이 매개변수를 통해 어플리케이션이 지정한 ID와 일치하는 todo 작업을 반환할 수 있다.

추가한 라우트를 테스트해보자.

curl -X 'POST' \
  'http://127.0.0.1:8080/todo' \
  -H 'accept: application/json' \
  -H 'Content-Type: application/json' \
  -d '{
  "id": 2,
  "item": {
    "item": "hello",
    "status": "world"
  }
}' -v

curl localhost:8080/todo/2 

다음의 결과를 얻을 수 있다.

{"todo":{"id":2,"item":{"item":"hello","status":"world"}}}

curl localhost:8080/todo/2을 통해서 id가 2인 todo가 반환되는 것이다.

추가적으로, FastAPI에는 Path라는 클래스를 지원하는데, Path는 FastAPI가 제공하는 클래스로, route 함수에 있는 다른 인수와 경로 매개변수를 구분하는 역할을 한다. Path는 또한 swagger와 ReDoc등으로 OpenAPI기반 문서를 자동 생성할 때 route관련 정보를 함께 문서화하도록 돕는다.

route 정의를 다음과 같이 수정해보도록 하자.

• todo.py

from fastapi import APIRouter, Path
from model import Todo

...

@todo_router.get("/todo/{todo_id}")
async def get_single_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
            }
    return {
        "message": "Todo with supplied ID dosen't exist."
    }

Path 클래스는 첫 인수로 None또는 ...을 받을 수 있다. 첫번째 인수가 ...이면 경로 매개변수를 반드시 지정해야한다. 즉, ...은 required이고, None이면 optional하다는 것이다. 또한, 경로 매개변수가 숫자이면 수치 검증을 위한 인수를 지정할 수 있다. 가령 gt, le와 같은 검증 기호를 사용할 수 있다. 이를 통해 경로 매개변수에 사용된 값이 특정 범위에 있는 숫자인지 검증 가능하다. Path(..., title="The ID of the item to get", ge=1)다음과 같이 쓸 수 있다.

쿼리 매개변수

쿼리 매개변수는 URL에서 ? 뒤에 나온다. 제공된 쿼리 기반으로 특정한 값을 반환하거나 요청을 필터링할 때 사용된다.

쿼리는 route handler(함수)의 인수로 사용되지만, 경로 매개변수와 다른 형태로 정의된다. 가령, 다음과 같이 FastAPI Query 클래스의 인스턴스를 만들어서 route handler의 인수로 쿼리를 정의할 수 있다.

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

Query클래스는 다양한 매개변수를 받을 수 있는데, 첫번째 값은 default값으로, None이면 None을 받을 수 있다는 것이다. 만약, rick으로 쓰면 query가 없을 경우 rick으로 온다.

쿼리 매개변수는 추후에 더 복잡한 어플리케이션을 만들 때 깊게 살펴보도록 하자.

Request body(요청 바디)

request body는 POST나 UPDATE 등 routing method를 사용해 API로 전달되는 데이터이다.

@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 doesn't exist."
    }

여기서 request body는

{
  "id": 2,
  "item": {
    "item": "hello",
    "status": "world"
  }
}

이 부분이다.

FastAPI는 추가 검증을 할 수 있는 Body클래스를 제공한다.

from fastapi import Body를 통해 Body클래스를 얻을 수 있고 함수 매개변수 부분에 Query처럼 todo: Todo = Body(...)로 쓰면 된다. ...을 쓰면 required가 되는 것이다.

FastAPI 자동 문서화

FastAPI는 모델의 JSON 스키마 정의를 생성하고 routes, request body type, path, query parameter, response model 등을 자동으로 문서화한다. swagger와 ReDoc 두개의 문서 타입으로 제공되는데, swagger를 알아보자.

서버를 열고, /docs url에 접근하면 브라우저에서 swagger화면을 볼 수 있다.

http://localhost:8080/docs

Redoc의 경우는 /redoc url에 접근하면 브라우저에서 볼 수 있다.

http://localhost:8080/redoc

documentation에 JSON스키마를 추가하고 싶다면, BaseModel클래스 안에 Config클래스를 정의하면 된다. 다음과 같이 Todo 모델 클래스에 샘플 데이터를 추가해보도록 하자.

from pydantic import BaseModel

class Item(BaseModel):
    item: str
    status: str

class Todo(BaseModel):
    id: int
    item: Item
    
    class Config:
        schema_extra = {
            "example": {
                "id": 1,
                "item": "Example Schema!"
            }
        }

다시 swagger에 들어가서 example부분을 확인해보도록 하자.

추가된 것을 볼 수 있다.

간단한 CRUD 어플리케이션 개발

이제 todo아이템을 추가하고 불러오는 것을 했으니, 수정하고, 삭제하는 코드를 추가해보도록 하자.

먼저 model.py에 route request body용 모델을 model.py에 추가해보도록 하자.

• model.py

...
class TodoItem(BaseModel):
    item: Item
    
    class Config:
        schema_extra = {
            "example": {
                "item": "Read the next chapter of the book"
            }
        }
...

다음으로 todo를 변경하기 위한 route를 todo.py에 추가하도록 하자.

• todo.py

...
@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 doesn't exist."
    }
...

새로 추가한 라우트를 테스트해보도록 하자. 먼저 신규 todo 아이템을 추가한다.

curl -X 'POST' \
  'http://127.0.0.1:8080/todo' \
  -H 'accept: application/json' \
  -H 'Content-Type: application/json' \
  -d '{
  "id": 1,
  "item": {
    "item": "hello",
    "status": "world"
  }
}'

...

{"message":"Todo added successfully"}

다음으로 todo item을 변경해보도록 하자.

curl -X 'PUT' \ 
  'http://127.0.0.1:8080/todo/1' \
  -H 'accept: application/json' \
  -H 'Content-Type: application/json' \
  -d '{
  "item": {
    "item": "bye",  
    "status": "world"
  }
}'

...

{"message":"Todo updated successfully"}

성공하였다면, 잘 변경되었는 지 확인해보도록 하자.

curl localhost:8080/todo/1

다음의 결과가 나올 것이다.

"todo":{"id":1,"item":{"item":"bye","status":"world"}}}

잘 바뀐 것이 확인되었다.

다음으로 todo.py에 삭제를 위한 DELETE 라우트를 추가해보자.

• todo.py

...
@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"
    }
...

추가한 DELETE 라우트 테스트를 해보도록 하자.

curl -X 'POST' \
  'http://127.0.0.1:8080/todo' \
  -H 'accept: application/json' \
  -H 'Content-Type: application/json' \
  -d '{
  "id": 1,
  "item": {
    "item": "hello",
    "status": "world"
  }
}'

curl -X 'POST' \
  'http://127.0.0.1:8080/todo' \
  -H 'accept: application/json' \
  -H 'Content-Type: application/json' \
  -d '{
  "id": 2,
  "item": {
    "item": "bye",
    "status": "world"
  }
}'

todo 아이템을 두 개 추가하였으니, id 1번을 삭제해보도록 하자.

curl -X DELETE localhost:8080/todo/1

다음의 결과가 나온다.

{"message":"Todo deleted successfully"}

todo_list를 확인해보면 없는 것을 볼 수 있다.

curl localhost:8080/todo

다음의 결과가 나온다.

{"todos":[{"id":2,"item":{"item":"bye","status":"world"}}]}

id 1번이 삭제된 것을 확인할 수 있다.

다음으로, 모두 삭제해보도록 하자.

curl -X DELETE localhost:8080/todo

...

{"message":"Todos deleted successfully"}

다음의 결과가 나온다.

curl localhost:8080/todo

...

"todos":[]}

성공한 것을 볼 수 있다.

이제 다음 장에서 FastAPI의 response, response model, pydantic model response model, status code와 error 처리에 대해서도 알아보도록 하자.