FastAPI 서버개발, 비동기 대응 CRUDs

최근 몇 년간 FastAPI는 Python에서 비동기 웹 프레임워크로 많은 인기를 얻고 있습니다. FastAPI는 비동기 프로그래밍의 장점을 제공하며, RESTful API를 쉽게 만들 수 있는 강력한 도구입니다. 이 글에서는 FastAPI를 사용하여 비동기적으로 CRUD(생성, 읽기, 수정, 삭제) 작업을 수행하는 방법에 대해 자세히 설명하겠습니다.

1. FastAPI란?

FastAPI는 Python으로 구축된 현대적인 웹 프레임워크로, 고성능 비동기 웹 API를 만드는 데 중점을 두고 설계되었습니다. 주요 특징으로는 다음과 같습니다:

  • 빠른 성능: Starlette와 Pydantic을 기반으로 하므로 속도가 매우 빠릅니다.
  • 자동 문서화: Swagger UI 및 ReDoc과 같은 자동화된 API 문서를 제공합니다.
  • 타입 힌트 지원: Python 타입 힌트를 사용하여 코드의 가독성을 높이고 오류를 줄일 수 있습니다.

2. 환경 설정

FastAPI를 시작하기 위해서는 Python과 pip(패키지 관리자)가 설치되어 있어야 합니다. 아래의 명령어로 FastAPI와 Uvicorn(ASGI 서버)를 설치할 수 있습니다:

pip install fastapi uvicorn

2.1. 프로젝트 구조 설정

간단한 프로젝트 구조를 설정합니다:


my_fastapi_app/
│
├── main.py        # FastAPI 애플리케이션
└── models.py      # Pydantic 모델

3. 기본 CRUD API 구현

이제 기본적인 CRUD 작업을 구현해보겠습니다. 우리는 사용할 데이터 모델을 정의하고, 이를 기반으로 CRUD 엔드포인트를 만들어 보겠습니다.

3.1. 모델 정의

우선 models.py 파일에 Pydantic 모델을 정의합니다:


from pydantic import BaseModel
from typing import Optional

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

3.2. CRUD 작업 구현

main.py 파일에 CRUD 작업을 구현합니다:


from fastapi import FastAPI, HTTPException
from typing import List
from models import Item

app = FastAPI()

# 메모리에 데이터를 저장하기 위한 리스트
fake_items_db = []

@app.post("/items/", response_model=Item)
async def create_item(item: Item):
    fake_items_db.append(item)
    return item

@app.get("/items/", response_model=List[Item])
async def read_items():
    return fake_items_db

@app.get("/items/{item_id}", response_model=Item)
async def read_item(item_id: int):
    item = next((item for item in fake_items_db if item.id == item_id), None)
    if item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

@app.put("/items/{item_id}", response_model=Item)
async def update_item(item_id: int, updated_item: Item):
    for index, item in enumerate(fake_items_db):
        if item.id == item_id:
            fake_items_db[index] = updated_item
            return updated_item
    raise HTTPException(status_code=404, detail="Item not found")

@app.delete("/items/{item_id}")
async def delete_item(item_id: int):
    global fake_items_db
    fake_items_db = [item for item in fake_items_db if item.id != item_id]
    return {"message": "Item deleted successfully"}

4. 비동기 처리

FastAPI는 비동기 프로그래밍을 지원합니다. 설정한 CRUD 작업은 이미 비동기 함수로 정의되어 있으므로, 비동기적 요청 처리를 자동으로 활용할 수 있습니다. 다수의 사용자가 동시에 요청을 보내더라도, FastAPI는 효율적으로 요청을 처리할 수 있습니다.

4.1. 비동기 데이터베이스 연결

실제로는 인메모리 데이터베이스를 사용하지 않고, 데이터베이스와의 비동기 연결을 사용하는 것이 일반적입니다. 아래는 Async SQLAlchemy를 사용한 예시입니다:


from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "sqlite+aiosqlite:///./test.db"

engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine, class_=AsyncSession)

# FastAPI 애플리케이션 생성
app = FastAPI()

# 데이터베이스 세션 종속성 정의
async def get_db() -> AsyncIterator[AsyncSession]:
    async with AsyncSessionLocal() as session:
        yield session

4.2. 비동기 CRUD 작업

이제 데이터베이스를 사용할 수 있으므로 CRUD 작업을 비동기 방식으로 업데이트합니다:


@app.post("/items/", response_model=Item)
async def create_item(item: Item, db: AsyncSession = Depends(get_db)):
    db.add(item)
    await db.commit()
    return item

@app.get("/items/", response_model=List[Item])
async def read_items(db: AsyncSession = Depends(get_db)):
    return await db.execute(select(Item)).scalars().all()

5. 테스트 및 디버깅

FastAPI는 자동으로 Swagger UI를 제공하므로, API를 테스트하는 데 매우 유용합니다. 서버를 실행하면 http://127.0.0.1:8000/docs에서 API 문서를 확인할 수 있습니다.

uvicorn main:app --reload

6. 결론

FastAPI는 비동기 프로그래밍을 통해 효율적인 API 서버를 구축할 수 있는 매우 강력한 도구입니다. 이번 글에서 소개한 CRUD 예제는 기본적인 개념을 이해하는 데 도움이 될 것입니다. 실제 프로젝트에서는 데이터베이스와의 연결, 사용자 인증, 그리고 비즈니스 로직을 추가하여 더욱 발전시킬 수 있습니다.

FastAPI의 장점인 고성능과 자동 문서화 기능을 활용하여 다양한 API를 손쉽게 개발해 보시기 바랍니다.

7. 참고 자료