FastAPI에 모듈화된 구조 적용을 통한 빠른 프로토타이핑

broccoli 25.02.06

3,932 3 1

facebook twitter kakao link

3 1

facebook twitter kakao link

DEVOTEE 요약: 본 블로그는 FastAPI와 SQLite를 사용해 사용자 정보 관리 기능을 구축하는 프로젝트 과정을 설명합니다. 프로젝트는 API 엔드포인트, 데이터베이스 연동, 모델 정의, 데이터 검증 및 CRUD 기능 등을 모듈화하여 코드의 유지 보수성과 확장성을 높였습니다. FastAPI의 의존성 주입과 SQLite, SQLAlchemy를 활용하여 빠른 프로토타입 개발을 가능하게 하고, RESTful API 설계 원칙과 예외 처리를 통해 사용자 경험을 개선하였습니다.

DEVOTEE 추천 블로그

시작하며 🚀

오늘은 FastAPI와 SQLite를 사용하여 간단한 사용자 정보 관리 기능을 구축하고, 프로젝트를 모범 사례에 맞게 그리고, Fast라는 말에 걸맞게 신속하게 모듈화하는 과정을 공유해보려 합니다.

이 글에서는 FastAPI의 주요 기능들을 활용해 어떻게 프로젝트 구조를 만들 수 있는지에 관한 간단한 예시를 소개합니다.

그럼 시작해 보겠습니다. 😊

프로젝트 구조 🗂️

이 프로젝트 구조는 API 엔드포인트(api/)와 데이터베이스 연동(db/, crud/)을 모듈화하여 관리합니다.

모델 정의는 models/에,
데이터 검증용 스키마는 schemas/에 분리하여 코드의 가독성과 유지 보수성을 높였습니다.
최상위 main.py는 FastAPI 애플리케이션을 초기화하고 라우터를 등록하는 역할을 합니다.

app/
├── api/
│   └── v1/
│       └── endpoints/
│           └── users.py
├── crud/
│   └── user.py
├── db/
│   └── session.py
├── models/
│   └── user.py
├── schemas/
│   └── user.py
└── main.py

각 모듈별 내용

이제, 각 모듈이 어떤 역할을 하는지 자세히 살펴보겠습니다.

1) 데이터베이스 설정 (db/session.py)

SQLite 데이터베이스를 설정하기 위해 session.py 파일을 작성했습니다.

# app/db/session.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

DATABASE_URL = "sqlite:///./users.db"

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

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

여기서는 get_db() 함수를 통해 **의존성 주입(Dependency Injection)**을 사용하여 데이터베이스 세션을 주입합니다.

2) 데이터베이스 모델 정의 (models/user.py)

사용자 테이블을 정의하기 위해 user.py 파일을 작성했습니다.

# app/models/user.py
from sqlalchemy import Column, Integer, String
from app.db.session import Base

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True)
    email = Column(String, unique=True, index=True)
    full_name = Column(String)

3) Pydantic 스키마(schemas/user.py)

입력과 출력을 검증하기 위해 Pydantic 모델을 사용했습니다.

# app/schemas/user.py
from pydantic import BaseModel, EmailStr

class UserBase(BaseModel):
    username: str
    email: EmailStr
    full_name: str

class UserCreate(UserBase):
    password: str

class User(UserBase):
    id: int

    class Config:
        orm_mode = True

4) CRUD 기능 구현 (crud/user.py)

데이터베이스와 상호작용하는 CRUD 함수들을 user.py에 정의했습니다.

# app/crud/user.py
from sqlalchemy.orm import Session
from app.models.user import User
from app.schemas.user import UserCreate

def get_user_by_username(db: Session, username: str):
    return db.query(User).filter(User.username == username).first()

def get_user_by_email(db: Session, email: str):
    return db.query(User).filter(User.email == email).first()

def get_users(db: Session, skip: int = 0, limit: int = 10):
    return db.query(User).offset(skip).limit(limit).all()

def create_user(db: Session, user: UserCreate):
    db_user = User(
        username=user.username, 
        email=user.email, 
        full_name=user.full_name
    )
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

5) API 엔드포인트 (api/v1/endpoints/users.py)

users.py에서는 사용자 관련 API를 정의했습니다.

# app/api/v1/endpoints/users.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from typing import List
from app.schemas.user import UserCreate, User
from app.crud.user import create_user, get_user_by_username, get_user_by_email, get_users
from app.db.session import get_db

router = APIRouter()

@router.post("/users/", response_model=User)
def create_user_endpoint(user: UserCreate, db: Session = Depends(get_db)):
    db_user = get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return create_user(db=db, user=user)

@router.get("/users/", response_model=List[User])
def read_users(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    return get_users(db, skip=skip, limit=limit)

@router.get("/users/{username}", response_model=User)
def read_user(username: str, db: Session = Depends(get_db)):
    user = get_user_by_username(db, username=username)
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return user

6) FastAPI 앱 설정 (main.py)

main.py 파일에서 FastAPI 애플리케이션을 생성하고 라우터를 등록합니다.

# main.py
from fastapi import FastAPI
from app.api.v1.endpoints import users
from app.db.session import engine, Base

# 데이터베이스 테이블 생성
Base.metadata.create_all(bind=engine)

app = FastAPI(title="User Management API", version="1.0.0")

# 라우터 등록
app.include_router(users.router, prefix="/api/v1", tags=["users"])

@app.get("/")
async def root():
    return {"message": "Welcome to the User Management API!"}

프로젝트 실행하기 🚀

1) 필요 패키지 설치

서버 기동에 필요한 패키지를 배포합니다.

poetry add "pydantic[email]"

poetry add sqlalchemy

poetry add fastapi uvicorn

2) 기동하기

이제 모든 준비가 끝났습니다! FastAPI 서버를 실행해서 API의 동작을 확인해 볼 수 있습니다.

uvicorn main:app --reload

3) 동작 확인하기

API 문서 확인: http://127.0.0.1:8000/docs
사용자 생성 테스트: POST /api/v1/users/

curl -X POST "http://127.0.0.1:8000/api/v1/users/" \
-H "Content-Type: application/json" \
-d '{
    "username": "testuser",
    "email": "testuser@example.com",
    "full_name": "Test User",
    "password": "testpassword123"
}'

사용자 생성 예외 처리 : Email already registered

curl -X POST "http://127.0.0.1:8000/api/v1/users/" \
-H "Content-Type: application/json" \
-d '{
    "username": "testuser",
    "email": "testuser@example.com",
    "full_name": "Test User",
    "password": "testpassword123"
}'

{"detail":"Email already registered"}

사용자 목록 조회 테스트: GET /api/v1/users/

curl -X GET "http://127.0.0.1:8000/api/v1/users/"
[{"username":"testuser","email":"testuser@example.com","full_name":"Test User","id":1}]

결론 💡

모듈화된 프로젝트 구조의 중요성 🚀
- api, db, models, schemas, crud로 기능을 분리하여 코드 가독성과 유지 보수성을 크게 향상할 수 있습니다.
  모듈화를 통해 재사용성과 확장성이 극대화되었으며, 추후 기능 추가나 데이터베이스 변경이 용이한 구조를 갖추었습니다.
FastAPI의 강력한 기능 활용 ⚡
- 의존성 주입(Dependency Injection)을 사용해 데이터베이스 세션을 안전하게 관리하고, Pydantic을 활용해 입력 데이터 검증을 강화할 수 있습니다.
SQLite와 SQLAlchemy를 통한 빠른 프로토타이핑 💡
- SQLite를 사용해 빠르게 데이터베이스 설정을 완료하고, SQLAlchemy로 ORM 방식을 적용해 데이터베이스 관련 처리를 직관적이고 효율적으로 구성할 수 있습니다.
API 설계와 예외 처리의 중요성 🛡️
- RESTful API 원칙을 준수하며, FastAPI의 HTTPException을 활용해 에러 처리를 구조화했습니다.
  이를 통해 명확한 에러 메시지 제공으로 사용자 경험을 개선할 수 있습니다.
자동화된 테스트 및 문서화 📑
- FastAPI의 자동 Swagger 문서 생성을 통해 API 테스트를 효율적으로 진행할 수 있었고, uvicorn을 사용해 빠르게 서버를 실행해 개발 속도를 높일 수 있습니다.