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()
비동기 작업 처리는 재능넷과 같은 플랫폼에서 매우 유용할 수 있습니다. 예를 들어:
- 대량의 이메일 발송 (예: 새로운 프로젝트 알림)
- 대용량 파일 업로드 및 처리 (예: 포트폴리오 이미지)
- 외부 API와의 통신 (예: 결제 시스템)
- 주기적인 데이터 분석 및 리포트 생성
이러한 작업들을 비동기적으로 처리함으로써, 사용자 경험을 향상시키고 서버 리소스를 효율적으로 사용할 수 있습니다. 🚀
7.4 비동기 작업의 장점
- 향상된 성능: I/O 바운드 작업을 비동기적으로 처리하여 전체적인 응답 시간을 줄일 수 있습니다.
- 리소스 효율성: 동시에 많은 요청을 처리할 수 있어 서버 리소스를 효율적으로 사용할 수 있습니다.
- 사용자 경험 개선: 긴 작업을 백그라운드에서 처리함으로써 사용자에게 빠른 응답을 제공할 수 있습니다.
- 확장성: 비동기 처리는 시스템의 확장성을 크게 향상시킵니다.
다음 섹션에서는 FastAPI 애플리케이션의 테스트와 배포에 대해 알아보겠습니다. 이는 안정적이고 신뢰할 수 있는 서비스를 제공하는 데 필수적인 단계입니다. 🧪🚀