728x90
안녕하세요.
이전 글에서는 FastAPI를 활용해 API서버를 생성해보았습니다.
2025.05.01 - [Python] - [Python] FastAPI 프레임워크로 api서버 만들기
[Python] FastAPI 프레임워크로 api서버 만들기
1. FastAPI란?FastAPI는 Python으로 작성된 최신 웹 프레임워크입니다. 주로 API 서버를 만들 때 사용합니다. 이름에서 알 수 있듯이 빠르고, 사용하기 쉽습니다.특히 자동으로 문서를 생성해주기 때문
stickode.tistory.com
이 글에서는 FastAPI의 의존성 주입(Dependency Injection) 시스템에 대해 알아보고, 실제 프로젝트에서 코드를 효율적으로 구성하는 방법에 대해 살펴보겠습니다.
1. 의존성(Dependency)이란 ?
소프트웨어 개발에서 '의존성'이란 한 컴포넌트가 다른 컴포넌트의 기능을 사용하기 위해 그 컴포넌트에 의존하는 관계를 말합니다. 쉽게 말해, A 모듈이 B 모듈의 기능을 사용한다면 "A는 B에 의존한다"고 표현합니다.
예를 들어, 웹 애플리케이션에서 사용자 인증 로직이 데이터베이스 연결에 의존하는 경우를 생각해볼 수 있습니다.
def authenticate_user(username: str, password: str):
# 데이터베이스 연결 생성 (의존성)
# 이 함수는 데이터베이스 연결에 '의존'합니다
db_connection = create_database_connection()
# 데이터베이스에서 사용자 조회
# db_connection 객체의 메서드를 사용하므로 의존성이 생깁니다
user = db_connection.query_user(username)
# 비밀번호 검증
if user and verify_password(password, user.hashed_password):
return user
return None
위 코드에서 authenticate_user 함수는 create_database_connection 함수에 의존하고 있습니다. 이러한 의존성이 코드 내부에 직접 구현되어 있으면 다음과 같은 문제가 발생할 수 있습니다.
- 테스트 어려움: 실제 데이터베이스 없이 함수를 테스트하기 어렵습니다.
- 유연성 부족: 데이터베이스 연결 방식을 변경하려면 함수 내부를 수정해야 합니다.
- 코드 중복: 여러 함수에서 같은 의존성 코드가 반복됩니다.
2. 의존성 주입(Dependency Injection)
의존성 주입은 이러한 문제를 해결하기 위한 디자인 패턴으로, 컴포넌트가 자신의 의존성을 직접 생성하는 대신 외부에서 제공받는 방식입니다. 즉, 필요한 의존성을 함수나 클래스 외부에서 "주입"받는 것입니다.
위 예제를 의존성 주입 방식을 적용하면 다음과 같이 수정할 수 있습니다.
def authenticate_user(username: str, password: str, db_connection):
# 외부에서 주입받은 데이터베이스 연결 사용
# 함수 내부에서 직접 생성하지 않고, 파라미터로 받습니다
user = db_connection.query_user(username)
# 비밀번호 검증
if user and verify_password(password, user.hashed_password):
return user
return None
# 함수 호출 시 의존성 주입
# db_connection을 외부에서 생성하여 함수에 전달합니다
db = create_database_connection()
user = authenticate_user("john_doe", "secret_password", db)
의존성 주입의 주요 이점
- 테스트 용이성: 테스트 시 실제 구현체 대신 모의(mock) 객체를 주입할 수 있습니다.
- 유연성 향상: 구현체를 쉽게 교체할 수 있습니다.
- 관심사 분리: 컴포넌트는 자신의 핵심 기능에만 집중할 수 있습니다.
- 코드 재사용: 의존성을 여러 컴포넌트에서 공유할 수 있습니다.
3. FastAPI의 의존성 주입 시스템
FastAPI는 Python의 타입 힌트와 결합된 강력한 의존성 주입 시스템을 제공합니다. 이 시스템의 핵심 특징은 다음과 같습니다.
- 자동 해결: FastAPI는 엔드포인트 함수에 필요한 의존성을 자동으로 해결하고 주입합니다.
- 계층적 의존성: 의존성은 다른 의존성에 의존할 수 있습니다(중첩 의존성).
- 타입 안전성: Python의 타입 힌트를 활용하여 의존성의 타입을 명확히 합니다.
- 성능 최적화: 동일한 요청 내에서 의존성을 캐싱하여 성능을 최적화합니다.
FastAPI에서 의존성 주입은 Depends 함수를 통해 이루어집니다.
from fastapi import Depends, FastAPI
app = FastAPI()
# 의존성 함수 정의
# 이 함수는 쿼리 파라미터를 처리하는 로직을 담당합니다
def get_query_params(q: str = None, skip: int = 0, limit: int = 100):
# 쿼리 파라미터를 딕셔너리로 반환
return {"q": q, "skip": skip, "limit": limit}
# 엔드포인트 함수에 의존성 주입
# Depends()를 사용하여 get_query_params 함수의 결과를 params로 받습니다
@app.get("/items/")
async def read_items(params: dict = Depends(get_query_params)):
# 주입받은 params 사용
return {"params": params}
위 예제에서 get_query_params 함수는 의존성 함수로, 쿼리 파라미터를 처리하는 로직을 담당합니다. read_items 엔드포인트 함수는 이 의존성을 주입받아 사용합니다.
4. FastAPI에서 의존성 주입 기본 사용법
FastAPI에서 의존성 주입의 기본 형태는 다음과 같습니다.
from fastapi import Depends, FastAPI
app = FastAPI()
# 공통 파라미터를 처리하는 의존성 함수
def common_parameters(q: str = None, skip: int = 0, limit: int = 100):
# 쿼리 파라미터를 딕셔너리로 반환
return {"q": q, "skip": skip, "limit": limit}
# 첫 번째 엔드포인트에 의존성 주입
@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
# 주입받은 commons 사용
return {"commons": commons}
# 두 번째 엔드포인트에도 동일한 의존성 주입
# 코드 중복 없이 동일한 로직 재사용
@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):
# 주입받은 commons 사용
return {"commons": commons}
위 예제에서 common_parameters 함수는 의존성으로 사용됩니다. Depends를 통해 이 함수의 반환값을 엔드포인트 함수에 주입받아 사용할 수 있습니다.
의존성 함수의 파라미터
의존성 함수는 일반 FastAPI 경로 함수와 마찬가지로 다양한 파라미터를 가질 수 있습니다.
이 예제에서는 HTTP 헤더에서 토큰과 키를 검증하는 두 개의 의존성 함수를 사용합니다.
from fastapi import Depends, FastAPI, Header, HTTPException
app = FastAPI()
# 토큰 검증 의존성 함수
def verify_token(x_token: str = Header(...)):
# Header(...)는 필수 헤더를 의미합니다
# x_token 헤더가 없으면 자동으로 오류 발생
# 토큰 검증 로직
if x_token != "fake-super-secret-token":
# 유효하지 않은 토큰이면 예외 발생
raise HTTPException(status_code=400, detail="X-Token header invalid")
# 유효한 토큰이면 반환
return x_token
# 키 검증 의존성 함수
def verify_key(x_key: str = Header(...)):
# 키 검증 로직
if x_key != "fake-super-secret-key":
# 유효하지 않은 키면 예외 발생
raise HTTPException(status_code=400, detail="X-Key header invalid")
# 유효한 키면 반환
return x_key
# 두 개의 의존성 함수를 모두 사용하는 엔드포인트
@app.get("/items/")
async def read_items(
# verify_token 함수의 결과를 token으로 받음
token: str = Depends(verify_token),
# verify_key 함수의 결과를 key로 받음
key: str = Depends(verify_key)
):
# 두 의존성이 모두 통과해야 이 코드가 실행됨
return {"token": token, "key": key}
5. 재사용 가능한 의존성 함수 작성하기
의존성 함수는 단순한 값 반환뿐만 아니라 복잡한 로직을 포함할 수 있습니다. 데이터베이스 연결, 설정 로드, 유효성 검증 등 다양한 작업을 처리하는 의존성 함수를 작성해 보겠습니다.
데이터베이스 세션 의존성
아래에서 get_db 함수는 데이터베이스 세션을 생성하고, 요청 처리가 끝나면 세션을 닫는 의존성 함수입니다. yield를 사용하여 컨텍스트 관리를 구현했습니다.
from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session
from . import crud, models, schemas
from .database import SessionLocal, engine
app = FastAPI()
# 데이터베이스 세션 의존성 함수
def get_db():
# 새 데이터베이스 세션 생성
db = SessionLocal()
try:
# yield를 사용하여 db 객체를 의존성으로 제공
# yield는 제너레이터를 만들어 컨텍스트 관리를 가능하게 함
yield db
finally:
# 요청 처리가 완료된 후 항상 세션 닫기
# 예외가 발생해도 finally 블록은 실행됨
db.close()
# 사용자 생성 엔드포인트
@app.post("/users/", response_model=schemas.User)
def create_user(
# 사용자 데이터는 요청 본문에서 받음
user: schemas.UserCreate,
# 데이터베이스 세션은 의존성으로 주입
db: Session = Depends(get_db)
):
# 이메일 중복 검사
db_user = crud.get_user_by_email(db, email=user.email)
if db_user:
raise HTTPException(status_code=400, detail="Email already registered")
# 사용자 생성 및 반환
return crud.create_user(db=db, user=user)
중첩 의존성
의존성은 다른 의존성에 의존할 수도 있습니다.
이 예제에서 query_or_default 의존성은 query_extractor 의존성에 의존합니다.
from fastapi import Depends, FastAPI
app = FastAPI()
# 첫 번째 의존성 함수: 쿼리 파라미터 추출
def query_extractor(q: str = None):
# q 파라미터가 있으면 그 값을, 없으면 None을 반환
return q
# 두 번째 의존성 함수: 첫 번째 의존성에 의존
def query_or_default(
# query_extractor 함수의 결과를 q로 받음
q: str = Depends(query_extractor)
):
# q가 있으면 그 값을, 없으면 "default"를 반환
if q:
return q
return "default"
# 엔드포인트 함수
@app.get("/items/")
async def read_items(
# query_or_default 함수의 결과를 query로 받음
# 이 함수는 내부적으로 query_extractor에 의존함
query: str = Depends(query_or_default)
):
# 최종 쿼리 값 반환
return {"q": query}
클래스 기반 의존성
의존성은 함수뿐만 아니라 클래스로도 구현할 수 있습니다.
from fastapi import Depends, FastAPI
from typing import Optional
app = FastAPI()
# 공통 쿼리 파라미터를 위한 클래스
class CommonQueryParams:
def __init__(self, q: Optional[str] = None, skip: int = 0, limit: int = 100):
# 초기화 메서드에서 쿼리 파라미터 저장
self.q = q
self.skip = skip
self.limit = limit
# 엔드포인트 함수
@app.get("/items/")
async def read_items(
# CommonQueryParams 클래스의 인스턴스를 의존성으로 주입
# 요청의 쿼리 파라미터를 사용하여 자동으로 인스턴스 생성
commons: CommonQueryParams = Depends(CommonQueryParams)
):
# 클래스 인스턴스의 속성에 접근
return {"q": commons.q, "skip": commons.skip, "limit": commons.limit}
클래스를 직접 의존성으로 사용할 수도 있습니다.
@app.get("/items/")
# Depends()에 인자 없이 사용하면 타입 힌트의 클래스가 의존성으로 사용됨
async def read_items(commons: CommonQueryParams = Depends()):
# 클래스 인스턴스의 속성에 접근
return {"q": commons.q, "skip": commons.skip, "limit": commons.limit}
6. 의존성 주입을 활용한 인증/인가 시스템 구현
FastAPI에서 의존성 주입은 인증 및 인가 시스템을 구현하는 데 매우 유용합니다. JWT 토큰 기반 인증 시스템을 구현해 보겠습니다.
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from datetime import datetime, timedelta
from typing import Optional
from . import schemas, crud
from .database import get_db
# 비밀 키 및 알고리즘 설정
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
app = FastAPI()
# OAuth2PasswordBearer는 토큰을 추출하는 의존성
# tokenUrl은 클라이언트가 토큰을 얻기 위해 요청할 URL
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
# JWT 액세스 토큰 생성 함수
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
# 인코딩할 데이터 복사
to_encode = data.copy()
# 만료 시간 설정
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=15)
# 만료 시간 추가
to_encode.update({"exp": expire})
# JWT 인코딩
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
# 현재 사용자 가져오는 의존성 함수
async def get_current_user(
# oauth2_scheme 의존성으로 토큰 추출
token: str = Depends(oauth2_scheme),
# get_db 의존성으로 데이터베이스 세션 가져오기
db = Depends(get_db)
):
# 인증 실패 시 발생시킬 예외 정의
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
# JWT 토큰 디코딩
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
# 사용자명 추출
username: str = payload.get("sub")
if username is None:
raise credentials_exception
# 토큰 데이터 검증
token_data = schemas.TokenData(username=username)
except JWTError:
# JWT 관련 오류 발생 시 인증 실패
raise credentials_exception
# 데이터베이스에서 사용자 조회
user = crud.get_user(db, username=token_data.username)
if user is None:
# 사용자가 없으면 인증 실패
raise credentials_exception
return user
# 활성 사용자 확인 의존성 함수
async def get_current_active_user(
# get_current_user 의존성으로 현재 사용자 가져오기
current_user = Depends(get_current_user)
):
# 사용자가 비활성 상태인지 확인
if not current_user.is_active:
raise HTTPException(status_code=400, detail="Inactive user")
return current_user
# 사용자 정보 조회 엔드포인트
@app.get("/users/me/", response_model=schemas.User)
async def read_users_me(
# get_current_active_user 의존성으로 활성 사용자 확인
current_user = Depends(get_current_active_user)
):
# 인증된 현재 사용자 정보 반환
return current_user
위 예제에서는 여러 의존성 함수를 체인으로 연결하여 인증 시스템을 구현했습니다.
- oauth2_scheme: 요청에서 JWT 토큰을 추출
- get_current_user: 토큰을 검증하고 사용자 정보를 반환
- get_current_active_user: 활성 사용자인지 확인
이 구조의 장점은 인증 로직을 한 곳에 모아 여러 엔드포인트에서 재사용할 수 있다는 점입니다. 또한 의존성 체인을 통해 인증 과정을 단계적으로 구성할 수 있습니다.
7. 실전 예제: 게시글 CRUD API 시스템 구축하기
이제 지금까지 배운 내용을 종합하여 간단한 게시글 CRUD API 시스템을 구축해 보겠습니다.
from fastapi import Depends, FastAPI, HTTPException, status
from sqlalchemy.orm import Session
from typing import List
from . import crud, models, schemas
from .database import get_db
from .auth import get_current_active_user
app = FastAPI()
# 게시글 권한 확인 의존성 함수
async def check_post_permission(
# 경로 파라미터에서 게시글 ID 받기
post_id: int,
# 현재 인증된 사용자 가져오기
current_user = Depends(get_current_active_user),
# 데이터베이스 세션 가져오기
db: Session = Depends(get_db)
):
# 게시글 조회
post = crud.get_post(db, post_id=post_id)
# 게시글이 없으면 404 오류
if not post:
raise HTTPException(status_code=404, detail="Post not found")
# 작성자나 관리자만 접근 가능
if post.author_id != current_user.id and not current_user.is_admin:
raise HTTPException(status_code=403, detail="Permission denied")
# 권한 확인 후 게시글 반환
return post
# 게시글 목록 조회 엔드포인트
@app.get("/posts/", response_model=List[schemas.Post])
def read_posts(
# 페이지네이션 파라미터
skip: int = 0,
limit: int = 100,
# 데이터베이스 세션
db: Session = Depends(get_db)
):
# 게시글 목록 조회 및 반환
posts = crud.get_posts(db, skip=skip, limit=limit)
return posts
# 게시글 상세 조회 엔드포인트
@app.get("/posts/{post_id}", response_model=schemas.Post)
def read_post(
# 경로 파라미터에서 게시글 ID 받기
post_id: int,
# 데이터베이스 세션
db: Session = Depends(get_db)
):
# 게시글 조회
post = crud.get_post(db, post_id=post_id)
# 게시글이 없으면 404 오류
if post is None:
raise HTTPException(status_code=404, detail="Post not found")
return post
# 게시글 작성 엔드포인트
@app.post("/posts/", response_model=schemas.Post)
def create_post(
# 요청 본문에서 게시글 데이터 받기
post: schemas.PostCreate,
# 인증된 사용자만 게시글 작성 가능
current_user = Depends(get_current_active_user),
# 데이터베이스 세션
db: Session = Depends(get_db)
):
# 게시글 생성 및 반환
# 현재 사용자를 작성자로 설정
return crud.create_post(db=db, post=post, user_id=current_user.id)
# 게시글 수정 엔드포인트
@app.put("/posts/{post_id}", response_model=schemas.Post)
def update_post(
# 경로 파라미터에서 게시글 ID 받기
post_id: int,
# 요청 본문에서 수정할 데이터 받기
post_update: schemas.PostUpdate,
# 권한 확인 의존성 (작성자나 관리자만 수정 가능)
post = Depends(check_post_permission),
# 데이터베이스 세션
db: Session = Depends(get_db)
):
# 게시글 수정 및 반환
return crud.update_post(db=db, post=post, post_update=post_update)
# 게시글 삭제 엔드포인트
@app.delete("/posts/{post_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_post(
# 경로 파라미터에서 게시글 ID 받기
post_id: int,
# 권한 확인 의존성 (작성자나 관리자만 삭제 가능)
post = Depends(check_post_permission),
# 데이터베이스 세션
db: Session = Depends(get_db)
):
# 게시글 삭제
crud.delete_post(db=db, post_id=post_id)
# 204 No Content 응답
return None
이 예제에서는 다음과 같은 의존성을 활용했습니다.
- get_db: 데이터베이스 세션 제공
- get_current_active_user: 인증된 활성 사용자 정보 제공
- check_post_permission: 게시글에 대한 권한 확인
이러한 의존성 주입 패턴을 통해 코드의 중복을 줄이고, 관심사를 분리하며, 테스트하기 쉬운 구조를 만들 수 있습니다.
'Python' 카테고리의 다른 글
| [Python] FastAPI 비동기 처리(1) 비동기 엔드포인트 만들기 (0) | 2025.05.15 |
|---|---|
| [Python] 텍스트 데이터 전처리- 불용어 제거 (0) | 2025.05.07 |
| [Python] FastAPI 프레임워크로 api서버 만들기 (0) | 2025.05.01 |
| [Python] 파일 자동 정리 프로그램 만들기 (0) | 2025.02.21 |
| [Python] smtplib로 이메일 발송하기 (0) | 2025.02.18 |
