[FastAPI]SQLAlchemy + Pydantic을 이용한 게시판 만들기-(2-2)Feed CRUD - Routes, main.py

목차

 

FastAPI + SQLAlchemy + Pydantic을 이용한 게시판 만들기

 

[FastAPI]FastAPI 튜토리얼

[FastAPI]SQLAlchemy + Pydantic를 이용한 게시판 만들기-(1) 회원가입 및 로그인 구현(JWT)

[FastAPI]SQLAlchemy + Pydantic를 이용한 게시판 만들기-(1.5) 디테일 수정

[FastAPI]SQLAlchemy + Pydantic를 이용한 게시판 만들기-(2-1)Feed CRUD - Model, Service

[FastAPI]SQLAlchemy + Pydantic을 이용한 게시판 만들기-(2-2)Feed CRUD - Routes, main.py

[FastAPI]SQLAlchemy + Pydantic을 이용한 게시판 만들기-(3)S3 버킷을 이용한 사진 업로드

[FastAPI]SQLAlchemy + Pydantic를 이용한 게시판 만들기-(4-1)Comment CRUD - Models, Services

[FastAPI]SQLAlchemy + Pydantic를 이용한 게시판 만들기-(4-2)Comment CRUD - Routes, main.py

[FastAPI]SQLAlchemy + Pydantic를 이용한 게시판 만들기-(5)async

 

지난 글에 이어서 feed관련 라우팅 함수와 메인 함수를 구현하도록 하겠다.

 

feed_router.py

 

관심사의 분리를 위해 라우터 파일에는 가능한 서비스 레이어를 호출하는 로직만 넣으려고 애를 썼다.

 

다만, JWT 관련 인증 로직은 이곳에서 처리하도록 했는데,

 

이런 식으로 구성하는 것이 보편적인 방법이라고 했다.

 

먼저 구현을 보자.

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from models.feed import FeedCreate, FeedResponse, FeedUpdate
from services import feed_service, auth_service
from config.db import get_db
from typing import List

router = APIRouter()


@router.post("/create", response_model=FeedResponse)
def create(
    feed: FeedCreate,
    db: Session = Depends(get_db),
    email: str = Depends(auth_service.get_current_user_authorization),
):
    if email is None:
        raise HTTPException(status_code=401, detail="Not authorized")

    return feed_service.create_feed(db, feed, email)


@router.get("/read/{feed_id}", response_model=FeedResponse)
def read_feed(feed_id: int, db: Session = Depends(get_db)):
    return feed_service.get_feed_by_id(db, feed_id)


@router.get("/list", response_model=List[FeedResponse])
def list_feeds(db: Session = Depends(get_db)):
    return feed_service.get_feeds(db)


@router.patch("/update/{feed_id}", response_model=FeedResponse)
def update(
    feed_id: int,
    feed: FeedUpdate,
    db: Session = Depends(get_db),
    email: str = Depends(auth_service.get_current_user_authorization),
):
    if email is None:
        raise HTTPException(status_code=401, detail="Not authorized")

    return feed_service.update_feed(db, feed_id, feed, email)


@router.delete("/delete/{feed_id}", response_model=None)
def delete(
    feed_id: int,
    db: Session = Depends(get_db),
    email: str = Depends(auth_service.get_current_user_authorization),
):
    if email is None:
        raise HTTPException(status_code=401, detail="Not Authorized")

    feed_service.delete_feed(db, feed_id, email)

순서대로 살펴보자.

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from models.feed import FeedCreate, FeedResponse, FeedUpdate
from services import feed_service, auth_service
from config.db import get_db
from typing import List

router = APIRouter()

먼저 필요한 패키지, 라이브러리, 함수 등을 가져온다.

 

그 뒤에 APIRouter()를 이용해 APIRouter 객체를 생성한다.

 

지난 글에선 그냥 넘어갔지만 이 부분을 좀 더 자세히 설명하면 위와 같은 인스턴스 생성은

 

해당 모듈을 마치 작은 FastAPI 앱처럼 동작하게 한다.

 

이후 메인 앱에서 이 작은 앱을 마운트 해서 사용하는 것이 FastAPI 아키텍처인데,

 

이처럼 다른 도메인과 역할에 따라 라우트를 모듈화 하면 유지관리와 협업에 좋다고 한다.

@router.post("/create", response_model=FeedResponse)
def create(
    feed: FeedCreate,
    db: Session = Depends(get_db),
    email: str = Depends(auth_service.get_current_user_authorization),
):
    if email is None:
        raise HTTPException(status_code=401, detail="Not authorized")

    return feed_service.create_feed(db, feed, email)

가장 먼저 앞으로 생성할 함수의 엔드포인트와 반환 형태를 지정한다.

 

그다음 게시글을 생성하는 create 함수를 선언한다. 이 함수는 아래와 같은 파라미터를 받는다.

 

• feed: FeedCreat - 요청 본문에서 'FeedCreate'모델에 해당하는 데이터를 가져온다.
• db: Session = Depends(get_db)
  Depends를 사용해 get_db를 주입받은 뒤 get_db가 반환하는 데이터베이스 세션을 가져온다. 여기서 get_db는 세션을 생성/반환하는 함수이며, 이를 통해 create 함수 내부에서 db를 사용해 데이터베이스 작업을 수행할 수 있다.
• email: str = Depends(auth_service.get_current_user_authorization)
  마찬가지로 auth_service.get_current_user_authorization를 주입받아 요청을 보낸 사용자의 이메일을 인증한다.
  여기서 auth_service.get_current_user_authorization는 JWT를 이용해 이메일을 검증 및 반환하는 함수이다.

계속해서 이메일이 존재하지 않는다면 401 에러를 반환하며,

 

모든 검증이 끝난 뒤 서비스 레이어의 create_feed를 호출하여 DB에 새로운 피드를 생성한다.

 

반환값은 서비스 레이어에서 만들어주는 FeedResponse에 따라 구성된다.

 

필요한 곳에서 필요한 함수를 주입받아 사용할 수 있어서 코드의 가독성이나 생산성이 올라갈 것 같다고 생각했다.

@router.get("/read/{feed_id}", response_model=FeedResponse)
def read_feed(feed_id: int, db: Session = Depends(get_db)):
    return feed_service.get_feed_by_id(db, feed_id)


@router.get("/list", response_model=List[FeedResponse])
def list_feeds(db: Session = Depends(get_db)):
    return feed_service.get_feeds(db)

특정 피드와 전체 피드를 읽는 라우팅 함수를 작성한다.

 

기본적으로 피드는 모든 사람이 볼 수 있기 때문에 인증 과정을 생략하고 간단하게 만들었다.

@router.patch("/update/{feed_id}", response_model=FeedResponse)
def update(
    feed_id: int,
    feed: FeedUpdate,
    db: Session = Depends(get_db),
    email: str = Depends(auth_service.get_current_user_authorization),
):
    if email is None:
        raise HTTPException(status_code=401, detail="Not authorized")

    return feed_service.update_feed(db, feed_id, feed, email)

다음으로 게시글 수정 부분이다.

 

생성 부분과 대체로 같지만 PathParameter로 게시글 번호를 받고 있다.

@router.delete("/delete/{feed_id}", response_model=None)
def delete(
    feed_id: int,
    db: Session = Depends(get_db),
    email: str = Depends(auth_service.get_current_user_authorization),
):
    if email is None:
        raise HTTPException(status_code=401, detail="Not Authorized")

    feed_service.delete_feed(db, feed_id, email)

마지막으로 삭제.

 

지난 글에서 작성한 삭제 로직을 호출하기 전 JWT를 이용한 이메일 인증을 한다.

 

위와 같이 라우터 레이어에서는 요청을 받고, 요청을 검증해 서비스로직으로 전달하는 식으로 구현해 보았다.

 

이는 당연하게도 확실한 계층 분리로 인해 유지보수를 쉽게 만든다.

 

main.py

 

마지막으로 main 파일을 수정하고 끝내자.

from fastapi import FastAPI
from sqlalchemy import create_engine
from models.user import Base as UserBase  # User의 Base 클래스
from models.feed import Base as FeedBase  # Feed의 Base 클래스
from routers import auth_router
from routers import feed_router

DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL)

UserBase.metadata.create_all(bind=engine)
FeedBase.metadata.create_all(bind=engine)

app = FastAPI()

app.include_router(auth_router.router, prefix="/api/auth", tags=["auth"])
app.include_router(feed_router.router, prefix="/api/feed", tags=["feed"])


@app.get("/")
def read_root():
    return {"message": "Hello, World!"}

이전 버전에서 크게 변한 것은 없다.

 

코드 일관성을 위해 테이블 생성 로직을 추가하고

 

라우터 함수를 마운트 해 사용할 수 있게 만든다.

 

여기까지 하면 일단 기본적인 게시글 로직은 끝이다.

 

확실히 FastAPI는 러닝커브가 가파르지만 직관적인 코딩스타일 때문인지 개발자 친화적이라는 생각이 많이 든다.

 

다음 글에서는 무엇을 구현해 볼까..

 

일단 끝!