FastAPI에서 MongoDB 데이터 연동 CRUD 예제

안녕하세요. 아래 링크의 지난 글에서는 fastapi의 CRUD 메소드 사용 방법을 알아보았습니다.

FastAPI get, post, patch, delete 예제 코드 정리

 

이어서 이번에는 실제 데이터베이스를 연동해서 fastapi를 활용하는 방법을 살펴볼 것인데요.

key-value 형태로 데이터를 저장하는 경우 대표적으로 활용 가능한 NoSQL 데이터베이스인

MongoDB를 연동하는 예제를 여기서 살펴보도록 하겠습니다.

 

 

MongoDB 생성 및 파이썬에서 불러오기

만일, MongoDB 데이터베이스 생성이 필요하다면 아래의 튜토리얼을 참고하시면 됩니다.

MongoDB) 계정 만들고 db 생성하고 CRUD 해보기

 

해당 튜토리얼대로 생성했던 DB의 uri를 복사하고, cluster 이름을 기억해둡니다.

파이썬에서 DB를 불러오는 예시 코드는 다음과 같습니다.

(참고로 처음으로 데이터 추가 시 자동으로 해당 이름의 DB가 생성됩니다.)

from motor.motor_asyncio import AsyncIOMotorClient # pip install motor로 설치

uri = "your MongoDB uri"
client = AsyncIOMotorClient(uri)

def get_database():
    cluster = client["Cluster0"] # cluster 이름(Cluster0으로 생성되었다고 가정)
    db = cluster["users"] # DB 이름(users로 사용한다고 가정)
    return db

 

저장할 데이터의 양식으로는 다음과 같이 간단한 유저 데이터를 가정해보도록 하겠습니다.

from pydantic import BaseModel

# 데이터 모델 정의
class User(BaseModel):
    id: int
    username: str
    email: str

 

 

get 메소드 예시

먼저, id를 기준으로 데이터를 조회하는 get 메소드의 코드 예시를 살펴보겠습니다.

from fastapi import FastAPI, HTTPException

app = FastAPI()
db = get_database()

@app.get("/users/{user_id}", response_model=User)
async def read_user(user_id: int):
    user = await db.find_one({"id": user_id})
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return user

위의 코드에서는 id = user_id에 해당하는 유저를 find_one 메소드로 DB에서 찾아서

있으면 해당 user의 정보를 반환하고, 없으면 404 status_code로 응답하고 있습니다.

 

참고로, 각 메소드의 호출 방법에 대해서는 맨 위의 이전 글과 동일하므로,

여기서는 추가 언급을 생략하도록 하겠습니다.

 

 

post 메소드 예시

이번에는 새로운 데이터를 추가하는 post의 예제 코드입니다.

@app.post("/users/", response_model=User, status_code=201)
async def create_user(user: User):
    existing_user = await db.find_one({"id": user.id})
    if existing_user:
        raise HTTPException(status_code=400, detail="User with this ID already exists")
    await db.insert_one(user.dict())
    return user

이미 유저가 존재하는 경우는 400 에러 코드를 반환하는 상황을 가정하였고,

새로운 데이터 추가 시에는 dict로 변환 후 insert_one 메소드를 사용하여

DB에 직접 추가하도록 하였습니다.

 

 

patch 메소드 예시

기존 데이터를 수정하는 patch 메소드의 코드 스니펫은 다음과 같습니다.

from typing import Optional

@app.patch("/users/{user_id}", response_model=User)
async def update_user(user_id: int, username: Optional[str] = Body(default=None), email: Optional[str] = Body(default=None)):
    user = await db.find_one({"id": user_id})
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    
    update_data = {}
    # username / email이 input으로 안 들어오면 해당 값 변경 X
    if username is not None:
        update_data["username"] = username
    if email is not None:
        update_data["email"] = email
	
    if update_data:
        await db.update_one({"id": user_id}, {"$set": update_data})
        updated_user = await db.find_one({"id": user_id})
        return updated_user
    else:
        return user

id가 존재하지 않는 경우는 404 에러 코드를 반환하며,

username or email은 선택적으로 input으로 받는 상황을 가정하였습니다.

 

데이터가 업데이트 되는 부분은 update_one 메소드로 진행하게 되는데요.

여기서 $set 연산자는 변경할 key의 이름을 유연하게 선택해주는 역할을 수행합니다.

 

 

delete 메소드 예시

마지막으로 id를 기준으로 데이터를 삭제하는 delete 메소드 코드 예시입니다.

(id가 존재하지 않는다면 404 에러 반환)

@app.delete("/users/{user_id}", status_code=204)
async def delete_user(user_id: int):
    user = await db.find_one({"id": user_id})
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    
    result = await db.delete_one({"id": user_id})
    return None

 

여기서는 delete_one 메소드를 통하여 데이터 삭제를 진행하게 됩니다.