Python FastAPI#
FastAPI는 2018년에 Sebastián Ramírez가 개발한 현대적인 Python 웹 프레임워크.
Python 3.6+ 의 타입 힌트를 기반으로 하며, 비동기 프로그래밍을 지원하는 고성능 웹 프레임워크.****
주요 개념#
- 타입 힌트: FastAPI는 파이썬의 타입 힌트를 적극적으로 활용하여 코드의 안정성과 가독성을 높인다.
- 비동기 프로그래밍: Starlette을 기반으로 하여 비동기 프로그래밍을 지원한다.
- 의존성 주입: 코드의 재사용성을 높이고 결합도를 낮추는 의존성 주입 시스템을 제공한다.
- Pydantic: 데이터 검증과 설정 관리를 위한 Pydantic 라이브러리를 사용한다.
- 빠른 성능: Starlette과 Pydantic을 기반으로 하여 NodeJS 및 Go와 대등한 수준의 높은 성능을 제공한다.
- 자동 문서화: Swagger UI와 ReDoc을 통해 API 문서를 자동으로 생성한다.
- 표준 기반: OpenAPI와 JSON Schema를 기반으로 한다.
- 쉬운 사용성: 직관적인 API로 빠른 개발이 가능하다.
- 빠른 개발 속도: 간결한 문법과 자동 문서화 기능으로 개발 속도가 빠르다.
- 높은 성능: 비동기 지원과 최적화된 코드로 높은 성능을 제공한다.
- 타입 안정성: 파이썬의 타입 힌트를 활용하여 코드의 안정성을 높인다.
- 현대적인 기능: 비동기 처리, 의존성 주입, 자동 검증 등 현대적인 기능을 지원한다.
- 상대적으로 작은 커뮤니티: 새로운 프레임워크이기 때문에 Django나 Flask에 비해 커뮤니티가 작다.
- 학습 곡선: 비동기 프로그래밍, 타입 힌트 등 현대적인 파이썬 개념에 익숙하지 않은 개발자에게는 학습 곡선이 있을 수 있다.
- 성숙도: 아직 1.0 버전이 출시되지 않아 안정성 면에서 우려가 있을 수 있다.
개발 특성#
- 모듈화: 앱 단위로 프로젝트를 구성하여 모듈화된 개발이 가능하다.
- RESTful API 개발: API 서버 구축에 최적화되어 있다.
- 마이크로서비스: 독립적인 서비스 컴포넌트 개발에 유용하다.
성능과 기능#
- 고성능: 비동기 처리와 최적화된 코드로 높은 성능을 제공한다.
- 데이터 검증: Pydantic을 통한 강력한 데이터 검증 기능을 제공한다.
- 보안: 내장된 보안 기능으로 SQL 인젝션, CSRF, XSS 등 다양한 보안 위협에 대응한다.
- 확장성: 다양한 미들웨어와 확장 기능을 지원한다.
서버 구현 예제#
1
2
3
4
5
6
7
8
9
10
11
| from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def root():
return {"message": "Hello World"}
@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str = None):
return {"item_id": item_id, "q": q}
|
이 예제는 루트 경로와 아이템 ID를 받는 두 개의 엔드포인트를 정의한다.
자주 사용되는 패키지#
- Uvicorn: ASGI 서버로 FastAPI 애플리케이션을 실행한다.
- SQLAlchemy: ORM을 통한 데이터베이스 작업에 사용된다.
- Pydantic: 데이터 검증과 설정 관리에 사용된다.
- Starlette: FastAPI의 기반이 되는 경량 ASGI 프레임워크.
- pytest: FastAPI 애플리케이션의 테스트에 사용된다.
- Alembic: 데이터베이스 마이그레이션
- Python-Jose: JWT 토큰 처리
- Passlib: 패스워드 해싱
- Python-Multipart: 폼 데이터 처리
- Gunicorn: 프로덕션 WSGI 서버
용어 정리#
참고 및 출처#
FastAPI
안녕하세요, 백엔드 개발자로서 FastAPI에 대해 자세히 설명해 드리겠습니다. FastAPI는 현대적인 파이썬 웹 프레임워크로, 기초부터 심화까지 단계별로 살펴보겠습니다.
FastAPI 종합 가이드#
1. FastAPI 개요#
FastAPI는 2018년 Sebastián Ramírez가 개발한 현대적이고 고성능의 파이썬 웹 프레임워크입니다. 이름에서 알 수 있듯이 빠른 API 개발에 중점을 둡니다.
주요 특징#
- 속도: 스타레트(Starlette)와 푸지(Pydantic)를 기반으로 하여 NodeJS 및 Go와 비슷한 수준의 높은 성능을 제공합니다.
- 타입 힌트: 파이썬의 타입 힌트를 활용하여 데이터 검증 및 자동 문서화를 지원합니다.
- 자동 문서화: Swagger UI와 ReDoc을 통해 API 문서를 자동으로 생성합니다.
- 비동기 지원: 비동기 프로그래밍을 완벽하게 지원합니다.
- 쉬운 학습 곡선: 직관적인 API 설계로 빠르게 학습하고 적용할 수 있습니다.
2. FastAPI 설치 및 기본 설정#
1
2
3
4
5
6
7
8
9
| # 가상환경 생성 (선택사항이지만 권장)
python -m venv fastapi-env
source fastapi-env/bin/activate # Windows: fastapi-env\Scripts\activate
# FastAPI 및 ASGI 서버 설치
pip install fastapi uvicorn
# 추가 기능을 위한 설치 (선택사항)
pip install "fastapi[all]" # 모든 선택적 의존성 설치
|
첫 번째 FastAPI 애플리케이션#
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
| # main.py
from fastapi import FastAPI
# FastAPI 인스턴스 생성
app = FastAPI()
# 루트 경로에 대한 GET 요청 핸들러 정의
@app.get("/")
def read_root():
return {"Hello": "World"}
# 경로 매개변수를 포함한 엔드포인트 정의
@app.get("/items/{item_id}")
def read_item(item_id: int, q: str = None):
return {"item_id": item_id, "q": q}
|
서버 실행#
1
2
| # 개발 서버 실행
uvicorn main:app --reload
|
이제 http://localhost:8000에서 API에 접근할 수 있으며, http://localhost:8000/docs에서 자동 생성된 문서를 확인할 수 있습니다.
3. 경로 매개변수 및 쿼리 매개변수#
경로 매개변수#
URL의 일부로 전달되는 변수입니다.
1
2
3
| @app.get("/users/{user_id}")
def read_user(user_id: int):
return {"user_id": user_id}
|
쿼리 매개변수#
URL의? 뒤에 오는 키-값 쌍입니다.
1
2
3
4
| @app.get("/users/")
def read_users(skip: int = 0, limit: int = 10):
# skip과 limit은 쿼리 매개변수
return {"skip": skip, "limit": limit}
|
혼합 사용#
1
2
3
4
5
6
| @app.get("/users/{user_id}/items/")
def read_user_items(
user_id: int, skip: int = 0, limit: int = 10, q: str = None
):
# user_id는 경로 매개변수, 나머지는 쿼리 매개변수
return {"user_id": user_id, "skip": skip, "limit": limit, "q": q}
|
4. 요청 본문 및 모델#
Pydantic 모델을 사용하여 요청 본문을 정의하고 검증합니다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
| 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/")
def create_item(item: Item):
# item은 자동으로 검증된 Pydantic 모델 인스턴스
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
|
5. 응답 모델#
응답의 형태를 명시적으로 정의할 수 있습니다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
| from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
class Item(BaseModel):
name: str
description: str = None
price: float
tax: float = None
class ItemResponse(BaseModel):
name: str
price: float
price_with_tax: float = None
@app.post("/items/", response_model=ItemResponse)
def create_item(item: Item):
# 응답은 ItemResponse 모델에 맞게 필터링됨
return {
"name": item.name,
"price": item.price,
"price_with_tax": item.price + (item.tax or 0)
}
|
6. 의존성 주입 시스템#
FastAPI는 강력한 의존성 주입 시스템을 제공합니다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
| from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer
app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
def get_current_user(token: str = Depends(oauth2_scheme)):
# 토큰 검증 로직 (실제로는 JWT 검증 등이 필요)
if token != "valid_token":
raise HTTPException(status_code=401, detail="Invalid token")
return {"username": "john_doe"}
@app.get("/users/me")
def read_users_me(current_user: dict = Depends(get_current_user)):
return current_user
|
7. 예외 처리#
HTTP 예외를 명시적으로 발생시킬 수 있습니다.
1
2
3
4
5
6
7
8
9
10
11
| from fastapi import FastAPI, HTTPException
app = FastAPI()
items = {"foo": "The Foo Item"}
@app.get("/items/{item_id}")
def read_item(item_id: str):
if item_id not in items:
raise HTTPException(status_code=404, detail="Item not found")
return {"item": items[item_id]}
|
8. 미들웨어#
요청/응답 처리 과정에 로직을 추가할 수 있습니다.
1
2
3
4
5
6
7
8
9
10
11
12
| from fastapi import FastAPI
import time
app = FastAPI()
@app.middleware("http")
async def add_process_time_header(request, call_next):
start_time = time.time()
response = await call_next(request)
process_time = time.time() - start_time
response.headers["X-Process-Time"] = str(process_time)
return response
|
9. 비동기 프로그래밍#
FastAPI는 비동기 프로그래밍을 완벽하게 지원합니다.
1
2
3
4
5
6
7
8
9
10
| import asyncio
from fastapi import FastAPI
app = FastAPI()
@app.get("/async-operation/")
async def read_async():
# 비동기 작업 수행
await asyncio.sleep(1)
return {"message": "Async operation completed"}
|
10. 데이터베이스 통합#
SQLAlchemy, Tortoise ORM 등을 통해 데이터베이스와 통합할 수 있습니다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
| from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
# 데이터베이스 설정
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
# 모델 정의
class UserDB(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
email = Column(String, unique=True, index=True)
username = Column(String, index=True)
password = Column(String)
# 데이터베이스 테이블 생성
Base.metadata.create_all(bind=engine)
app = FastAPI()
# 의존성: 데이터베이스 세션
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
# 사용자 생성 엔드포인트
@app.post("/users/")
def create_user(
email: str, username: str, password: str, db: Session = Depends(get_db)
):
db_user = UserDB(email=email, username=username, password=password)
db.add(db_user)
db.commit()
db.refresh(db_user)
return {"id": db_user.id, "email": db_user.email, "username": db_user.username}
|
11. API 보안#
FastAPI는 다양한 보안 메커니즘을 지원합니다.
OAuth2 및 JWT 인증#
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
| from datetime import datetime, timedelta
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel
# 보안 설정
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, expires_delta: timedelta = None):
to_encode = data.copy()
expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
# 모델
class User(BaseModel):
username: str
email: str = None
full_name: str = None
disabled: bool = None
class UserInDB(User):
hashed_password: str
# 가상 사용자 데이터베이스
fake_users_db = {
"johndoe": {
"username": "johndoe",
"full_name": "John Doe",
"email": "johndoe@example.com",
"hashed_password": pwd_context.hash("secret"),
"disabled": False,
}
}
app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
# 사용자 인증
def authenticate_user(fake_db, username: str, password: str):
user = get_user(fake_db, username)
if not user:
return False
if not pwd_context.verify(password, user.hashed_password):
return False
return user
# 사용자 가져오기
def get_user(db, username: str):
if username in db:
user_dict = db[username]
return UserInDB(**user_dict)
return None
# 현재 사용자 가져오기
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"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
raise credentials_exception
except JWTError:
raise credentials_exception
user = get_user(fake_users_db, username=username)
if user is None:
raise credentials_exception
return user
# 토큰 엔드포인트
@app.post("/token")
async def login_for_access_token(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_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = create_access_token(
data={"sub": user.username}, expires_delta=access_token_expires
)
return {"access_token": access_token, "token_type": "bearer"}
# 보호된 엔드포인트
@app.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_user)):
return current_user
|
12. 테스팅#
FastAPI는 TestClient를 통해 간편한 테스트를 지원합니다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
| from fastapi.testclient import TestClient
from main import app # FastAPI 앱 임포트
client = TestClient(app)
def test_read_root():
response = client.get("/")
assert response.status_code == 200
assert response.json() == {"Hello": "World"}
def test_read_item():
response = client.get("/items/42?q=test")
assert response.status_code == 200
assert response.json() == {"item_id": 42, "q": "test"}
|
13. 배포#
FastAPI 애플리케이션은 다양한 방법으로 배포할 수 있습니다.
Docker 컨테이너 배포#
1
2
3
4
5
6
7
8
9
10
| FROM python:3.9
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
|
Gunicorn과 함께 사용#
1
2
3
4
| pip install gunicorn
# 실행
gunicorn -w 4 -k uvicorn.workers.UvicornWorker main:app
|
14. 고급 기능#
웹소켓 지원#
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
| from fastapi import FastAPI, WebSocket, WebSocketDisconnect
app = FastAPI()
class ConnectionManager:
def __init__(self):
self.active_connections = []
async def connect(self, websocket: WebSocket):
await websocket.accept()
self.active_connections.append(websocket)
def disconnect(self, websocket: WebSocket):
self.active_connections.remove(websocket)
async def broadcast(self, message: str):
for connection in self.active_connections:
await connection.send_text(message)
manager = ConnectionManager()
@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int):
await manager.connect(websocket)
try:
while True:
data = await websocket.receive_text()
await websocket.send_text(f"Message: {data}")
await manager.broadcast(f"Client #{client_id} says: {data}")
except WebSocketDisconnect:
manager.disconnect(websocket)
await manager.broadcast(f"Client #{client_id} left the chat")
|
백그라운드 태스크#
1
2
3
4
5
6
7
8
9
10
11
12
13
14
| from fastapi import FastAPI, BackgroundTasks
app = FastAPI()
def write_log(message: str):
with open("log.txt", "a") as f:
f.write(message + "\n")
@app.post("/send-notification/{email}")
async def send_notification(
email: str, background_tasks: BackgroundTasks
):
background_tasks.add_task(write_log, f"Notification sent to {email}")
return {"message": "Notification sent in the background"}
|
15. 실제 프로젝트 구조#
대규모 FastAPI 프로젝트의 권장 구조:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
| myapp/
│
├── app/
│ ├── __init__.py
│ ├── main.py # FastAPI 앱 인스턴스 및 기본 설정
│ ├── dependencies.py # 의존성 함수
│ ├── routers/ # API 라우터
│ │ ├── __init__.py
│ │ ├── users.py
│ │ ├── items.py
│ │ └── …
│ ├── models/ # 데이터베이스 모델
│ │ ├── __init__.py
│ │ ├── user.py
│ │ ├── item.py
│ │ └── …
│ ├── schemas/ # Pydantic 모델
│ │ ├── __init__.py
│ │ ├── user.py
│ │ ├── item.py
│ │ └── …
│ ├── crud/ # CRUD 작업
│ │ ├── __init__.py
│ │ ├── user.py
│ │ ├── item.py
│ │ └── …
│ ├── core/ # 핵심 설정
│ │ ├── __init__.py
│ │ ├── config.py # 환경 변수 및 설정
│ │ └── security.py # 보안 관련 유틸리티
│ └── utils/ # 유틸리티 함수
│ ├── __init__.py
│ └── …
│
├── tests/ # 테스트
│ ├── __init__.py
│ ├── conftest.py
│ ├── test_users.py
│ └── …
│
├── alembic/ # 데이터베이스 마이그레이션
│ ├── versions/
│ └── …
│
├── logs/ # 로그 파일
├── .env # 환경 변수
├── .gitignore
├── requirements.txt
├── Dockerfile
└── docker-compose.yml
|
16. 성능 최적화 팁#
- 비동기 활용: I/O 바운드 작업에 비동기 함수를 사용하세요.
- 연결 풀링: 데이터베이스 연결 풀을 활용하세요.
- 캐싱: Redis 등을 활용한 응답 캐싱을 구현하세요.
- 페이지네이션: 대량의 데이터를 처리할 때 페이지네이션을 사용하세요.
- 적절한 worker 수: Gunicorn 설정에서 시스템 리소스에 맞게 worker 수를 조정하세요.
17. 비교: FastAPI vs. 다른 프레임워크#
FastAPI vs. Flask#
- FastAPI는 비동기를 기본 지원하지만, Flask는 기본적으로 동기식입니다.
- FastAPI는 자동 문서화와 타입 검증이 내장되어 있지만, Flask는 추가 라이브러리가 필요합니다.
- 성능 측면에서 FastAPI가 일반적으로 더 빠릅니다.
FastAPI vs. Django#
- Django는 완전한 풀스택 프레임워크이지만, FastAPI는 API 개발에 특화되어 있습니다.
- Django는 ORM, 관리자 패널 등 추가 기능을 기본 제공합니다.
- FastAPI는 경량화되어 있고 더 높은 성능을 제공합니다.
18. 결론#
FastAPI는 현대적인 파이썬 웹 애플리케이션 개발을 위한 강력한 도구입니다. 타입 힌트를 활용한 자동 문서화와 검증, 비동기 지원, 높은 성능을 제공하여 빠르고 안정적인 API를 개발할 수 있게 해줍니다. 쉬운 학습 곡선과 풍부한 기능을 통해 소규모 프로젝트부터 대규모 엔터프라이즈 애플리케이션까지 다양한 용도로 활용할 수 있습니다.
FastAPI를 학습하실 때는 공식 문서를 참고하시는 것이 가장 좋습니다. 또한 실제 프로젝트를 구현해보면서 경험을 쌓는 것이 중요합니다. 작은 API부터 시작하여 점차 복잡한 기능을 추가해 나가며 프레임워크의 다양한 기능을 익히시길 권장합니다.
FastAPI 종합 가이드#
1. FastAPI 개요#
FastAPI는 2018년 Sebastián Ramírez가 개발한 현대적이고 고성능의 파이썬 웹 프레임워크입니다. 이름에서 알 수 있듯이 빠른 API 개발에 중점을 둡니다.
주요 특징#
- 속도: 스타레트(Starlette)와 푸지(Pydantic)를 기반으로 하여 NodeJS 및 Go와 비슷한 수준의 높은 성능을 제공합니다.
- 타입 힌트: 파이썬의 타입 힌트를 활용하여 데이터 검증 및 자동 문서화를 지원합니다.
- 자동 문서화: Swagger UI와 ReDoc을 통해 API 문서를 자동으로 생성합니다.
- 비동기 지원: 비동기 프로그래밍을 완벽하게 지원합니다.
- 쉬운 학습 곡선: 직관적인 API 설계로 빠르게 학습하고 적용할 수 있습니다.
2. FastAPI 설치 및 기본 설정#
1
2
3
4
5
6
7
8
9
| # 가상환경 생성 (선택사항이지만 권장)
python -m venv fastapi-env
source fastapi-env/bin/activate # Windows: fastapi-env\Scripts\activate
# FastAPI 및 ASGI 서버 설치
pip install fastapi uvicorn
# 추가 기능을 위한 설치 (선택사항)
pip install "fastapi[all]" # 모든 선택적 의존성 설치
|
첫 번째 FastAPI 애플리케이션#
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
| # main.py
from fastapi import FastAPI
# FastAPI 인스턴스 생성
app = FastAPI()
# 루트 경로에 대한 GET 요청 핸들러 정의
@app.get("/")
def read_root():
return {"Hello": "World"}
# 경로 매개변수를 포함한 엔드포인트 정의
@app.get("/items/{item_id}")
def read_item(item_id: int, q: str = None):
return {"item_id": item_id, "q": q}
|
서버 실행#
1
2
| # 개발 서버 실행
uvicorn main:app --reload
|
이제 http://localhost:8000에서 API에 접근할 수 있으며, http://localhost:8000/docs에서 자동 생성된 문서를 확인할 수 있습니다.
3. 경로 매개변수 및 쿼리 매개변수#
경로 매개변수#
URL의 일부로 전달되는 변수입니다.
1
2
3
| @app.get("/users/{user_id}")
def read_user(user_id: int):
return {"user_id": user_id}
|
쿼리 매개변수#
URL의? 뒤에 오는 키-값 쌍입니다.
1
2
3
4
| @app.get("/users/")
def read_users(skip: int = 0, limit: int = 10):
# skip과 limit은 쿼리 매개변수
return {"skip": skip, "limit": limit}
|
혼합 사용#
1
2
3
4
5
6
| @app.get("/users/{user_id}/items/")
def read_user_items(
user_id: int, skip: int = 0, limit: int = 10, q: str = None
):
# user_id는 경로 매개변수, 나머지는 쿼리 매개변수
return {"user_id": user_id, "skip": skip, "limit": limit, "q": q}
|
4. 요청 본문 및 모델#
Pydantic 모델을 사용하여 요청 본문을 정의하고 검증합니다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
| 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/")
def create_item(item: Item):
# item은 자동으로 검증된 Pydantic 모델 인스턴스
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
|
5. 응답 모델#
응답의 형태를 명시적으로 정의할 수 있습니다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
| from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
class Item(BaseModel):
name: str
description: str = None
price: float
tax: float = None
class ItemResponse(BaseModel):
name: str
price: float
price_with_tax: float = None
@app.post("/items/", response_model=ItemResponse)
def create_item(item: Item):
# 응답은 ItemResponse 모델에 맞게 필터링됨
return {
"name": item.name,
"price": item.price,
"price_with_tax": item.price + (item.tax or 0)
}
|
6. 의존성 주입 시스템#
FastAPI는 강력한 의존성 주입 시스템을 제공합니다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
| from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer
app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
def get_current_user(token: str = Depends(oauth2_scheme)):
# 토큰 검증 로직 (실제로는 JWT 검증 등이 필요)
if token != "valid_token":
raise HTTPException(status_code=401, detail="Invalid token")
return {"username": "john_doe"}
@app.get("/users/me")
def read_users_me(current_user: dict = Depends(get_current_user)):
return current_user
|
7. 예외 처리#
HTTP 예외를 명시적으로 발생시킬 수 있습니다.
1
2
3
4
5
6
7
8
9
10
11
| from fastapi import FastAPI, HTTPException
app = FastAPI()
items = {"foo": "The Foo Item"}
@app.get("/items/{item_id}")
def read_item(item_id: str):
if item_id not in items:
raise HTTPException(status_code=404, detail="Item not found")
return {"item": items[item_id]}
|
8. 미들웨어#
요청/응답 처리 과정에 로직을 추가할 수 있습니다.
1
2
3
4
5
6
7
8
9
10
11
12
| from fastapi import FastAPI
import time
app = FastAPI()
@app.middleware("http")
async def add_process_time_header(request, call_next):
start_time = time.time()
response = await call_next(request)
process_time = time.time() - start_time
response.headers["X-Process-Time"] = str(process_time)
return response
|
9. 비동기 프로그래밍#
FastAPI는 비동기 프로그래밍을 완벽하게 지원합니다.
1
2
3
4
5
6
7
8
9
10
| import asyncio
from fastapi import FastAPI
app = FastAPI()
@app.get("/async-operation/")
async def read_async():
# 비동기 작업 수행
await asyncio.sleep(1)
return {"message": "Async operation completed"}
|
10. 데이터베이스 통합#
SQLAlchemy, Tortoise ORM 등을 통해 데이터베이스와 통합할 수 있습니다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
| from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
# 데이터베이스 설정
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
# 모델 정의
class UserDB(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
email = Column(String, unique=True, index=True)
username = Column(String, index=True)
password = Column(String)
# 데이터베이스 테이블 생성
Base.metadata.create_all(bind=engine)
app = FastAPI()
# 의존성: 데이터베이스 세션
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
# 사용자 생성 엔드포인트
@app.post("/users/")
def create_user(
email: str, username: str, password: str, db: Session = Depends(get_db)
):
db_user = UserDB(email=email, username=username, password=password)
db.add(db_user)
db.commit()
db.refresh(db_user)
return {"id": db_user.id, "email": db_user.email, "username": db_user.username}
|
11. API 보안#
FastAPI는 다양한 보안 메커니즘을 지원합니다.
OAuth2 및 JWT 인증#
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
| from datetime import datetime, timedelta
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel
# 보안 설정
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, expires_delta: timedelta = None):
to_encode = data.copy()
expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
# 모델
class User(BaseModel):
username: str
email: str = None
full_name: str = None
disabled: bool = None
class UserInDB(User):
hashed_password: str
# 가상 사용자 데이터베이스
fake_users_db = {
"johndoe": {
"username": "johndoe",
"full_name": "John Doe",
"email": "johndoe@example.com",
"hashed_password": pwd_context.hash("secret"),
"disabled": False,
}
}
app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
# 사용자 인증
def authenticate_user(fake_db, username: str, password: str):
user = get_user(fake_db, username)
if not user:
return False
if not pwd_context.verify(password, user.hashed_password):
return False
return user
# 사용자 가져오기
def get_user(db, username: str):
if username in db:
user_dict = db[username]
return UserInDB(**user_dict)
return None
# 현재 사용자 가져오기
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"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
raise credentials_exception
except JWTError:
raise credentials_exception
user = get_user(fake_users_db, username=username)
if user is None:
raise credentials_exception
return user
# 토큰 엔드포인트
@app.post("/token")
async def login_for_access_token(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_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = create_access_token(
data={"sub": user.username}, expires_delta=access_token_expires
)
return {"access_token": access_token, "token_type": "bearer"}
# 보호된 엔드포인트
@app.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_user)):
return current_user
|
12. 테스팅#
FastAPI는 TestClient를 통해 간편한 테스트를 지원합니다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
| from fastapi.testclient import TestClient
from main import app # FastAPI 앱 임포트
client = TestClient(app)
def test_read_root():
response = client.get("/")
assert response.status_code == 200
assert response.json() == {"Hello": "World"}
def test_read_item():
response = client.get("/items/42?q=test")
assert response.status_code == 200
assert response.json() == {"item_id": 42, "q": "test"}
|
13. 배포#
FastAPI 애플리케이션은 다양한 방법으로 배포할 수 있습니다.
Docker 컨테이너 배포#
1
2
3
4
5
6
7
8
9
10
| FROM python:3.9
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
|
Gunicorn과 함께 사용#
1
2
3
4
| pip install gunicorn
# 실행
gunicorn -w 4 -k uvicorn.workers.UvicornWorker main:app
|
14. 고급 기능#
웹소켓 지원#
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
| from fastapi import FastAPI, WebSocket, WebSocketDisconnect
app = FastAPI()
class ConnectionManager:
def __init__(self):
self.active_connections = []
async def connect(self, websocket: WebSocket):
await websocket.accept()
self.active_connections.append(websocket)
def disconnect(self, websocket: WebSocket):
self.active_connections.remove(websocket)
async def broadcast(self, message: str):
for connection in self.active_connections:
await connection.send_text(message)
manager = ConnectionManager()
@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int):
await manager.connect(websocket)
try:
while True:
data = await websocket.receive_text()
await websocket.send_text(f"Message: {data}")
await manager.broadcast(f"Client #{client_id} says: {data}")
except WebSocketDisconnect:
manager.disconnect(websocket)
await manager.broadcast(f"Client #{client_id} left the chat")
|
백그라운드 태스크#
1
2
3
4
5
6
7
8
9
10
11
12
13
14
| from fastapi import FastAPI, BackgroundTasks
app = FastAPI()
def write_log(message: str):
with open("log.txt", "a") as f:
f.write(message + "\n")
@app.post("/send-notification/{email}")
async def send_notification(
email: str, background_tasks: BackgroundTasks
):
background_tasks.add_task(write_log, f"Notification sent to {email}")
return {"message": "Notification sent in the background"}
|
15. 실제 프로젝트 구조#
대규모 FastAPI 프로젝트의 권장 구조:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
| myapp/
│
├── app/
│ ├── __init__.py
│ ├── main.py # FastAPI 앱 인스턴스 및 기본 설정
│ ├── dependencies.py # 의존성 함수
│ ├── routers/ # API 라우터
│ │ ├── __init__.py
│ │ ├── users.py
│ │ ├── items.py
│ │ └── ...
│ ├── models/ # 데이터베이스 모델
│ │ ├── __init__.py
│ │ ├── user.py
│ │ ├── item.py
│ │ └── ...
│ ├── schemas/ # Pydantic 모델
│ │ ├── __init__.py
│ │ ├── user.py
│ │ ├── item.py
│ │ └── ...
│ ├── crud/ # CRUD 작업
│ │ ├── __init__.py
│ │ ├── user.py
│ │ ├── item.py
│ │ └── ...
│ ├── core/ # 핵심 설정
│ │ ├── __init__.py
│ │ ├── config.py # 환경 변수 및 설정
│ │ └── security.py # 보안 관련 유틸리티
│ └── utils/ # 유틸리티 함수
│ ├── __init__.py
│ └── ...
│
├── tests/ # 테스트
│ ├── __init__.py
│ ├── conftest.py
│ ├── test_users.py
│ └── ...
│
├── alembic/ # 데이터베이스 마이그레이션
│ ├── versions/
│ └── ...
│
├── logs/ # 로그 파일
├── .env # 환경 변수
├── .gitignore
├── requirements.txt
├── Dockerfile
└── docker-compose.yml
|
16. 성능 최적화 팁#
- 비동기 활용: I/O 바운드 작업에 비동기 함수를 사용하세요.
- 연결 풀링: 데이터베이스 연결 풀을 활용하세요.
- 캐싱: Redis 등을 활용한 응답 캐싱을 구현하세요.
- 페이지네이션: 대량의 데이터를 처리할 때 페이지네이션을 사용하세요.
- 적절한 worker 수: Gunicorn 설정에서 시스템 리소스에 맞게 worker 수를 조정하세요.
17. 비교: FastAPI vs. 다른 프레임워크#
FastAPI vs. Flask#
- FastAPI는 비동기를 기본 지원하지만, Flask는 기본적으로 동기식입니다.
- FastAPI는 자동 문서화와 타입 검증이 내장되어 있지만, Flask는 추가 라이브러리가 필요합니다.
- 성능 측면에서 FastAPI가 일반적으로 더 빠릅니다.
FastAPI vs. Django#
- Django는 완전한 풀스택 프레임워크이지만, FastAPI는 API 개발에 특화되어 있습니다.
- Django는 ORM, 관리자 패널 등 추가 기능을 기본 제공합니다.
- FastAPI는 경량화되어 있고 더 높은 성능을 제공합니다.
18. 결론#
FastAPI는 현대적인 파이썬 웹 애플리케이션 개발을 위한 강력한 도구입니다. 타입 힌트를 활용한 자동 문서화와 검증, 비동기 지원, 높은 성능을 제공하여 빠르고 안정적인 API를 개발할 수 있게 해줍니다. 쉬운 학습 곡선과 풍부한 기능을 통해 소규모 프로젝트부터 대규모 엔터프라이즈 애플리케이션까지 다양한 용도로 활용할 수 있습니다.
FastAPI를 학습하실 때는 공식 문서를 참고하시는 것이 가장 좋습니다. 또한 실제 프로젝트를 구현해보면서 경험을 쌓는 것이 중요합니다. 작은 API부터 시작하여 점차 복잡한 기능을 추가해 나가며 프레임워크의 다양한 기능을 익히시길 권장합니다.