FastAPI에 SQLAlchemy 연동하기

2022.09.16 - [개발이야기/Python] - 정말 빠른 Fast API

2023.01.18 - [개발이야기/Python] - FastAPI로 CRUD 구현하기

지난 2개의 포스팅에서 FastAPI를 사용해 CRUD를 구현하는 방법에 대해 알아 보았다.

 

이젠 실제 Database를 사용하여 Data를 다뤄보도록 하겠다.

FastAPI에서 Database를 사용하려면 ORM이 필요하다. 여기서는 SQLAlchemy를 사용하겠다.

먼저 SQLAlchemy를 설치 하자.

pip install sqlalchemy

db.py

Database와 연동하기 위한 파일을 작성해보겠다.
main.py 와 같은 위치에 db.py 파일을 만들고 아래와 같이 작성해보자.

#db.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# SQLite를 DB로 사용
SQLALCHEMY_DATABASE_URL = "sqlite:///./blog_app.db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
# connect_args={"check_same_thread": False} 부분은 SQLite에서만 사용
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Database 객체
Base = declarative_base()

models.py

그리고 main.py에 작성했단 Post model과 동일하게 model calss를 만들어 보자.

마찬 가지로 main.py와 동일한 위치에 models.py 파일을 만들자.
아래와 같이 PostTable class를 만들고 앞서 만든 DataBase 객체를 상속 받도록 작성했다.
이렇게 되면 Database 에 posts라는 Table이 만들어지게 된다.
기존에 없었던 id 필드를 추가하고 해당 필드를 Primary key로 설정한다.

# models.py
from sqlalchemy import Column, Integer, String
from db import Base # 선언된 DB import

class PostTable(Base):
    __tablename__ = "posts" # 실제 DB에 생성될 Table 이름

    id = Column(Integer, primary_key=True, index=True) # PK
    title = Column(String)
    author = Column(String)
    contents = Column(String)

main.py에 적용

이제 db.py와 models.py를 main.py에 적용해보자.

from typing import List # Optional 삭제, List 추가
from fastapi import FastAPI, Depends # Depends 추가
from pydantic import BaseModel

from sqlalchemy.orm import Session
from db import SessionLocal, engine # 연동된 DB의 Sesssion과 DB Engine Import
from models import Base # 선언된 DB import 
from models import PostTable # DB Table import

Base.metadata.create_all(bind=engine) # DB 바인딩

app = FastAPI()

class Post(BaseModel):
    title: str
    author: str
    contents: str

    class Config: # ORM 설정 추가
        orm_mode = True


def get_db():
    # DB 핸들러
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

...

Main.py 윗 부분의 코드 이다.
필요한 모듈들을 import 하고 db.py, models.py의 내용들을 import 했다.
DB를 바인딩한 다음에 get_db 라는 DB 헨들러를 추가했다.

이제 각 함수에서 database 를 사용할 수 있도록 만들어 보자.

1. get_posts
   get_posts는 전체 Post list 를 조회하는 함수이다.위에서 response_model=List[Post] 부분은 응답할 데이터의 타입을 지정하는 부분이다.
   List객체 안에 Post객체가 담겨있는 응답이 발생한다.

@app.get('/posts/', response_model=List[Post]) 
def get_posts(db: Session = Depends(get_db)): 
# db인자에 Default 값으로 DB와 연동된 ORM Session 객체를 전달
    return db.query(PostTable).all() 
    # 전달 받은 ORM Session을 사용해 PostTable의 모든 Row를 Select

2. get_post
   get_post는 post_id를 받아 해당 post의 내용을 반환하는 함수이다.마찬가지로 response_model 옵션을 이용해 응답 타입을 지정했다.

@app.get('/posts/{post_id}', response_model=Post)
def get_post(post_id: int, db: Session = Depends(get_db)): 
    return db.query(PostTable).filter(PostTable.id == post_id).first()
    # PostTable의 id와 일치하는 Post를 반환

3. create_post
   이번엔 새로운 Post 를 생성하는 create_post 함수 이다.
   create_post 함수는 추가될 Post 객체를 인자로 받는다.
   그리고, PostTable class를 사용해 새로운 Post(new_post)객체를 생성하고 해당 객체를 Database에 추가한다.

@app.post('/posts/', response_model=Post)
def create_post(post: Post, db: Session = Depends(get_db)):
    new_post = PostTable(
        title=post.title,
        author=post.author,
        contents=post.contents
        )
    db.add(new_post)
    db.commit()
    db.refresh(new_post)
    return new_post # 추가된 Post 내용을 반환

4. update_post
   기존 post를 업데이트 하는 update_post를 수정해보자.
   update_post에서는 업데이트할 Post 객체와 수정 대상의 id를 인자로 받는다.

@app.put('/posts/{post_id}', response_model=Post)
def update_post(post_id: int, post: Post, db: Session = Depends(get_db)):
    target_post = db.query(PostTable).filter(PostTable.id == post_id)
    target_post.update(post) # 전달받은 Post객체의 내용으로 업데이트
    db.commit()
    return target_post.first() # 업데이트 완료된 Post내용을 반환

5. delete_post
   마지막으로 작성된 Post를 삭제하는 delete_post 함수이다.
   삭제 대상의 id를 인자로 받아 db에서 삭제하도록 작성했다.

@app.delete('/posts/{post_id}', response_model=Post)
def delete_post(post_id: int, db: Session = Depends(get_db)):
    target_post = db.query(PostTable).filter(PostTable.id == post_id)\
        .first()
    db.delete(target_post) # 조회된 Post 삭제
    db.commit()
    return target_post

이처럼 각 함수에서는 db를 default인자로 전달 받아 all, filter, add, update, commit, delete 등의 SQLAlchemy 메소드들을 사용할 수 있다.

마무리

지금까지 FastAPI와 SQLAlchemy를 이용해 간단한 Blog의 Back-end를 구성해봤다.
FastAPI는 Python의 type hinting을 사용해 입력,출력 데이터들의 Validation을 해주 부분과 Swagger document를 자동으로 생성해 주는 부분이 너무 좋은거같다.
그리고 아주 빠르고 쉽게 API를 구성할 수 있다라는게 큰 장점이라고 생각한다.