FastAPI로 고성능 API 서버 구축하기

Python 개발자들에게 반가운 소식! FastAPI를 사용하여 고성능 API 서버를 구축하는 방법에 대해 상세히 알아보겠습니다. 이 가이드는 FastAPI의 기본 개념부터 고급 기능까지 다루며, 실제 프로젝트에 적용할 수 있는 실용적인 팁과 트릭을 제공합니다. 🚀

FastAPI는 현대적이고 빠르며(고성능), 파이썬 표준 타입 힌트에 기반한 Python 3.6+ 웹 프레임워크로, API를 구축하는 데 특화되어 있습니다. 자동 문서 생성, 보안, 의존성 주입 등 다양한 기능을 제공하여 개발자의 생산성을 크게 향상시킵니다.

이 가이드를 통해 여러분은 FastAPI의 강력한 기능을 활용하여 효율적이고 확장 가능한 API 서버를 구축하는 방법을 배우게 될 것입니다. 재능넷과 같은 플랫폼에서 API 서비스를 제공하거나 개선하고자 하는 개발자들에게 특히 유용할 것입니다. 😊

1. FastAPI 소개 및 설치

FastAPI는 현대적인 Python 웹 프레임워크로, 빠른 성능과 개발 속도를 동시에 제공합니다. RESTful API를 구축하는 데 특화되어 있으며, 자동 문서화, 타입 검사, 비동기 지원 등 다양한 기능을 제공합니다.

1.1 FastAPI의 주요 특징

빠른 성능: Starlette과 Pydantic을 기반으로 하여 NodeJS 및 Go와 대등한 수준의 매우 높은 성능을 제공합니다.
쉬운 사용: 직관적인 API로 빠르게 개발할 수 있습니다.
적은 버그: 사람(프로그래머)에 의한 오류를 약 40% 줄입니다.
직관적: 훌륭한 편집기 지원. 모든 곳에서 자동완성. 적은 디버깅 시간.
쉬움: 쉽게 사용하고 배우도록 설계되었습니다. 적은 문서 읽기 시간.
짧음: 코드 중복을 최소화합니다. 각 매개변수 선언의 여러 기능. 적은 버그.
견고함: 프로덕션 준비가 된 코드를 얻습니다. 자동 대화형 문서와 함께.
표준 기반: API에 대한 (완전히 호환되는) 개방형 표준 기반: OpenAPI (이전의 Swagger) 및 JSON 스키마.

1.2 FastAPI 설치하기

FastAPI를 설치하는 방법은 매우 간단합니다. Python 3.6 이상이 설치된 환경에서 다음 명령어를 실행하면 됩니다:

pip install fastapi

또한, FastAPI 애플리케이션을 실행하기 위해 ASGI 서버가 필요합니다. 가장 일반적으로 사용되는 것은 Uvicorn입니다:

pip install uvicorn

이제 FastAPI와 Uvicorn이 설치되었습니다. 다음 섹션에서는 첫 번째 FastAPI 애플리케이션을 만들어 보겠습니다. 🛠️

2. 첫 번째 FastAPI 애플리케이션 만들기

이제 FastAPI를 설치했으니, 첫 번째 애플리케이션을 만들어 보겠습니다. 이 과정을 통해 FastAPI의 기본 구조와 작동 방식을 이해할 수 있습니다.

2.1 기본 애플리케이션 구조

다음은 가장 기본적인 FastAPI 애플리케이션의 구조입니다:

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Hello World"}

이 코드를 main.py 파일에 저장합니다. 각 줄의 의미를 살펴보겠습니다:

from fastapi import FastAPI: FastAPI 클래스를 임포트합니다.
app = FastAPI(): FastAPI 인스턴스를 생성합니다. 이것이 우리 애플리케이션의 주요 지점이 됩니다.
@app.get("/"): 경로 연산 데코레이터. 이는 FastAPI에게 바로 아래 함수가 GET 요청을 처리하며, 경로 "/"에 해당한다고 알려줍니다.
async def root():: 경로 "/"에 대한 요청을 처리할 함수를 정의합니다.
return {"message": "Hello World"}: JSON 응답을 반환합니다.

2.2 애플리케이션 실행하기

애플리케이션을 실행하려면 터미널에서 다음 명령어를 입력합니다:

uvicorn main:app --reload

이 명령어의 각 부분을 설명하면:

main: 파일 main.py (Python 모듈).
app: main.py 내부에서 생성한 객체 app = FastAPI().
--reload: 코드 변경 시 서버를 자동으로 재시작합니다. 개발 시에만 사용하세요.

이제 브라우저에서 http://127.0.0.1:8000에 접속하면 "Hello World" 메시지를 볼 수 있습니다.

2.3 자동 API 문서

FastAPI의 강력한 기능 중 하나는 자동 API 문서 생성입니다. 두 가지 문서 인터페이스를 제공합니다:

http://127.0.0.1:8000/docs: Swagger UI를 사용한 대화형 API 문서
http://127.0.0.1:8000/redoc: ReDoc을 사용한 대안적 API 문서

이러한 문서는 자동으로 생성되며, API의 모든 엔드포인트와 그 사용법을 보여줍니다. 이는 API를 테스트하고 다른 개발자들과 공유하는 데 매우 유용합니다. 🔍

이렇게 간단한 코드로 강력한 API 서버를 구축할 수 있습니다. FastAPI는 이러한 기본 구조를 바탕으로 복잡한 비즈니스 로직, 데이터베이스 연동, 인증 등을 쉽게 추가할 수 있도록 설계되어 있습니다. 다음 섹션에서는 좀 더 복잡한 기능들을 살펴보겠습니다. 💡

3. 경로 매개변수와 쿼리 매개변수

FastAPI에서는 URL의 일부를 동적으로 처리하는 경로 매개변수와 URL 뒤에 붙는 쿼리 매개변수를 쉽게 처리할 수 있습니다. 이를 통해 유연하고 동적인 API를 구축할 수 있습니다.

3.1 경로 매개변수

경로 매개변수는 URL의 일부로, 중괄호 {}로 둘러싸여 있습니다. 예를 들어:

from fastapi import FastAPI

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

이 예제에서 {item_id}는 경로 매개변수입니다. item_id: int로 선언함으로써 FastAPI는 자동으로 이 값을 정수로 변환하고 유효성을 검사합니다.

3.2 쿼리 매개변수

쿼리 매개변수는 URL의 ? 뒤에 오는 키-값 쌍입니다. FastAPI에서는 함수 매개변수로 쉽게 선언할 수 있습니다:

from fastapi import FastAPI

app = FastAPI()

@app.get("/items/")
async def read_item(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": limit}

이 예제에서 skip과 limit은 쿼리 매개변수입니다. 예를 들어, /items/?skip=20&limit=30과 같이 사용할 수 있습니다.

3.3 경로 매개변수와 쿼리 매개변수 조합

경로 매개변수와 쿼리 매개변수를 함께 사용할 수도 있습니다:

from fastapi import FastAPI

app = FastAPI()

@app.get("/users/{user_id}/items/")
async def read_user_item(
    user_id: int, item_id: str, q: str = None, short: bool = False
):
    item = {"item_id": item_id, "owner_id": user_id}
    if q:
        item.update({"q": q})
    if not short:
        item.update(
            {"description": "This is an amazing item that has a long description"}
        )
    return item

이 예제에서 user_id는 경로 매개변수이고, item_id, q, short는 쿼리 매개변수입니다.

경로 매개변수와 쿼리 매개변수를 적절히 활용하면 유연하고 강력한 API를 설계할 수 있습니다. 경로 매개변수는 필수적이고 URL 구조의 일부를 형성하는 데이터에 사용하고, 쿼리 매개변수는 선택적이거나 필터링, 정렬 등의 추가 정보를 전달하는 데 사용합니다.

이러한 매개변수들을 활용하면 재능넷과 같은 플랫폼에서 사용자 프로필, 재능 목록, 검색 기능 등을 구현할 때 매우 유용할 것입니다. 예를 들어, /users/{user_id}/talents/?category=design&sort=rating와 같은 URL로 특정 사용자의 디자인 카테고리 재능을 평점 순으로 정렬하여 가져올 수 있습니다. 🎨✨

다음 섹션에서는 요청 본문(Request Body)을 처리하는 방법에 대해 알아보겠습니다. 이를 통해 클라이언트로부터 더 복잡한 데이터를 받아 처리할 수 있게 됩니다.

4. 요청 본문(Request Body) 처리하기

API를 개발할 때, 클라이언트로부터 복잡한 데이터를 받아야 하는 경우가 많습니다. 이럴 때 요청 본문(Request Body)을 사용합니다. FastAPI에서는 Pydantic 모델을 사용하여 요청 본문을 쉽게 정의하고 검증할 수 있습니다.

4.1 Pydantic 모델 정의하기

먼저, Pydantic을 사용하여 데이터 모델을 정의합니다:

from pydantic import BaseModel

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

이 모델은 아이템의 구조를 정의합니다. name과 price는 필수 필드이고, description과 tax는 선택적 필드입니다.

4.2 요청 본문 처리하기

이제 이 모델을 사용하여 요청 본문을 처리하는 엔드포인트를 만들어 봅시다:

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

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

@app.post("/items/")
async def create_item(item: Item):
    item_dict = item.dict()
    if item.tax:
        price_with_tax = item.price + item.tax
        item_dict.update({"price_with_tax": price_with_tax})
    return item_dict

이 예제에서:

@app.post("/items/"): POST 요청을 처리하는 엔드포인트를 정의합니다.
async def create_item(item: Item): Item 타입의 매개변수를 받습니다. FastAPI는 자동으로 요청 본문을 파싱하여 Item 객체로 변환합니다.
함수 내부에서는 받은 데이터를 처리하고 응답을 반환합니다.

4.3 요청 본문 검증

Pydantic 모델을 사용하면 자동으로 데이터 검증이 이루어집니다. 예를 들어, price에 문자열이 전달되면 자동으로 오류가 발생합니다. 추가적인 검증 로직도 쉽게 추가할 수 있습니다:

from pydantic import BaseModel, Field

class Item(BaseModel):
    name: str = Field(..., min_length=1, max_length=50)
    description: str = Field(None, max_length=200)
    price: float = Field(..., gt=0)
    tax: float = Field(None, ge=0)

이 예제에서는 Field를 사용하여 각 필드에 대한 추가적인 검증 규칙을 정의했습니다.

요청 본문을 처리하는 것은 API 개발에서 매우 중요한 부분입니다. FastAPI와 Pydantic을 사용하면 복잡한 데이터 구조도 쉽게 처리하고 검증할 수 있습니다. 이는 재능넷과 같은 플랫폼에서 사용자 프로필 업데이트, 새로운 재능 등록, 프로젝트 제안 등의 기능을 구현할 때 매우 유용합니다. 🚀

예를 들어, 재능 등록 API를 다음과 같이 구현할 수 있습니다:

from fastapi import FastAPI
from pydantic import BaseModel, Field

app = FastAPI()

class Talent(BaseModel):
    title: str = Field(..., min_length=5, max_length=100)
    description: str = Field(..., min_length=20, max_length=1000)
    category: str
    price_per_hour: float = Field(..., gt=0)
    skills: list[str] = Field(..., min_items=1, max_items=10)

@app.post("/talents/")
async def create_talent(talent: Talent):
    # 여기서 데이터베이스에 저장하는 로직을 구현할 수 있습니다.
    return {"message": "Talent registered successfully", "talent": talent}

이 예제에서는 재능 등록에 필요한 데이터를 Talent 모델로 정의하고, 각 필드에 대한 검증 규칙을 설정했습니다. 이를 통해 클라이언트로부터 받은 데이터가 유효한지 자동으로 확인할 수 있습니다.

다음 섹션에서는 데이터베이스 연동에 대해 알아보겠습니다. 실제 애플리케이션에서는 받은 데이터를 영구적으로 저장하고 관리해야 하므로, 데이터베이스 연동은 매우 중요한 주제입니다. 💾

5. 데이터베이스 연동

FastAPI는 다양한 데이터베이스와 쉽게 연동할 수 있습니다. 여기서는 SQLAlchemy를 사용하여 관계형 데이터베이스와 연동하는 방법을 살펴보겠습니다. SQLAlchemy는 Python에서 가장 널리 사용되는 ORM(Object-Relational Mapping) 라이브러리입니다.

5.1 SQLAlchemy 설정

먼저 필요한 라이브러리를 설치합니다:

pip install sqlalchemy

그리고 데이터베이스 연결을 설정합니다:

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
# PostgreSQL을 사용하는 경우:
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

5.2 모델 정의

SQLAlchemy 모델을 정의합니다:

from sqlalchemy import Column, Integer, String, Float

class TalentModel(Base):
    __tablename__ = "talents"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    description = Column(String)
    category = Column(String, index=True)
    price_per_hour = Column(Float)

Base.metadata.create_all(bind=engine)

5.3 데이터베이스 작업을 위한 의존성 함수

데이터베이스 세션을 관리하기 위한 의존성 함수를 생성합니다:

from fastapi import Depends
from sqlalchemy.orm import Session

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

5.4 CRUD 작업 구현

이제 데이터베이스 CRUD(Create, Read, Update, Delete) 작업을 구현할 수 있습니다:

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from pydantic import BaseModel

app = FastAPI()

class TalentCreate(BaseModel):
    title: str
    description: str
    category: str
    price_per_hour: float

class Talent(TalentCreate):
    id: int

    class Config:
        orm_mode = True

@app.post("/talents/", response_model=Talent)
def create_talent(talent: TalentCreate, db: Session = Depends(get_db)):
    db_talent = TalentModel(**talent.dict())
    db.add(db_talent)
    db.commit()
    db.refresh(db_talent)
    return db_talent

@app.get("/talents/{talent_id}", response_model=Talent)
def read_talent(talent_id: int, db: Session = Depends(get_db)):
    db_talent = db.query(TalentModel).filter(TalentModel.id == talent_id).first()
    if db_talent is None:
        raise HTTPException(status_code=404, detail="Talent not found")
    return db_talent

@app.get("/talents/", response_model=list[Talent])
def read_talents(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    talents = db.query(TalentModel).offset(skip).limit(limit).all()
    return talents

@app.put("/talents/{talent_id}", response_model=Talent)
def update_talent(talent_id: int, talent: TalentCreate, db: Session = Depends(get_db)):
    db_talent = db.query(TalentModel).filter(TalentModel.id == talent_id).first()
    if db_talent is None:
        raise HTTPException(status_code=404, detail="Talent not found")
    for key, value in talent.dict().items():
        setattr(db_talent, key, value)
    db.commit()
    db.refresh(db_talent)
    return db_talent

@app.delete("/talents/{talent_id}", response_model=Talent)
def delete_talent(talent_id: int, db: Session = Depends(get_db)):
    db_talent = db.query(TalentModel).filter(TalentModel.id == talent_id).first()
    if db_talent is None:
        raise HTTPException(status_code=404, detail="Talent not found")
    db.delete(db_talent)
    db.commit()
    return db_talent

이렇게 FastAPI와 SQLAlchemy를 사용하여 데이터베이스 연동을 구현할 수 있습니다. 이 구조를 사용하면 재능넷과 같은 플랫폼에서 사용자 정보, 재능 목록, 프로젝트 정보 등을 효율적으로 관리할 수 있습니다. 🗃️

데이터베이스 연동은 API의 핵심 기능 중 하나입니다. 이를 통해 데이터를 영구적으로 저장하고, 필요할 때 빠르게 검색하고 수정할 수 있습니다. 재능넷 플랫폼에서는 이러한 기능을 활용하여 사용자 프로필, 재능 목록, 프로젝트 정보, 리뷰 등을 관리할 수 있습니다.

다음 섹션에서는 인증과 권한 관리에 대해 알아보겠습니다. 이는 사용자의 개인 정보를 보호하고, 특정 작업에 대한 접근을 제한하는 데 필수적인 기능입니다. 🔐

6. 인증과 권한 관리

API의 보안은 매우 중요합니다. FastAPI에서는 JWT(JSON Web Tokens)를 사용한 인증 시스템을 쉽게 구현할 수 있습니다. 여기서는 간단한 JWT 기반 인증 시스템을 구현해 보겠습니다.

6.1 필요한 라이브러리 설치

pip install python-jose[cryptography] passlib[bcrypt]

6.2 JWT 토큰 생성 및 검증

from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext

SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def create_access_token(data: dict):
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

def verify_token(token: str):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except JWTError:
        return None

6.3 사용자 인증 엔드포인트

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel

app = FastAPI()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

class User(BaseModel):
    username: str
    email: str = None
    full_name: str = None
    disabled: bool = None

class UserInDB(User):
    hashed_password: str

def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)

def fake_hash_password(password: str):
    return "fakehashed" + password

fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "johndoe@example.com",
        "hashed_password": "fakehashedsecret",
        "disabled": False,
    }
}

def authenticate_user(fake_db, username: str, password: str):
    user = get_user(fake_db, username)
    if not user:
        return False
    if not fake_hash_password(password) == user.hashed_password:
        return False
    return user

@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token = create_access_token(data={"sub": user.username})
    return {"access_token": access_token, "token_type": "bearer"}

async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    payload = verify_token(token)
    if payload is None:
        raise credentials_exception
    username: str = payload.get("sub")
    if username is None:
        raise credentials_exception
    user = get_user(fake_users_db, username=username)
    if user is None:
        raise credentials_exception
    return user

@app.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_user)):
    return current_user

이 예제에서는 JWT를 사용한 기본적인 인증 시스템을 구현했습니다. 사용자가 로그인하면 JWT 토큰이 발급되고, 이후의 요청에서는 이 토큰을 사용하여 인증을 수행합니다. 🔑

실제 애플리케이션에서는 더 강력한 보안 조치가 필요할 수 있습니다. 예를 들어:

비밀번호 해싱에 더 강력한 알고리즘 사용 (예: bcrypt)
토큰 만료 시간 설정
HTTPS 사용
rate limiting 구현
CORS (Cross-Origin Resource Sharing) 설정

인증과 권한 관리는 재능넷과 같은 플랫폼에서 매우 중요합니다. 사용자의 개인 정보를 보호하고, 특정 작업(예: 재능 등록, 프로젝트 제안 등)에 대한 접근을 제한할 수 있습니다. 또한, 관리자 권한을 구현하여 플랫폼 관리에 필요한 특별한 기능에 대한 접근을 제어할 수 있습니다.

다음 섹션에서는 비동기 작업 처리에 대해 알아보겠습니다. 이는 시간이 오래 걸리는 작업을 효율적으로 처리하는 데 중요합니다. 🚀

7. 비동기 작업 처리

FastAPI는 비동기 프로그래밍을 완벽하게 지원합니다. 이를 통해 I/O 바운드 작업을 효율적으로 처리할 수 있으며, 특히 대규모 동시 요청을 처리할 때 유용합니다. 여기서는 비동기 작업 처리의 기본과 백그라운드 작업 처리 방법에 대해 알아보겠습니다.

7.1 비동기 엔드포인트

FastAPI에서 비동기 엔드포인트를 만드는 것은 매우 간단합니다:

from fastapi import FastAPI
import asyncio

app = FastAPI()

@app.get("/async-operation")
async def async_operation():
    # 시간이 걸리는 작업을 시뮬레이션합니다
    await asyncio.sleep(1)
    return {"message": "Async operation completed"}

7.2 백그라운드 작업

때로는 요청을 즉시 응답하고 시간이 오래 걸리는 작업은 백그라운드에서 처리해야 할 때가 있습니다. FastAPI에서는 BackgroundTasks를 사용하여 이를 구현할 수 있습니다:

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def long_running_task(name: str):
    # 실제로는 여기에 시간이 오래 걸리는 작업을 구현합니다
    print(f"Long running task for {name} is complete")

@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(long_running_task, email)
    return {"message": "Notification sent in the background"}

7.3 외부 서비스와의 비동기 통신

외부 API와 통신할 때도 비동기 라이브러리를 사용하면 성능을 크게 향상시킬 수 있습니다. 여기서는 httpx 라이브러리를 사용한 예제를 보여드리겠습니다:

from fastapi import FastAPI
import httpx

app = FastAPI()

@app.get("/fetch-external-data")
async def fetch_external_data():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.example.com/data")
    return response.json()