FastAPI & Svelte - 질문 목록 API

https://wikidocs.net/176226

2-04-1 라우터

 

* 해당 프로젝트는 위키독스를 참조하며 프로젝트에 대한 모든 내용을 담고 있지 않습니다.

진행되는 프로젝트의 모든 내용은 위키독스를 참조해주시기 바랍니다.

 

---

2-1. 라우터

Svelte 서버 실행: npm run dev (proj/myapi/frontend)

FastAPI 서버 실행: uvicorn main:app --reload(proj/myapi)

라우터 만들기

디렉터리 생성: proj\myapi\domain\question

# proj/myapi/domain/question/question_router.py
from fastapi import APIRouter

from database import SessionLocal
from models import Question

router = APIRouter(
    prefix="/api/question",
)

@router.get("/list")
def question_list():
    db = SessionLocal()
    _question_list = db.query(Question).order_by(Question.create_date.desc()).all()
    db.close()
    return _question_list

 

라우팅

 

FastAPI가 요청받은 URL을 해석하여 그에 맞는 함수를 실행. 결과값을 리턴하는 행위

• router 객체 생성시 사용한 prefix 속성은 요청 URL에 항상 포함되어야 한다.
• question_list 함수는 db 세션을 생성, 질문 목록을 조회하여 리턴하는 함수.
• db.close()는 세션 종료가 아닌 세션을 커넥션 풀에 반환.

# main.py
from fastapi import FastAPI
from starlette.middleware.cors import CORSMiddleware

from domain.question import question_router

app = FastAPI()

origins = [
    "http://localhost:5173", # 또는 "http://127.0.0.1:5173"
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials = True,
    allow_methods = ["*"],
    allow_headers = ["*"],
)

app.include_router(question_router.router)

질문 목록 API 테스트

http://127.0.0.1:8000/docs

• question_list의 응답으로 리스트를 리턴하더라도 실제 리턴되는 값은 json 문자열로 자동 변환된다.

 

2-2 의존성 주입

데이터베이스 세션의 생성과 반환 자동화

question_list 함수를 보면 db 세션 객체를 생성하고 함수 종료 직전에 다시 db.close()를 호출한다.

db.close()를 수행하지 않으면 SQLAlchemy가 사용하는 커넥션 풀에 db 세션이 반환되지 않아 문제가 생긴다.

이러한 부분을 자동화 하려면 FastAPI의 Dependency Injection(의존성 주입)을 사용하여 자동화 할 수 있다.

# proj/myapi/database.py
import contextlib

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

SQLALCHEMY_DATABASE_URL = "sqlite:///./myapi.db"

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

Base = declarative_base()

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

• yield 함수를 사용하여 이터레이터 생성 (제너레이터)
• @contextlib.contextmanager 어노테이션을 적용했으므로 다음과 같이 with 문과 함께 사용 가능

 

with get_db() as db:
    # db 세션 객체 사용

• with 문을 벗어나는 순간 get_db 함수 finally에 작성한 db.close() 함수가 자동 실행

 

# proj/myapi/domain/question/quesion_router.py
from fastapi import APIRouter

from database import get_db
from models import Question

router = APIRouter(
    prefix="/api/question",
)

@router.get("/list")
def question_list():
    with get_db() as db:
        _question_list = db.query(Question).order_by(Question.create_date.desc()).all()
    return _question_list

• 오류 여부에 상관 없이 with 문을 벗어나는 순간 db.close()가 실행

 

Depends 사용하기

Dependencies - FastAPI

Dependencies - FastAPI

 

FastAPI의 Depends를 사용하면 with 문을 사용하는 것 보다 더 간단하게 사용할 수 있다.

# proj/myapi/domain/question/question_router.py
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session

from database import get_db
from models import Question

router = APIRouter(
    prefix="/api/question",
)

@router.get("/list")
def question_list(db: Session = Depends(get_db)):
    _question_list = db.query(Question).order_by(Question.create_date.desc()).all()
    return _question_list

• question_list 함수의 매개변수로 db: Session = Depends(get_db) 객체를 주입.
• db: Session 문장의 의미는 db 객체가 Session 타입임을 의미한다.
• FastAPI의 Depends는 매개변수로 전달 받은 함수를 실행시킨 결과를 리턴한다.
  따라서 db: Session = Depends(get_db)의 db 객체에는 get_db 제너레이터에 의해 생성된 세션 객체가 주입된다.
• 이 때 get_db 함수에 자동으로 contextmanager가 적용되기 때문에(_Depends_에서 contextmanager를 적용하게끔 설계되어 있다.) database.py의 get_db 함수는 다음과 같이 적용한 @contextlib.contextmanager 어노테이션을 제거해야 한다.
  제거하지 않으면 2중으로 적용되어 오류가 발생한다.

#proj/myapi/database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./myapi.db"

engine = create_engine(
    SQLALCHEMY_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()

 

스키마

_question_list = db.query(Question).order_by(Question.create_date.desc()).all()

 

현재 질문 목록 API는 Question 모델의 모든 항목이 출력으로 리턴된다. 이 경우 외부로 공개되면 안되는 출력 항목이 출력되거나 출력 값의 검증이 어렵다는 단점이 존재한다.

이러한 상황에 Pydantic 라이브러리를 사용할 수 있다.

Pydantic

FastAPI의 입출력 스펙을 정의하고 그 값을 검증하기 위해 사용하는 라이브러리.

API의 입출력 항목을 다음과 같이 정의하고 검증할 수 있다.

• 입출력 항목의 갯수와 타입 설정
• 입출력 항목의 필수값 체크
• 입출력 항목의 데이터 검증

Pydantic을 적용하기 위해 가장 먼저 할 일은 질문 목록 API의 출력 스키마를 생성해야 한다.

스키마는 보통 데이터의 구조와 명세를 의미한다. 즉, 출력 스키마는 출력 항목이 몇 개인지, 제약 조건은 어떠한 것이 있는지 등을 기술하는 것을 의미한다.

 

Pydantic 스키마 작성하기

# proj/myapi/domain/question/question_schema.py
import datetime

from pydantic import BaseModel

class Question(BaseModel):
    id: int
    subject: str
    content: str
    create_date: datetime.datetime

• Question 스키마 작성
• 4개의 출력항목 지정. 각 4개 항목은 모두 디폴트 값이 없기 때문에 필수 항목.
• 만약 subject 항목이 필수 항목이 아니게 설정하려면 다음처럼 수행

subject: str | None = None

 

subject 항목은 문자열 또는 None을 가질 수 있고 디폴트 값을 None으로 처리한다는 의미.

 

다만, Python 3.10 이상의 버전에서만 위의 | 연산자를 사용할 수 있으며 Python 3.9 버전 이하에서는

아래와 같이 Optional 또는 Union 을 사용하면 된다.

 

# Optional 사용 예시
# proj/myapi/domain/question/question_schema.py
import datetime

from pydantic import BaseModel
from typing import Union, Optional

class Question(BaseModel):
    id: int
    subject: Optional[str] = None
    content: str
    create_date: datetime.datetime
    
# Union 사용 예시. 위 코드와 아래 코드 둘 중 하나를 골라서 사용하면 된다.
# proj/myapi/domain/question/question_schema.py
import datetime

from pydantic import BaseModel
from typing import Union, Optional

class Question(BaseModel):
    id: int
    subject: Union[str, None]
    content: str
    create_date: datetime.datetime

 

 

라우터에 Pydantic 적용하기

# proj/myapi/domain/question/question_router.py
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session

from database import get_db
from domain.question import question_schema
from models import Question

router = APIRouter(
    prefix="/api/question",
)

@router.get("/list", response_model=list[question_schema.Question])
def question_list(db: Session = Depends(get_db)):
    _question_list = db.query(Question).order_by(Question.create_date.desc()).all()
    return _question_list

• response_model 속성을 추가하여 Question 스키마로 구성된 리스트임을 지정.
• 만약 Question 스키마에서 content 항목을 제거한다면 질문 목록 API의 출력 항목에도 content 항목이 제거될 것. 이 때, 실제 리턴되는 _question_list_를 수정할 필요가 없다.

 

CRUD

_questoin_list = db.query(Question).order_by(Question.create_date.desc()).all()

 

라우터에 위와 같이 데이터를 조회하는 부분을 포함해도 문제는 없지만, 서로 다른 라우터에서 데이터를 처리하는 부분이 동일하여 중복되는 현상이 나타날 수 있기 때문에 분리하여 작성.

 

# proj/myapi/domain/question/question_crud.py
from models import Question
from sqlalchemy.orm import Session

def get_question_list(db: Session):
    question_list = db.query(Question)\
        .order_by(Question.create_date.desc())\
        .all()
    return question_list

 

 

• 작성한 get_question_list 함수를 사용할 수 있도록 질문 목록 라우터 함수 수정

# proj/myapi/domain/question/question_router.py
from fastapi import APIRouter,Depends
from sqlalchemy.orm import Session

from database import get_db
from domain.question import question_schema,quesiton_crud
from models import Question

router = APIRouter(
    prefix="/api/question",
)

@router.get("/list", response_model=list[question_schema.Question])
def question_list(db: Session = Depends(get_db)):
    _question_list = quesiton_crud.get_question_list(db)
    return _question_list