FastAPI 서버개발, SQLAlchemy와 FastAPI를 이용한 데이터베이스 연동

소개

FastAPI는 Python기반의 현대적인 웹 프레임워크로, 비동기 프로그래밍을 지원하고 자동 생성된 API 문서와 유효성 검사를 제공합니다.
본 글에서는 FastAPI와 SQLAlchemy를 사용하여 데이터베이스와 연동하는 방법에 대해 자세히 알아보겠습니다.

FastAPI와 SQLAlchemy 개요

SQLAlchemy는 Python용 SQL 툴킷이자 Object Relational Mapper(ORM)이다.
ORM은 데이터를 클래스로 표현하고, 데이터베이스의 테이블을 객체로 매핑하여 SQL 쿼리를 객체지향적으로 처리할 수 있게 해준다.
FastAPI는 이러한 SQLAlchemy와 잘 통합되어 강력한 RESTful API를 쉽게 구축할 수 있도록 지원한다.

필요한 패키지 설치하기

pip install fastapi uvicorn sqlalchemy databases

프로젝트 구조

이번 예제에서는 다음과 같은 파일 구조를 사용할 것입니다:


    project/
    ├── main.py
    ├── models.py
    ├── database.py
    └── schemas.py

1. 데이터베이스 설정

이제 데이터베이스에 연결하기 위한 설정을 해보겠습니다.

database.py


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

    # 데이터베이스 URL 설정
    DATABASE_URL = "sqlite:///./test.db"  # SQLite 데이터베이스 사용

    # 데이터베이스 엔진을 생성
    engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})

    # 세션을 설정
    SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

    # 기본 클래스 생성
    Base = declarative_base()

    # 데이터베이스 메타데이터
    metadata = MetaData()

2. 모델 정의하기

이제 SQLAlchemy 모델을 정의할 차례입니다. 모델은 데이터베이스의 테이블 구조를 정의한 것입니다.

models.py


    from sqlalchemy import Column, Integer, String
    from database import Base

    class User(Base):
        __tablename__ = "users"  # 테이블 이름

        id = Column(Integer, primary_key=True, index=True)  # ID 필드
        name = Column(String, index=True)  # Name 필드
        email = Column(String, unique=True, index=True)  # Email 필드

3. Pydantic 스키마 생성하기

FastAPI는 데이터 유효성 검사 및 직렬화를 위해 Pydantic을 사용합니다. 우리는 요청 및 응답에 대한 스키마를 생성해야 합니다.

schemas.py


    from pydantic import BaseModel
    from typing import Optional

    class UserBase(BaseModel):
        name: str
        email: str

    class UserCreate(UserBase):
        pass

    class User(UserBase):
        id: int

        class Config:
            orm_mode = True  # ORM 모드를 활성화

4. FastAPI 애플리케이션 생성하기

이제 FastAPI 애플리케이션을 설정할 시간입니다. 우리는 CRUD(Create, Read, Update, Delete) 작업을 구현할 것입니다.

main.py


    from fastapi import FastAPI, Depends, HTTPException
    from sqlalchemy.orm import Session
    from database import SessionLocal, engine
    from models import User, Base
    from schemas import User, UserCreate

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

    app = FastAPI()

    # 데이터베이스 세션을 가져오는 의존성
    def get_db():
        db = SessionLocal()
        try:
            yield db
        finally:
            db.close()

    @app.post("/users/", response_model=User)
    def create_user(user: UserCreate, db: Session = Depends(get_db)):
        db_user = User(name=user.name, email=user.email)
        db.add(db_user)
        db.commit()
        db.refresh(db_user)
        return db_user

    @app.get("/users/{user_id}", response_model=User)
    def read_user(user_id: int, db: Session = Depends(get_db)):
        user = db.query(User).filter(User.id == user_id).first()
        if user is None:
            raise HTTPException(status_code=404, detail="User not found")
        return user

    @app.get("/users/", response_model=list[User])
    def read_users(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
        users = db.query(User).offset(skip).limit(limit).all()
        return users

    @app.put("/users/{user_id}", response_model=User)
    def update_user(user_id: int, user: UserCreate, db: Session = Depends(get_db)):
        db_user = db.query(User).filter(User.id == user_id).first()
        if db_user is None:
            raise HTTPException(status_code=404, detail="User not found")
        db_user.name = user.name
        db_user.email = user.email
        db.commit()
        db.refresh(db_user)
        return db_user

    @app.delete("/users/{user_id}")
    def delete_user(user_id: int, db: Session = Depends(get_db)):
        db_user = db.query(User).filter(User.id == user_id).first()
        if db_user is None:
            raise HTTPException(status_code=404, detail="User not found")
        db.delete(db_user)
        db.commit()
        return {"message": "User deleted"}

5. 애플리케이션 실행하기

이제 모든 설정이 완료되었습니다. 아래 명령어로 FastAPI 애플리케이션을 실행할 수 있습니다.

uvicorn main:app --reload

6. API 테스트하기

API가 성공적으로 실행되면, 브라우저에서 http://127.0.0.1:8000/docs로 이동하여 Swagger UI에서 API를 테스트할 수 있습니다.

예제 요청


    POST /users/
    {
        "name": "Alice",
        "email": "alice@example.com"
    }

성공적으로 생성되면 새 사용자 ID가 포함된 응답을 받을 수 있습니다:


    {
        "id": 1,
        "name": "Alice",
        "email": "alice@example.com"
    }

결론

이번 글에서는 FastAPI와 SQLAlchemy를 이용하여 간단한 사용자 관리 API를 구축하는 방법을 알아보았습니다.
FastAPI는 직관적인 문법과 비동기 처리 덕분에 신속한 개발이 가능하며, SQLAlchemy와의 통합 덕분에 강력한 데이터베이스 작업을 손쉽게 수행할 수 있습니다.
이를 통해 여러분도 멋진 백엔드 애플리케이션을 개발할 수 있게 되기를 바랍니다.

더 알아보세요

FastAPI와 SQLAlchemy에 대해 더 깊이 있는 정보가 필요하다면 그들의 공식 문서를 참고하시길 권장합니다.

소개