[FastAPI] FastAPI 입문 part2 - CRUD 어플리케이션 만들기

지난 part1에서 FastAPI를 활용해 간단한 “hello world”와 “routing”을 살펴보았다. 이번 글에서는 CRUD 기능을 수행하는 간단한 어플리케이션을 만들어보면서 FastAPI 프레임워크에 친숙해져보자.

아래의 실습 과정은 지난 part1 글에서 이어서 해보는 것이므로, 앞의 글을 먼저 읽고 따라해보면 좋겠다.

🖐🏻!여기서 잠깐!🖐🏻

CRUD 를 간단하게 설명하자면, C: Create(생성), R: Read(읽기), U: Update(갱신), Delete(삭제)를 일컫는 말이다. API 를 구축할 때 사용자 혹은 클라이언트가 요구하는 기본적인 기능이라고 이해하면 좋겠다.

1. Alarm 어플리케이션 만들기

지금부터 우리가 까먹기 쉽고 상기시켜야 하는 알람들의 리스트를 보여주고, 추가하는 기능을 갖고 있는 어플리케이션을 만들어보려고 한다. 먼저 지난 part1에서 만들었던 main.py와 routers/new_router.py에 코드를 추가하는 형태로 이어가보려고 한다. 먼저 new_router.py에 다음과 같은 코드를 추가하자.

1-1. 새로운 router 추가

# routers/new_router.py

alarm_list = []

@router.get("/alarm")
async def show_alarm() -> dict:
    return {
        "alarms": alarm_list
    }

@router.post("/alarm")
async def add_alarm(alarm: dict) -> dict:
    alarm_list.append(alarm)
    return {
        "message": "Your ALARM added successfully."
    }

위 코드를 살펴보면, 먼저 전역 alarm_list라는 빈 리스트를 실행해주고, 한 개의 GET 메서드와 POST 메서드를 만들어 준다.

GET메서드의 코드부터 살펴보면, uri(/alarm)가 GET 메서드로 들어오면, (우리가 미리 작성했던 코드에서 주의할 점은 우리는 new_router.py를 만들 때, prefix=”/new” 라고 주었다. 따라서 전체 uri는 /new/alarm이 되겠다.) 아래의 show_alarm()이 실행된다. 이 함수는 우리가 미리 선언해 두었던 alarm_list를 반환하는 함수이다.

POST메서드의 코드를 살펴보면, uri(/alarm)가 POST 메서드로 들어오면, 아래의 add_alarm() 함수가 실행된다. 이 함수는 인자로 받은 alarm을 alram_list에 추가하고, 성공적으로 추가 되었다는 메시지까지 반환하는 함수이다.

1-2. 테스트 해보기

여기까지 작성한 코드를 테스트 해 보자.

다음의 명령어로 우리의 어플리케이션을 실행해보자.

uvicorn main:app --port 8080 --reload

새롭게 만든 uri 로 테스트하기 위해 다음의 명령어를 사용한다. 먼저 GET 메소드에 대해서는,

(1) GET 메소드 테스트 명령어

curl -X 'GET' 'http://localhost:8080/new/alarm' -H 'accept: application/json'

아직 우리의 alarm_list에 추가된 것이 없으므로 빈 리스트로 나오는 것이 당연하다.

(2) POST 메소드 테스트 명령어

이제 다음의 명령어를 통해 우리의 alarm_list에 새로운 데이터를 추가해보자.

curl -X 'POST' 'http://localhost:8080/new/alarm' -H 'accept: application/json' -H 'Content-Type: application/json' -d '{"id": 1, "alarm": "You pushed the first alarm in your list!"}'

성공적으로 추가된 뒤에 다시 (1)의 GET메소드를 실행해보면, 처음과는 다르게 방금 추가한 메시지가 출력되는 것을 확인 할 수 있다.

이렇게 우리는 CRUD 중 C, R을 수행하는 기능을 만들어 보았다.

2. 데이터 검증 : pydantic

데이터를 수정하고 지우는 것은 API, 어플리케이션 입장에서는 매우 critical 한 기능이다. 서버를 운영하고 개발하는 입장에서는 수집된 데이터를 변경하거나 삭제하는 액션이 실행되기 때문이다. 따라서 우리는 수정하려고 하는 새로운 데이터가 기존 형식에 알맞게 수정되거나 삭제되는지 검증해야한다. 이 작업은 python 에서는 “Pydantic”이라고 하는 편한 라이브러리를 이용해 해당 데이터를 검증한다.

2-1. 모델 세우기.

Web Application 에서 데이터가 어떻게 구성되고 처리되어야 하는지 미리 선언해 놓는 것을 모델이라고 한다. 우리 어플리케이션에도 model.py를 만들고 해당 파일에 다음과 같이 작성하자.

# model.py

from pydantic import BaseModel

class Alarm(BaseModel):
    id: int
    alarm: str

class AlarmData(BaseModel):
    alarm: str

우리가 다룰 Alarm이라는 클래스는 id 와 alarm이라는 내용으로 구성되고 이 형식에 맞게 데이터가 들어와야 한다는 것이다. 그 아래 AlarmData는 뒤이어 데이터 업데이트 부분에서 활용될 내용이다. 수정할 데이터가 str 형식으로 들어와야 됨을 의미한다.

2-2. model.py를 활용한 코드 수정

위에 model.py 를 만들었으면 우리의 라우터에도 해당 데이터 검증 작업이 추가 될 수 있도록 코드를 수정하자. 먼저 데이터가 가장 먼저 추가 되는 @router.post(”/alarm”)부분의 함수 add_alarm(alarm: Alarm) 이다.

@router.post("/alarm")
async def add_alarm(alarm: Alarm) -> dict:
    alarm_list.append(alarm)
    return {
        "message": "Your ALARM added successfully."
    }

이렇게 되면 POST 되어 전달되는 데이터가 우리가 미리 작성해둔 모델(Alarm)의 id, alarm 이 각각 int이고 string 인지 검증하고 실행되게 된다.

3. 데이터 수정

3-1. 새로운 router 추가

새로운 라우터를 추가하여 우리가 가지고 있는 alarm_list의 데이터를 수정해보자.

@router.put("/alarm/{alarm_id}")
async def update_alarm(alarm_data: AlarmData, alarm_id: int=Path(..., title="ID to be updated")) -> dict:
    for each_alarm in alarm_list:
        if each_alarm.id == alarm_id:
            each_alarm.alarm = alarm_data.alarm
            return {
                "message": "Your Alarm is successfully updated."
            }
        
    return {
        "message": "The ID which you gave doesn't exist."
    }

PUT 메소드로 /alarm/alarm_id 형태로 uri 가 전달 되면, 아래의 update_alarm() 함수가 실행된다. update_alarm 함수는 alarm_data와 변경하고자 하는 alarm_id 를 인자로 받는다. 그리고 아래의 로직은 우리가 미리 갖고 있던 alarm_list를 순서대로 조회하며 인자로 주어진 alarm_id 가 우리의 리스트에 존재한다면 alarm 데이터를 변경하는 것이다.

3-2. 테스트 해보기

먼저 위와 동일하게 서버를 실행해주고, 새로운 데이터 2개 를 추가한다.

curl -X 'POST' 'http://localhost:8080/new/alarm' -H 'accept: application/json' -H 'Content-Type: application/json' -d '{"id": 1, "alarm": "You pushed the first alarm in your list!"}'

curl -X 'POST' 'http://localhost:8080/new/alarm' -H 'accept: application/json' -H 'Content-Type: application/json' -d '{"id": 2, "alarm": "You pushed the second alarm in your list!"}'

다음의 명령어로 첫번째로 추가한 데이터를 수정해보자.

curl -X 'PUT' 'http://localhost:8080/new/alarm/1' -H 'accept: application/json' -H 'Content-Type: application/json' -d '{"alarm": "Your first Item is changed with this updated ONE!!!"}'

수정한 뒤에 GET 메소드 테스트 명령어를 수행해 데이터가 잘 수정되었는지 확인해보자.

curl -X 'GET' 'http://localhost:8080/new/alarm' -H 'accept: application/json'

4. 데이터 삭제

4-1. 새로운 router 추가

마지막으로 삭제에 대한 기능을 추가해보자. 다음과 같이 router에 추가한다.

@router.delete("/alarm/{alarm_id}")
async def delete_alarm(alarm_id: int) -> Dict:
    for each_index in range(len(alarm_list)):
        alarm = alarm_list[each_index]
        if alarm.id == alarm_id:
            alarm_list.pop(each_index)
            return {
                "message": "Alarm is successfully deleted."
            }
        
    return {
        "message": "The ID which you gave doesn't exist."
    }

로직은 간단하다. 전달된 id 값과 기존에 alarm_list 의 id 가 같으면 해당 데이터를 리스트에서 삭제하는 것이다.

4-2. 테스트 해보기.

먼저 위와 동일하게 서버를 실행해주고, 새로운 데이터 2개 를 추가한다.

curl -X 'POST' 'http://localhost:8080/new/alarm' -H 'accept: application/json' -H 'Content-Type: application/json' -d '{"id": 1, "alarm": "You pushed the first alarm in your list!"}'

curl -X 'POST' 'http://localhost:8080/new/alarm' -H 'accept: application/json' -H 'Content-Type: application/json' -d '{"id": 2, "alarm": "You pushed the second alarm in your list!"}'

첫번째 데이터를 삭제해보자.

curl -X 'DELETE' 'http://localhost:8080/new/alarm/1' -H 'accept: application/json'

삭제한 뒤에 GET 메소드 테스트 명령어를 수행해 데이터가 잘 수정되었는지 확인해보자.

curl -X 'GET' 'http://localhost:8080/new/alarm' -H 'accept: application/json'

5. 정리하기

이번 글에서는 Pydantic을 통한 데이터 검증 방법과 CRUD 어플리케이션을 만들어보면서 FastAPI 프레임워크에 한단계 친숙해지는 실습을 해보았다. 글을 다시 한 번 돌이켜보면, 우리가 기능을 추가할 때 하는 행동은 간단하다. 단지 Router 만 추가하면 되는 것이다! 어떤 uri (trigger)가 왔을 때, 어떤 동작을 하게끔 하는 것이 router이다. 이 router가 동작하게끔 하기 위해 부수적인 model과 pydantic 등의 기능을 추가하기만 하면 된다. 마지막으로, 오늘의 실습이 잘 동작하지 않는 사람들 위해 우리가 실습한 파일 디렉토리 구조와 main, model, router 코드 전체를 첨부하며 이번 글을 마무리 한다.

5-0. tree

./my_first_app
├── main.py
├── model.py
└── routers
    └── new_router.py

5-1. main.py

# main.py
from fastapi import FastAPI
from typing import Dict

from routers.new_router import router

app = FastAPI()

app.include_router(router)

@app.get("/")
async def helloworld() -> Dict:
    return {
        "message": "Hello World"
    }

5-2. new_router.py

from fastapi import APIRouter, Path
from typing import Dict
from model import Alarm, AlarmData

router = APIRouter(
    prefix="/new"
)

alarm_list = []

@router.get("/alarm")
async def show_alarm() -> Dict:
    return {
        "alarms": alarm_list
    }

@router.post("/alarm")
async def add_alarm(alarm: Alarm) -> Dict:
    alarm_list.append(alarm)
    return {
        "message": "Your ALARM added successfully."
    }

@router.put("/alarm/{alarm_id}")
async def update_alarm(alarm_data: AlarmData, alarm_id: int=Path(..., title="ID to be updated")) -> Dict:
    for each_alarm in alarm_list:
        if each_alarm.id == alarm_id:
            each_alarm.alarm = alarm_data.alarm
            return {
                "message": "Your Alarm is successfully updated."
            }
        
    return {
        "message": "The ID which you gave doesn't exist."
    }

@router.delete("/alarm/{alarm_id}")
async def delete_alarm(alarm_id: int) -> Dict:
    for each_index in range(len(alarm_list)):
        alarm = alarm_list[each_index]
        if alarm.id == alarm_id:
            alarm_list.pop(each_index)
            return {
                "message": "Alarm is successfully deleted."
            }
        
    return {
        "message": "The ID which you gave doesn't exist."
    }

@router.get("/test")
async def helloworld_new( ) -> Dict:
    return {
        "message": "new router Hello!"
        }

5-3. model.py

# model.py

from pydantic import BaseModel

class Alarm(BaseModel):
    id: int
    alarm: str

class AlarmData(BaseModel):
    alarm: str