FastAPI 공식문서의 SQL Database 부터 Testing 까지의 내용을 요약하고, 부족한 설명을 보충하여 정리한 문서입니다.

SQL Databases

ORM

• Object Relational Mapping 의 약자로, 데이터베이스의 내용(table, row)등을 프로그래밍 언어의 객체로 취급하여 다루는 개념
• 예를들어 SQL의 테이블 pets를 파이썬 클래스 Pet으로 맵핑하여 아래와 같은 동작을 수행할 수 있음
  • pet = Pet() 이라고 가정하고,
  • pet.type -> SQL pets 테이블의 type 컬럼에 접근 가능
  • pet.owner -> SQL pets 테이블의 owner 컬럼에 접근 가능
• Python의 대표적인 ORM 라이브러리는 Django-ORM, SQLAlchemy, TortoiseORM, Peewee 가 있음
  • Django-ORM : 장고 프레임워크에 내장된 ORM으로 FastAPI 개발시 사용할 일이 없음
  • SQLAlchemy : 파이썬의 가장 표준적인 ORM
  • Peewee : 가볍고 간단한 기능을 제공하는 ORM으로 sqlite / mysql / postgresql / cockroachdb 만 지원함
  • Tortoise ORM : 등장한지 얼마 안된 ORM으로 비동기 처리에 적합
    • SQLAlchemy가 아직 비동기 처리에 있어서는 알파단계이며, 따라서 제한적인 기능으로 제공하고 있기 때문에 DB와 비동기 통신을 위해서는 Tortoise ORM 사용을 추천
• SQLAlchemy 튜토리얼 한국어 번역 문서 (비공식) : 링크

SQLAlchemy 를 활용한 FastAPI에서의 DB 통신

DB 통신 설정 - create_engine

from sqlalchemy import create_engine

SQLALCHEMY_DATABASE_URL = "mysql://user:password@postgresserver/db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, pool_pre_ping=True, pool_size=15, max_overflow=0, encoding='utf8', convert_unicode=True
)

• 우선 DB와 연결을 어떻게 할지에 대한 설정이 필요하다. create_enigne 객체에 각종 설정 값을 인자로 넣어 초기화하여 이를 수행한다.
• 위처럼 DB와 통신할 때 각종 설정들을 추가할 수 있다.
• SQLite 를 사용한다면 connect_args={"check_same_thread": False} 옵션을 켜줘야함
  • FastAPI는 기본 함수 호출시 여러 쓰레드에서 이를 돌게하는데, SQLite는 한번의 DB 통신에 하나의 쓰레드만 사용하도록 해놓음
  • 이 때문에, SQLite에게 지금 한개의 쓰레드에만 연결되어있다고 알려줘야 함 -> 이 때 사용하는 인자
• 설정값 참고

DB 세션 설정 - sessionmaker

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "mysql://user:password@postgresserver/db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, pool_pre_ping=True, pool_size=15, max_overflow=0, encoding='utf8', convert_unicode=True
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

• create_eninge을 초기화하여 DB 연결에 대한 설정을 마쳤으면, 실질적인 통신을 할 세션에 대한 정의를 해야한다. sessionmaker라는 객체를 초기화하여 이를 수행한다.

• SessionLocal로 이름을 지정한 이유는 추후에 사용할 Session과 분리하기 위함

• 여기서 선언한 SessionLocal은 그 자체가 데이터베이스 세션이 아니다.

  이는 설정값을 넘겨받아 선언된 새로운 클래스일 뿐이다.

  이렇게 설정값을 넘겨받아 새로 선언된 SessionLocal클래스의 인스턴스 객체가 생성되면 그 인스턴스 객체가 실질적인 데이터베이스 세션 역할을 수행한다.
  (* sessionmaker의 docstring을 읽어보면 자세히 이해가능)
  

모델 관리를 위한 Base 클래스 선언 및 데이터 모델링

from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

• ORM 모델에 대해 설정할 때 사용할 Base 클래스를 선언해준다.

from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)

    items = relationship("Item", back_populates="owner")


class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    description = Column(String, index=True)
    owner_id = Column(Integer, ForeignKey("users.id"))

    owner = relationship("User", back_populates="items")

• 데이터 모델을 지정해준다. 이때의 모델 클래스들은 위에서 선언해둔 Base를 상속받는다.
• 따라서 사전에 Base Class 에 대해 설정을 해두면 여러가지 반복된 코드를 줄이거나, 모델 클래스 정의의 일부분을 자동화 할 수 있다.

import re
from datetime import datetime

from sqlalchemy import Column, DateTime
from sqlalchemy.ext.declarative import as_declarative, declared_attr
from sqlalchemy.ext.declarative import declarative_base


@as_declarative()
class Base:
    created_at = Column(DateTime, default=datetime.now)
    updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now)
    __name__: str

    # CamelCase의 클래스 이름으로부터 snake_case의 테이블 네임 자동 생성
    @declared_attr
    def __tablename__(cls) -> str:
        return re.sub(r'(?<!^)(?=[A-Z])', '_', cls.__name__).lower()

• 위처럼 Base 클래스를 별도로 오버라이딩 할 수 있다.
  • 위의 예시의 경우 모든 ORM 모델에 created_at 과 updated_at 이라는 컬럼을 자동으로 생성해주고,
  • 클래스의 이름으로 부터 테이블 이름을 자동으로 생성해주는 기능을 담고있다.

Pydantic 클래스를 추가하여 데이터 다루기

• ORM 으로 다루는 데이터베이스의 구성 요소들을 쉽게 읽고/반환 하기 위해 Pydantic 모델과 추가로 맵핑할 수 있다. (FastAPI 쓰는데 이거 안하면 손해)

Pydantic 클래스의 orm_mode 설정

• Pydantic 모델에 Config 서브 클래스를 선언하여 각종 설정값을 수정할 수 있다.
• ORM으로 다뤄지는 데이터베이스의 모델들은 ORM의 객체이지, dict와 같은 파이썬의 내장 자료형이 아니다.
• 이러한 ORM 객체에 접근하여 객체의 속성을 사용하거나, ORM 객체를 json 타입으로 API를 통해 반환해주기 위해서는 별도의 처리가 핋요하다.
• 이러한 역할을 해주는 것이 Pydantic 클래스의 orm_mode = True 설정이다.

즉! ORM 객체를 읽거나, 변수로 할당하여 처리하거나, json 반환을 위해 ORM 객체에 접근이 필요할 때, orm_mode 설정을 True로 줘야하는 것이다.

• 이렇게 할경우 python의 클래스 속성에 대한 getter 메서드 처럼 .을 찍어 객체의 속성에 접근할 수 있다.
• 예를 들어 dict 자료형에서 id = data.get('id') 이렇게 접근해야 하는것을 단순히 data.id 이렇게 접근할 수 있는 것이다.

• SQLAlchemy는 lazy loading을 기본값으로 한다.
• 만약 데이터베이스 모델을 리턴하는 특정 엔드포인트 함수가 orm_mode = True로 설정해둔 Pydantic 모델을 response_model로 설정하지 않았을 경우, 관계 테이블의 데이터는 표시되지 않는다. (어찌보면 당연)

sessionmaker로 만들어둔 SessionLocal과 Session의 차이

• 위에서 sessionmaker를 이용하여 다음과 같이 SessionLocal 객체를 선언했고, SessionLocal은 callable한 객체로 호출될 때마다 새로운 세션 객체들을 생성해낸다.

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

• 즉 위의 get_db는 호출될 때마다 db라는 변수를 생성해내는데, 이 db는 SessionLocal클래스의 인스턴스인 것이다.
• 그리고 이렇게 생성된 db는 데이터베이스 세션으로 작동하며 한 번의 db 연결동안 존재하며 db 연결에 대한 모든 요청/응답이 종료되면 close()된다.

그렇다면 from sqlalchemy.orm import Session과 SessionLocal은 어떻게 다른가?

• 위에서 db 연결 설정 (create_engine)을 바인딩해 정의한 sessionmaker로 SessionLocal을 만들었다. \
  그리고 이 SessionLocal을 호출해 db connection 객체를 매 커넥션마다 만들어내는 get_db함수를 정의했다. \
  이 get_db함수는 db 커넥션을 필요로 하는 엔드포인트 함수에 dependency로 주입되어 db connection에 대한 request가 있을 때 호출되어 request가 종료되면 close 된다.
• 하지만 엔드포인트 함수에 db 연결에 대한 type hinting을 SessionLocal이 아닌 Session으로 하게 된다.\
  
• 이렇게 하는 이유는 IDE의 자동완성 기능을 위해서 이다. 해당 부분의 type hinting을 SessionLocal로 지정해두어도 엔드포인트는 문제없이 작동한다.\
  하지만 Session 으로 지정해둠으로써 Session에서 제공하는 메서드에 대해 자동완성 기능을 제공 받을 수 있다. \
  
• 이렇게 작동하는 이유는 sessionmaker로 초기화되는 SessionLocal은 create_engine을 바인딩해서 Session을 랩핑한 클래스인데, \
  어차피 이 SessionLocal이 생성하는 세션은 sqlalchemy의 Session 객체 이다. 다만 sessionmaker함수로 한번 감싸져있기 때문에 IDE가 이것이 Session임을 인식 못할 뿐이다.

async def와 SQLAlchemy

• 기본적으로 SQLAlchemy 는 비동기 처리를 지원하지 않는다. 따라서 세션에 대한 비동기 응답처리를 할 수 없다. \
  user = await db.query(User).first() <- 이렇게 쓸 수 없다는 뜻
• FastAPI 공식 문서에서는 encode/databases 패키지와 함께 SQLAlchemy Core를 활용하여 비동기 연결을 구현할 수 있다고 소개한다.
• 하지만 Tortoise ORM 을 쓰는게 더 좋아보임 (더 쉬울듯)
• async sql docs
• encode/databases
• tortois orm /w fastapi

Database Migration

• ORM 객체들을 재정의하고 이를 실제 DB에 반영하기 위해서 Alembic 을 사용할 수 있다.

파이썬 버전에 따른 DB Session 설정

• yield 키워드를 포함한 함수를 dependency 로 주입할 수 없는 파이썬 3.5 이하에서는 db connection을 dependency 주입이 아닌, 미들웨어 작성을 통해 해결할 수 있다.

from fastapi import Request, Response, FastAPI
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "mysql://user:password@postgresserver/db"

app = FastAPI()

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, pool_pre_ping=True, pool_size=15, max_overflow=0, encoding='utf8', convert_unicode=True
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

@app.middleware("http")
async def db_session_middleware(request: Request, call_next):
    response = Response("Internal server error", status_code=500)
    try:
        request.state.db = SessionLocal()
        response = await call_next(request)
    finally:
        request.state.db.close()
    return response


# Dependency
def get_db(request: Request):
    return request.state.db

• 그냥 파이썬 버전 업뎃해서 쓰는게 좋다. 이렇게 미들웨어로 작성할 경우
  1. 더 복잡하고 더 많은 코드가 필요하다
  2. db 세션이 호출되는 동안 기다려야 하기 때문에 더 느릴 수 있다.
  3. 미들웨어로 작성되었기 때문에 DB와 관계없는 엔드포인트에 접근할 때도 db 세션이 만들어지고 닫힌다.

---

DB 사용시 발생했던 이슈 정리

격리수준에 따른 문제

• 하나의 DB Session으로 2개의 커넥션을 담당하는 엔드포인트 함수가 있었음

• 프론트 페이지 구현 때문에 Create 함수의 반환값으로 성공/실패 여부가 아닌, Create 작업이 반영된 DB를 Read 하는 것으로 구성함

• Create와 Read 모두 하나의 Session으로 처리하는 꼴로 구성됨

• 개발 당시 원하던 형태는 Create의 결과가 Read에 반영되어 Read의 결과 값이 반환되도록 하는 거였지만, Create 하기 전의 결과만 계속 반환되는 문제 발생

• create 함수가 commit 되기 전에 get 함수가 실행되는 것인지(너무 빨라서) 중간에 wait을 주거나 하지 않으면 계속 오류 발생

• 찾아보니 격리 수준에 따라 create 함수가 db connection을 어느 수준까지 다뤘느냐에 따라 get 함수의 결과가 달라질 수 있다는 내용 발견

• create 함수가 커밋 되던 말던 수정사항이 생기면 무조건 읽어올 수 있도록 db의 격리수준을 READ UNCOMMITED 로 변경

• 문제 해결 되었지만 READ UNCOMMITED 격리수준은 매우 불안정한 격리 수준으로 가급적 사용해서는 안되는 것

• 더 좋은 해결책은 create 엔드포인트 함수는 create에 대해서만 책임지고, 반영된 수정사항이 필요했으면 프론트에서 리다이렉트 하는게 더 좋지 않았을까?!

• 격리 수준 관련 게시글 링크

---

Router 분리하기

• FastAPI의 APIRouter 클래스를 통해 복잡한 API 엔드포인트를 정리할 수 있다.
• 메인 앱을 app = FastAPI() 이런식으로 정의하는데, router = APIRouter() 이런식으로 똑같이 쓸 수 있는 미니 FastAPI 객체 인 것이다.
• 각각 분리된 파일에서 API 엔드포인트를 관리하고 이것을 한곳에 합쳐 서빙할 수 있다.
  
• 이렇게 하면 아래처럼 하위하위라우터 > 하위 라우터 > 메인 라우터 이렇게 나눌 수도 있다.
  • admin/monitoring
  • admin/log
  • v1/product
  • v1/user/login \

전역 Dependency 주입

• APIRouter() 클래스에 인자를 넣어 전역 dependency 를 주입할 수 있다.

from fastapi import APIRouter, Depends, HTTPException, Header

async def get_token_header(x_token: str = Header(...)):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")


router = APIRouter(
    prefix="/items",
    tags=["items"],
    dependencies=[Depends(get_token_header)],
    responses={404: {"description": "Not found"}},
)

@router.get("")
async def create_item(item_id: str):
    ...

@router.get("")
async def read_items():
    ...

@router.get("/{item_id}")
async def read_item(item_id: str):
    ...

• 이렇게 하면 각 엔드포인트 함수에 get_token_header라는 의존성을 주입하지 않더라도 전역적으로 의존성을 주입할 수 있다.

• 위의 예시의 경우 모든 엔드포인트에서 헤더의 토큰 여부를 체크하게 된다.

• 전역 Dependency 를 지정하였어도, 각 엔드포인트 함수별로의 Dependency도 추가로 지정할 수 있다.

  • 이때는 전역 Dependency 함수가 먼저 실행되고 그 후 엔드포인트에 별도로 지정된 Dependency 함수가 실행된다.

• ! 개별 함수에 Dependency를 지정하는 것처럼 tags와 response를 개별로 한번 더 지정할 수 있다.

  • 이때 tag는 중복으로 표시되며, response는 여러개가 표시된다. <- 요 두가지 모두 실제 API에 영향을 주는게 아니라 swagger에 영향을 주는 기능

---

백그라운드 작업 (스케쥴작업)

• FastAPI 는 내장 백그라운드 작업 모듈을 탑재하고있다.

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


def write_notification(email: str, message=""):
    with open("log.txt", mode="w") as email_file:
        content = f"notification for {email}: {message}"
        email_file.write(content)


@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_notification, email, message="some notification")
    return {"message": "Notification sent in the background"}

• 이렇게 사용할 수 있다.

의존성 함수에 백그라운드 사용하기

• Dependency에 이 백그라운드 작업을 넣어서 관리할 수도 있다.
• 아래의 코드는 쿼리파라미터에 -> 쿼리를 txt 파일로 기록하는 백그라운드 작업을 수행하도록 하는 의존성 함수를 주입한 예시이다.
• 이렇게 하면 모든 API 요청에 추가된 쿼리 파라미터가 txt 파일로 기록될 것이다.
• 이렇게 따로 의존성 전용 함수로 빼면 재사용성이 증가한다.

from typing import Optional

from fastapi import BackgroundTasks, Depends, FastAPI

app = FastAPI()


def write_log(message: str):
    with open("log.txt", mode="a") as log:
        log.write(message)


def get_query(background_tasks: BackgroundTasks, q: Optional[str] = None):
    if q:
        message = f"found query: {q}\n"
        background_tasks.add_task(write_log, message)
    return q


@app.post("/send-notification/{email}")
async def send_notification(
    email: str, background_tasks: BackgroundTasks, q: str = Depends(get_query)
):
    message = f"message to {email}\n"
    background_tasks.add_task(write_log, message)
    return {"message": "Message sent"}

기타 백그라운드 작업 관련

• 위의 예시나, 아래의 Background 클래스의 코드 예시에서 볼 수 있듯이 FastAPI 내장 백그라운드 작업 모듈은 굉장히 제한적인 기능만 제공한다.
  

• 대안으로 사용할 수 있는 2개 패키지가 있다.

APScheduler의 Background

• 쓰레드 기반
• 크론탭 작업 / 특정 시간 작업 등 유동적인 작업 스케쥴링 가능 ( <> FastAPI 내장 모듈은 즉시 실행만 가능)
• 작업 데이터를 메모리에 저장할수도있고, DB에 연동할 수도 있음

Celery

• 아예 별도의 포트를 차지하는 별도의 프로세스로 굴러감
• 대용량(?) 처리 작업에 좋음
• flower 같은 패키지를 추가로 이용하여 GUI 에서 스케쥴링 관리 가능
• redis / kafka / rabbitmq 와 같은 작업을 담을 큐를 제공해주는 서비스를 붙여서 써야함

비교

항목	FastAPI 내장 모듈	APScheduler	Celery
워커	쓰레드	쓰레드	프로세스
스케쥴링 가능	X	O	O
사용 난이도	쉬움	덜쉬움	꽤나 복잡
대용량 처리 능력	낮음	덜낮음	꽤나 높음
GUI 관리 툴	X	X	O

• 추천하는 celery 예제 파일 링크

---

정적 파일 서빙

• 에러 코드에 대한 정적 파일을 서빙할 수 있다.

from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles

app = FastAPI()

app.mount("/static", StaticFiles(directory="static"), name="static")

• 이렇게 하면 전체 API 서버 (app의로 정의한)에 모든 HTTP Status 코드에 따른 정적 파일을 리턴할 수 있다.
• /static 폴더에 404.html 405.html 파일들을 저장해두면 알아서 서빙한다.

---

테스팅

• FastAPI 의 내장 테스팅 모듈은 pytest를 랩핑한 것으로 client만 연결해주면 pytest 처럼 쓸 수 있다.

from .main import app

client = TestClient(app)


def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"msg": "Hello World"}

• 기본적인 사용법은 pytest와 동일하고, client 객체를 이용하여 get/post/put/delete 등의 메서드로 각각의 API 엔드포인트에 접근할 수 있게 된다.

테스팅 관련 몇가지 내용

• 테스트에 실제 DB를 연결해서 하는게 좋나 ? 아니면 그냥 목업 모델을 만들어서 쓰는게 좋나 ?
  • 읽은 책에서는 굳이 실제 DB 커넥션을 할 필요는 없다고 하였고, FastAPI 문서에서도 가짜 모델로 테스트 진행함
  • 근데 FastAPI 제공 템플릿 프로젝트는 실제 DB와 커넥션하여 가짜 유저/가짜 아이템을 DB에 입력하여 테스트 진행
• 기본적인 status code 정도만 확인하는 테스트도 괜찮을까 (통합테스트) ? -> 단위테스트 막상 접목시켜보려니 공수가 너무 많이 드는 듯 함

• 부하 테스트는 어떻게?

테스트 함수에 커스텀 인자 주기

• 로그인 테스트 등을 위해 특정 엔드포인트 함수에 인자로 토큰 같은걸 줘야할 때가 있다.
• 이때는 conftest.py 라는 파일을 정의해둠으로써 테스트 함수에 여러 인자를 넘겨줄 수 있다. conftest.py 파일은 pytest를 위한 비품(fixture) 집합소라고 생각 하면 된다.

# confest.py

import random
import pytest

@pytest.fixture(scope="module")
def random_endpoint() -> int:
    return random.randint(1,100)

• confest.py를 위와 같이 정의해두고

# test.py
from fastapi.testclient import TestClient

from app.main import app

client = TestClient(app)

# Check if endpoint is incorrect
def test_get_series_contents_bad_request(random_endpoint: int):
    response = client.get(f"/v1/series/{random_endpoint}")
    assert response.status_code == 404

• test.py(테스트 파일)에서 테스트할 함수에 인자로 confest.py의 인자로 정의해둔 함수를 넣어주면 된다.
• 이 때 굳이 confest.py의 함수를 test.py로 import 하지 않아도 된다.

• 로그인 토큰에 접목시킨다면 이렇게 할 수 있을 것이다. (FastAPI 공식 Docs의 템플릿 프로젝트 출처)

• confest.py
  

• test_items.py
  

• confest.py에서 정의한 superuser_token_header 함수가 test_items.py에서 테스트 될때 작동하여 토큰을 발급받고 이를 헤더에 주입한다.