FastAPI 베이스캠프

CRUD 구현하기

1. 라우팅 및 세팅

1.1 URL 정보

이번 챕터의 URL 구성은 아래와 같습니다.

경로	함수명	메서드	설명
/items	read_items	GET	모든 물품 목록을 반환합니다.
/items/{item_id}	create_item	POST	새로운 물품을 등록합니다.
/items/{item_id}	read_item	GET	특정 물품의 상세 정보를 반환합니다.
/items/{item_id}	update_item	PUT	특정 물품의 정보를 업데이트합니다.
/items/{item_id}	delete_item	DELETE	특정 물품을 삭제합니다.

1.2 기본 세팅

이번 실습 폴더는 03_1_crud입니다. VSC 터미널에서 사용할 명령어 입니다. 가상환경은 벗어난 상태에서 실행해야 합니다. 만약 터미널 입력창 앞에 (venv)라고 되어 있다면 deactivate 명령어로 가상환경을 나간 상태에서 cd ..으로 상위 폴더로 나와 아래 명령어를 실행해주세요.

mkdir 03_1_crud
cd 03_1_crud
python -m venv venv
.\venv\Scripts\activate
pip install fastapi
pip install uvicorn

mkdir 03_1_crud
cd 03_1_crud
python -m venv venv
.\venv\Scripts\activate
pip install fastapi
pip install uvicorn

2. CRUD 애플리케이션 소개

CRUD는 Create(생성), Read(읽기), Update(갱신), Delete(삭제)의 앞글자를 따서 만든 약어로, 대부분의 웹 애플리케이션에서 기본이 되는 네 가지 핵심 기능을 의미합니다. 이 네 가지 기능은 데이터를 다루는 거의 모든 애플리케이션에서 필수적인 요소입니다.

Create(생성): 새로운 데이터를 시스템에 추가하는 기능입니다. 예를 들어, 새로운 사용자 계정을 만들거나 새 상품을 데이터베이스에 추가하는 것이 이에 해당합니다.
Read(읽기): 저장된 데이터를 조회하는 기능입니다. 단일 항목을 조회하거나 여러 항목의 목록을 가져오는 것 모두 읽기 작업에 해당합니다.
Update(갱신): 기존 데이터를 수정하는 기능입니다. 사용자 정보 변경이나 상품 가격 수정 등이 이에 해당합니다.
Delete(삭제): 시스템에서 데이터를 제거하는 기능입니다. 사용자 계정 삭제나 재고에서 상품 제거 등이 여기에 해당합니다.

이번 챕터에서는 FastAPI를 사용하여 간단한 CRUD 애플리케이션을 구현해 보겠습니다. 우리의 예제에서는 '아이템' 관리 시스템을 만들 것입니다. 사용자는 아이템을 생성하고, 조회하고, 수정하고, 삭제할 수 있습니다.

이 예제에서는 데이터베이스 대신 메모리 내 파이썬 데이터 구조를 사용하여 데이터를 저장할 것입니다. 이는 개념을 간단히 설명하기 위한 것이며, 실제 애플리케이션에서는 보통 영구적인 저장소인 데이터베이스를 사용합니다.

FastAPI는 자동으로 대화형 API 문서(Swagger UI)를 생성하므로, API를 쉽게 테스트하고 사용할 수 있습니다.

3. 기본 설정

먼저 필요한 모듈을 임포트하고 FastAPI 애플리케이션을 생성합니다. 데이터는 items라는 딕셔너리 객체에 저장할 것입니다. 이 딕셔너리는 메모리 내 데이터 저장소로 사용됩니다. 이 딕셔너리는 아이템 ID를 키로 사용하고, 아이템 정보를 값으로 사용합니다. 이 딕셔너리는 애플리케이션 실행 중에만 유지되며, 애플리케이션을 다시 시작하면 초기화됩니다.

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
 
app = FastAPI()
 
# 메모리 내 데이터 저장소
items = {}

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
 
app = FastAPI()
 
# 메모리 내 데이터 저장소
items = {}

4. 모델 정의

아이템을 표현할 Pydantic 모델을 정의합니다. 파이썬 3.9 이전 버전에서 사용했던 코드를 먼저 살펴보도록 하겠습니다. 아래코드도 많이 사용합니다.

class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
 
class ItemUpdate(BaseModel):
    name: Optional[str] = None
    description: Optional[str] = None
    price: Optional[float] = None

class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
 
class ItemUpdate(BaseModel):
    name: Optional[str] = None
    description: Optional[str] = None
    price: Optional[float] = None

Optional[str] = None은 해당 필드가 선택적이라는 것을 의미합니다. 즉, 필수가 아니라는 뜻입니다. 파이썬 3.9 이상에서는 str | None으로 대체할 수 있습니다. 이전 버전에는 |연산자가 없기 때문에 동작하지 않습니다. 가능하면 최신 버전을 사용하는 것이 좋습니다. 이 코드를 최신 버전으로 변경하면 아래와 같은 코드가 됩니다. 모듈도 포함시킬 필요가 없습니다.

class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
 
class ItemUpdate(BaseModel):
    name: str | None = None
    description: str | None = None
    price: float | None = None

class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
 
class ItemUpdate(BaseModel):
    name: str | None = None
    description: str | None = None
    price: float | None = None

우리 수업에서는 최신 버전에 맞춰서 코드를 작성하겠습니다.

5. Create 기능 구현

새로운 아이템을 생성하는 엔드포인트를 구현합니다.

@app.post("/items/{item_id}", response_model=Item)
async def create_item(item_id: int, item: Item):
    if item_id in items:
        raise HTTPException(status_code=400, detail="Item already exists")
    items[item_id] = item
    return item

@app.post("/items/{item_id}", response_model=Item)
async def create_item(item_id: int, item: Item):
    if item_id in items:
        raise HTTPException(status_code=400, detail="Item already exists")
    items[item_id] = item
    return item

썬더 클라이언트에서 POST 요청을 보내면 아이템이 생성되고, 생성된 아이템이 반환됩니다. 만약 이미 존재하는 아이템 ID를 사용하려고 하면 400 에러가 발생합니다. 아이템은 2개를 생성하도록 하겠습니다.

여기까지 구현된 main.py의 전체 코드는 아래와 같습니다.

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
 
app = FastAPI()
 
# 메모리 내 데이터 저장소
items = {}
 
 
class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
 
 
class ItemUpdate(BaseModel):
    name: str | None = None
    description: str | None = None
    price: float | None = None
 
 
@app.post("/items/{item_id}", response_model=Item)
async def create_item(item_id: int, item: Item):
    if item_id in items:
        raise HTTPException(status_code=400, detail="Item already exists")
    items[item_id] = item
    return item

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
 
app = FastAPI()
 
# 메모리 내 데이터 저장소
items = {}
 
 
class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
 
 
class ItemUpdate(BaseModel):
    name: str | None = None
    description: str | None = None
    price: float | None = None
 
 
@app.post("/items/{item_id}", response_model=Item)
async def create_item(item_id: int, item: Item):
    if item_id in items:
        raise HTTPException(status_code=400, detail="Item already exists")
    items[item_id] = item
    return item

썬더 클라이언트에서 다음과 같이 POST 요청을 보내면 아이템이 생성됩니다.

POST 127.0.0.1:8000/items/1
{
    "name": "item1",
    "description": "This is item1",
    "price": 100.0
}

POST 127.0.0.1:8000/items/2
{
    "name": "item2",
    "description": "This is item2",
    "price": 100.0
}

POST 127.0.0.1:8000/items/1
{
    "name": "item1",
    "description": "This is item1",
    "price": 100.0
}

POST 127.0.0.1:8000/items/2
{
    "name": "item2",
    "description": "This is item2",
    "price": 100.0
}

6. Read 기능 구현

아이템 목록을 조회하는 엔드포인트와 특정 아이템을 조회하는 엔드포인트를 구현합니다. 위 구현된 전체 코드 아래에 추가합니다.

@app.get("/items", response_model=List[Item])
async def read_items():
    return list(items.values())
 
@app.get("/items/{item_id}", response_model=Item)
async def read_item(item_id: int):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
    return items[item_id]

@app.get("/items", response_model=List[Item])
async def read_items():
    return list(items.values())
 
@app.get("/items/{item_id}", response_model=Item)
async def read_item(item_id: int):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
    return items[item_id]

썬더 클라이언트에서 GET 요청을 보내면 아이템 목록을 조회할 수 있습니다. 다만 위 코드가 수정되었기 때문에 지금 메모리 영역에는 아이템들이 없습니다. 다시 POST로 데이터를 넣어야 합니다. 이렇게 여러개의 아이템을 생성하고, 조회하고, 지우고, 수정하는 것을 할 때에는 썬더 클라이언트에 collections를 사용하면 편리합니다. 모든 코드를 구현한 다음 썬더 클라이언트 collections를 이용하여 테스트 해보도록 하겠습니다.

GET 127.0.0.1:8000/items

GET 127.0.0.1:8000/items

7. Update 기능 구현

기존 아이템을 업데이트하는 엔드포인트를 구현합니다.

@app.put("/items/{item_id}", response_model=Item)
async def update_item(item_id: int, item: ItemUpdate):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
 
    stored_item = items[item_id]
    update_data = item.dict(exclude_unset=True)
    updated_item = stored_item.copy(update=update_data)
    items[item_id] = updated_item
    return updated_item

@app.put("/items/{item_id}", response_model=Item)
async def update_item(item_id: int, item: ItemUpdate):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
 
    stored_item = items[item_id]
    update_data = item.dict(exclude_unset=True)
    updated_item = stored_item.copy(update=update_data)
    items[item_id] = updated_item
    return updated_item

썬더 클라이언트에서 PUT 요청을 보내면 아이템이 업데이트되고, 업데이트된 아이템이 반환됩니다. 모든 필드가 수정이 안될 수도 있으므로 기본값이 None인 ItemUpdate 모델을 사용합니다.

PUT 127.0.0.1:8000/items/1
{
    "name": "item1_updated",
    "price": 200.0
}

PUT 127.0.0.1:8000/items/1
{
    "name": "item1_updated",
    "price": 200.0
}

8. Delete 기능 구현

아이템을 삭제하는 엔드포인트를 구현합니다.

@app.delete("/items/{item_id}", response_model=Item)
async def delete_item(item_id: int):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
 
    item = items.pop(item_id)
    return item

@app.delete("/items/{item_id}", response_model=Item)
async def delete_item(item_id: int):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
 
    item = items.pop(item_id)
    return item

썬더 클라이언트에서 DELETE 요청을 보내면 아이템이 삭제되고, 삭제된 아이템이 반환됩니다. 보통은 삭제된 아이템을 반환하지 않고, 삭제되었다는 메시지만 반환합니다.

9. 전체 코드

이제 모든 CRUD 기능을 구현한 전체 코드를 살펴보겠습니다.

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
 
app = FastAPI()
 
# 메모리 내 데이터 저장소
items = {}
 
 
class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
 
 
class ItemUpdate(BaseModel):
    name: str | None = None
    description: str | None = None
    price: float | None = None
 
 
@app.post("/items/{item_id}", response_model=Item)
async def create_item(item_id: int, item: Item):
    if item_id in items:
        raise HTTPException(status_code=400, detail="Item already exists")
    items[item_id] = item
    return item
 
 
@app.get("/items", response_model=List[Item])
async def read_items():
    return list(items.values())
 
 
@app.get("/items/{item_id}", response_model=Item)
async def read_item(item_id: int):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
    return items[item_id]
 
 
@app.put("/items/{item_id}", response_model=Item)
async def update_item(item_id: int, item: ItemUpdate):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
 
    stored_item = items[item_id]
    update_data = item.dict(exclude_unset=True)
    updated_item = stored_item.copy(update=update_data)
    items[item_id] = updated_item
    return updated_item
 
 
@app.delete("/items/{item_id}", response_model=Item)
async def delete_item(item_id: int):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
 
    item = items.pop(item_id)
    return item
 
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8000, reload=True)

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
 
app = FastAPI()
 
# 메모리 내 데이터 저장소
items = {}
 
 
class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
 
 
class ItemUpdate(BaseModel):
    name: str | None = None
    description: str | None = None
    price: float | None = None
 
 
@app.post("/items/{item_id}", response_model=Item)
async def create_item(item_id: int, item: Item):
    if item_id in items:
        raise HTTPException(status_code=400, detail="Item already exists")
    items[item_id] = item
    return item
 
 
@app.get("/items", response_model=List[Item])
async def read_items():
    return list(items.values())
 
 
@app.get("/items/{item_id}", response_model=Item)
async def read_item(item_id: int):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
    return items[item_id]
 
 
@app.put("/items/{item_id}", response_model=Item)
async def update_item(item_id: int, item: ItemUpdate):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
 
    stored_item = items[item_id]
    update_data = item.dict(exclude_unset=True)
    updated_item = stored_item.copy(update=update_data)
    items[item_id] = updated_item
    return updated_item
 
 
@app.delete("/items/{item_id}", response_model=Item)
async def delete_item(item_id: int):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
 
    item = items.pop(item_id)
    return item
 
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8000, reload=True)

맨 마지막 줄 코드는 아래와 같이 파이썬 코드를 실행하게 해주는 코드입니다.

python main.py

python main.py

10. 썬더 클라이언트 collections 사용하기

썬더 클라이언트에서 collections를 사용하여 CRUD 애플리케이션을 테스트해보겠습니다. 이렇게 collections로 만들면 한 번에 여러개의 요청을 보낼 수 있어서 편리합니다.

우선 filter collections 옆에 메뉴를 클릭해 New Collection을 선택합니다. 새로운 컬랙션 이름을 생성하라고 나오면 fastapi라고 입력하고 Enter를 누릅니다.

생성된 collections에서 이름 옆에 메뉴를 클릭해 New Request를 선택합니다. 새로운 요청 이름을 생성하라고 나오면 생성 이라고 입력하고 Enter를 누릅니다.

아래 순서에 따라 차례대로 POST, GET, PUT, DELETE 요청을 생성하고, 각각의 요청에 대한 정보를 입력합니다. 이름은 편한대로 지정하시면 됩니다.

POST 요청 생성

Method: POST
URL: http://127.0.0.1:8000/items/1

Body:

{
      "name": "item1",
      "description": "This is item1",
      "price": 100.0
}

{
      "name": "item1",
      "description": "This is item1",
      "price": 100.0
}

POST 요청 생성

Method: POST
URL: http://127.0.0.1:8000/items/2

Body:

{
      "name": "item2",
      "description": "This is item2",
      "price": 100.0
}

{
      "name": "item2",
      "description": "This is item2",
      "price": 100.0
}

GET 요청 생성
- Method: GET
- URL: http://127.0.0.1:8000/items

PUT 요청 생성

Method: PUT
URL: http://127.0.0.1:8000/items/1

Body:

{
      "name": "item1_updated",
      "price": 200.0
}

{
      "name": "item1_updated",
      "price": 200.0
}

DELETE 요청 생성
- Method: DELETE
- URL: http://127.0.0.1:8000/items/1

이렇게 생성된 요청들을 실행하면서 CRUD 애플리케이션을 테스트해보세요. 완성된 collections는 다음과 같습니다.

중간중간 GET을 만들어 조회를 하면서 확인해보세요. 이렇게 collections를 사용하면 여러 요청을 한 번에 보내고, 결과를 한 눈에 볼 수 있어서 편리합니다.

연습문제

위의 CRUD 애플리케이션에 검색 기능을 추가해보세요. 아이템 이름으로 검색할 수 있는 엔드포인트를 구현하세요.
아이템에 "category" 필드를 추가하고, 카테고리별로 아이템을 필터링하는 기능을 구현해보세요.
아이템 생성 시 자동으로 고유한 ID를 생성하는 기능을 추가해보세요. (힌트: itertools.count()를 사용할 수 있습니다)