[FastAPI]SQLAlchemy + Pydantic을 이용한 게시판 만들기-(2-1)Feed CRUD - Models, Services

목차

 

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

 

지난 글에선 JWT를 이용한 회원가입과 로그인 로직을 구현했다.

 

이번 글에선 그에 이어 게시글(Feed) CRUD를 구현해 보도록 하겠다.

 

목적지는 다음과 같다.

 

구현해야 하는 코드의 길이가 꽤 길지만, 일단 시작해 보자.

 

feed.py

 

가장 먼저 피드 모델 파일을 구현해야 한다.

 

지난번 회원 때와 마찬가지로 SQLAlchemy 모델과 Pydantic 모델을 구현한다.

from sqlalchemy import Column, String, Integer, ForeignKey
from sqlalchemy.orm import relationship
from config.db import Base
from pydantic import BaseModel


class Feed(Base):
    __tablename__ = "feeds"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    content = Column(String)
    author_email = Column(String, ForeignKey("users.email"))

    author = relationship("User", back_populates="feeds")


class FeedCreate(BaseModel):
    title: str
    content: str


class FeedUpdate(BaseModel):
    title: str
    content: str


class FeedInDB(FeedCreate):
    pass


class FeedResponse(FeedCreate):
    id: int
    author_email: str
    author_nickname: str

import 부분은 생략하고 모델을 뜯어보자.

class Feed(Base):
    __tablename__ = "feeds"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    content = Column(String)
    author_email = Column(String, ForeignKey("users.email"))

    author = relationship("User", back_populates="feeds")

SQLAlchemy 모델이다. 테이블 이름을 feeds로 정하며 시작한다.

 

id, title, content, author_email 이렇게 네 개로 이루어진 간단한 클래스이며,

 

author_email은 유저 테이블의 email 필드를 참조하는 외래키가 된다.

 

놀랍게도 SQLAlchemy에서는 이것만으로도 1:N 관계를 표현할 수 있으며,

 

경험해 본 결과 기능상 아무런 이상이 없었다.

 

하지만 좀 더 명확하게 1:N 관계를 표현하기 위해선 마지막 줄과 같은 설정을 해주어야 하며,

 

user 모델에서도 아래와 같이 설정해주어야 한다.

class User(Base):
    __tablename__ = "users"

    email = Column(String, primary_key=True, unique=True, index=True)
    password = Column(String())
    nickname = Column(String())

    # 1:N 관계 설정
    feeds = relationship("Feed", back_populates="author")

이는 잠시 후 보게 될 명시적인 join 연산을 사용하지 않아도 될 정도로 편하게 만들지만,

 

성능을 중요시하는 경우엔 join 연산을 그대로 사용하는 것이 좋다고 한다.

class FeedCreate(BaseModel):
    title: str
    content: str


class FeedUpdate(BaseModel):
    title: str
    content: str


class FeedInDB(FeedCreate):
    pass


class FeedResponse(FeedCreate):
    id: int
    author_email: str
    author_nickname: str

다음으로는 Pydantic 모델이다.

 

게시글 생성, 수정 및 응답에 사용될 필드와 디비에 추가로 저장할 필드를 정의한다.

 

회원 모델 때와 마찬가지로 FeedInDB는 비워둔다.

 

feed_service.py

 

계속해서 서비스 레이어 구현을 해보자.

 

우선 간단한 CRUD를 구현하고 복잡한 기능은 다음 글에서 하기로 하자.

from fastapi import HTTPException
from sqlalchemy.orm import Session
from sqlalchemy import join
from models.feed import Feed, FeedCreate, FeedUpdate
from models.user import User


def create_feed(db: Session, feed: FeedCreate, author_email: str):
    feed_dict = feed.model_dump()
    feed_dict["author_email"] = author_email

    author = db.query(User).filter(User.email == author_email).first()
    if author is None:
        raise HTTPException(status_code=404, detail="Author Not Found")

    author_nickname = author.nickname

    db_feed = Feed(**feed_dict)
    db.add(db_feed)
    db.commit()
    db.refresh(db_feed)

    return {
        "id": db_feed.id,
        "title": db_feed.title,
        "content": db_feed.content,
        "author_email": db_feed.author_email,
        "author_nickname": author_nickname,
    }


def update_feed(db: Session, feed_id: int, feed_update: FeedUpdate, email: str):
    db_feed = db.query(Feed).filter(Feed.id == feed_id).first()

    if db_feed is None:
        raise HTTPException(status_code=404, detail="Feed Not Found")

    if db_feed.author_email != email:
        raise HTTPException(status_code=403, detail="Permission Denied")

    author = db.query(User).filter(User.email == email).first()
    author_nickname = author.nickname

    db_feed.title = feed_update.title
    db_feed.content = feed_update.content
    db_feed.author_nickname = author_nickname

    db.commit()
    db.refresh(db_feed)

    return db_feed


def get_feed_by_id(db: Session, feed_id: int):
    feed_data = (
        db.query(Feed, User.nickname)
        .join(User, User.email == Feed.author_email)
        .filter(Feed.id == feed_id)
        .first()
    )

    if feed_data is None:
        raise HTTPException(status_code=404, detail="Feed not found")

    feed, nickname = feed_data
    feed_response = {
        "id": feed.id,
        "title": feed.title,
        "content": feed.content,
        "author_email": feed.author_email,
        "author_nickname": nickname,  # 닉네임 추가
    }

    return feed_response


def get_feeds(db: Session):
    feeds = db.query(Feed, User.nickname).join(User, User.email == Feed.author_email).all()
    feed_responses = []

    for feed, nickname in feeds:
        feed_dict = {
            "id": feed.id,
            "title": feed.title,
            "content": feed.content,
            "author_email": feed.author_email,
            "author_nickname": nickname,
        }
        feed_responses.append(feed_dict)

    return feed_responses


def delete_feed(db: Session, feed_id: int, email: str):
    db_feed = db.query(Feed).filter(Feed.id == feed_id).first()

    if db_feed is None:
        raise HTTPException(status_code=404, detail="Feed Not Found")

    if db_feed.author_email != email:
        raise HTTPException(status_code=403, detail="Permission Denied")

    db.delete(db_feed)
    db.commit()

코드가 길지만 위에서부터 하나씩 뜯어보자.

 

역시 import문은 생략한다.

def create_feed(db: Session, feed: FeedCreate, author_email: str):
    feed_dict = feed.model_dump()
    feed_dict["author_email"] = author_email

    author = db.query(User).filter(User.email == author_email).first()
    if author is None:
        raise HTTPException(status_code=404, detail="Author Not Found")

    author_nickname = author.nickname

    db_feed = Feed(**feed_dict)
    db.add(db_feed)
    db.commit()
    db.refresh(db_feed)

    return {
        "id": db_feed.id,
        "title": db_feed.title,
        "content": db_feed.content,
        "author_email": db_feed.author_email,
        "author_nickname": author_nickname,
    }

먼저 create_feed에 대해 뜯어보자.

def create_feed(db: Session, feed: FeedCreate, author_email: str):

create_feed 함수를 정의한다. 이 함수는 db 세션, 생성될 피드의 정보(feed),

 

그리고 작성자의 이메일('author_email)을 매개변수로 받는다.

 

독특한 점은 Session을 핵심 인자로 받는 부분인데, 위와 같이 설정하면 해당 함수가 호출될 때마다

 

새로운 Session 객체가 생성되어 DB에 관련 작업을 할 수 있다.

 

구체적인 의존성 주입과 매개변수로의 할당은 라우팅 함수를 보면 더 자세히 알 수 있다.

    feed_dict = feed.model_dump()
    feed_dict["author_email"] = author_email

윗 줄에선 FeedCreate 모델의 객체인 feed로부터 feed_dict라는 딕셔너리를 생성한다.

 

이어지는 줄에는 feed_dict["author_email"]을 선언하고 새로운 값을 할당하는데,

 

그냥 이렇게만 작업해도 feed_dict에는 새로운 키-밸류가 생성된다.

    author = db.query(User).filter(User.email == author_email).first()
    if author is None:
        raise HTTPException(status_code=404, detail="Author Not Found")

    author_nickname = author.nickname

계속해서 디비에서 이메일이 'auth_email'과 일치하는 사용자를 조회하며,

 

해당 이용자가 없다면 예외를 던진다.

 

만약 있다면 해당 사용자의 닉네임은 author_nickname에 저장한다.

    db_feed = Feed(**feed_dict)
    db.add(db_feed)
    db.commit()
    db.refresh(db_feed)

가장 윗줄에서는 여태 만들어놓은 딕셔너리를 언패킹해 Feed 클래스의 인스턴스를 생성, 및 초기화한다.

 

여기서 '**feed_dict' 문법은 딕셔너리에 존재하는 모든 키-값 쌍을 인자로 전달하는 역할을 한다.

 

즉, 위에서 딕셔너리에 넣어준 author_email 역시 자연스럽게 피드 인스턴스로 들어간다는 뜻이다.

 

지금까지 이어온 부분이 굉장히 직관적이며 편리하다는 생각이 들었다.

 

어쨌거나 그 이후엔 디비에 새로 생성된 db_feed를 추가하고, 커밋하고, 피드 최신정보로 DB를 갱신한다.

    return {
        "id": db_feed.id,
        "title": db_feed.title,
        "content": db_feed.content,
        "author_email": db_feed.author_email,
        "author_nickname": author_nickname,
    }

함수의 결과로 딕셔너리를 반환한다. 이 딕셔너리는 라우팅 메서드에서 받아 그대로 처리하게 된다.

 

참고로, 여기서 위 딕셔너리처럼 반환하지 않고 db_feed를 리턴해도 되지만

 

필요한 정보와 보안에 민감한 부분을 위해 반환 값을 명시적으로 주는 것이 나중을 위해 좋다고 한다.

def get_feed_by_id(db: Session, feed_id: int):
    feed_data = (
        db.query(Feed, User.nickname)
        .join(User, User.email == Feed.author_email)
        .filter(Feed.id == feed_id)
        .first()
    )

    if feed_data is None:
        raise HTTPException(status_code=404, detail="Feed not found")

    feed, nickname = feed_data

    return {
        "id": feed.id,
        "title": feed.title,
        "content": feed.content,
        "author_email": feed.author_email,
        "author_nickname": nickname,
    }

다음으로는 피드 번호로 피드를 읽어오는 로직이다.

 

여기선 디비 세션과 피드 아이디만 받는다.

 

이어서 두 테이블을 조인해서 필요한 정보를 얻어오는데, 처음이니까 이 부분도 뜯어보자.

    feed_data = (
        db.query(Feed, User.nickname)

여기서는 가져올 정보를 명시한다. Feed 테이블과 User테이블의 nickcame 컬럼을 목표로 한다.

        .join(User, User.email == Feed.author_email)

유저 테이블과 피드 테이블을 이메일 컬럼을 기준으로 조인한다.

 

즉, 두 테이블에서의 이메일 값이 같은 행만 가져오라는 뜻이다.

        .filter(Feed.id == feed_id)

가져온 행에서 feed_id와 일치하는 행만 필터링해서 가져온다.

        .first()
    )

일치하는 행 중 첫 번째 결과만 가져오라는 의미이다.

 

쿼리가 실행되면 feed_data 변수에는 (Feed 객체, 닉네임) 형태의 튜플이 저장된다.

 

만약 없다면 None을 반환해 뒤에서 예외처리를 할 수 있게 해준다.

    if feed_data is None:
        raise HTTPException(status_code=404, detail="Feed not found")

검색된 데이터가 없을 시 예외처리 로직이다.

    feed, nickname = feed_data

튜플로 되어있는 feed_data에서 필요한 정보를 각각 가져와 할당한다.

    return {
        "id": feed.id,
        "title": feed.title,
        "content": feed.content,
        "author_email": feed.author_email,
        "author_nickname": nickname,
    }

마지막으론 위와 같이 최종 반환 값을 결정한다.

 

조인하고 쿼리한 뒤 해당 데이터를 사용하는 방법 역시 직관적이고 편리했다.

def get_feeds(db: Session):
    feeds = db.query(Feed, User.nickname).join(User, User.email == Feed.author_email).all()
    feed_responses = []

    for feed, nickname in feeds:
        feed_dict = {
            "id": feed.id,
            "title": feed.title,
            "content": feed.content,
            "author_email": feed.author_email,
            "author_nickname": nickname,
        }
        feed_responses.append(feed_dict)

    return feed_responses

이번에는 존재하는 모든 피드를 가져오는 로직이다.

 

마찬가지로 유저와 피드를 조인한 뒤에 모든 행을 가져오지만 여기선 필터링을 하지 않는다.

 

다음 줄에서는 결괏값을 담을 feed_responses 리스트를 생성하고,

 

반복문을 돌면서 feed_dict를 생성해 feed_response에 저장한다.

 

그 다음은 출력.

 

현재로서는 어떤 복잡한 로직도 없기 때문에 일단 여기서 넘어간다.

def update_feed(db: Session, feed_id: int, feed_update: FeedUpdate, email: str):
    db_feed = db.query(Feed).filter(Feed.id == feed_id).first()

    if db_feed is None:
        raise HTTPException(status_code=404, detail="Feed Not Found")

    if db_feed.author_email != email:
        raise HTTPException(status_code=403, detail="Permission Denied")

    author = db.query(User).filter(User.email == email).first()
    author_nickname = author.nickname

    db_feed.title = feed_update.title
    db_feed.content = feed_update.content
    db_feed.author_nickname = author_nickname

    db.commit()
    db.refresh(db_feed)

    return {
        "id": db_feed.id,
        "title": db_feed.title,
        "content": db_feed.content,
        "author_email": db_feed.author_email,
        "author_nickname": author_nickname,
    }

다음으로는 게시글 수정 로직이다.

def update_feed(db: Session, feed_id: int, feed_update: FeedUpdate, email: str):
    db_feed = db.query(Feed).filter(Feed.id == feed_id).first()

    if db_feed is None:
        raise HTTPException(status_code=404, detail="Feed Not Found")

    if db_feed.author_email != email:
        raise HTTPException(status_code=403, detail="Permission Denied")

이 부분은 반복되는 부분이니 생략한다. feed_id라는 매개변수가 하나 늘었다.

    author = db.query(User).filter(User.email == email).first()
    author_nickname = author.nickname

피드의 작성자 정보를 가져와 닉네임을 author_nickname에 할당한다.

    db_feed.title = feed_update.title
    db_feed.content = feed_update.content
    db_feed.author_nickname = author_nickname

    db.commit()
    db.refresh(db_feed)

제목과 내용, 작성자 닉네님을 업데이트 한 뒤,

 

DB에 커밋 후 객체를 새로고침 해준다.

    return {
        "id": db_feed.id,
        "title": db_feed.title,
        "content": db_feed.content,
        "author_email": db_feed.author_email,
        "author_nickname": author_nickname,
    }

끝으로 리턴.

def delete_feed(db: Session, feed_id: int, email: str):
    db_feed = db.query(Feed).filter(Feed.id == feed_id).first()

    if db_feed is None:
        raise HTTPException(status_code=404, detail="Feed Not Found")

    if db_feed.author_email != email:
        raise HTTPException(status_code=403, detail="Permission Denied")

    db.delete(db_feed)
    db.commit()

마지막으로 피드를 삭제하는 로직이다. 역시 feed_id를 매개변수로 받으며,

 

해당 피드가 없는 경우, 혹은 사용자 이메일과 요청을 보낸 이메일이 다른 경우 예외처리를 해주며

 

두 개의 필터를 지나도 걸리지 않으면 디비에서 해당 피드를 삭제하고 커밋한다.

 

---

이렇게 해서 모델과 서비스 레이어의 구현까지 끝났다.

 

라우팅 함수와 메인 함수에 대해서도 이어서 적고 싶지만 읽어야 할 코드가 길어

 

나부터도 피로감이 드는 느낌이라 여기서 한 번 잘랐다 가야겠다.

 

바로 이어지는 글에서는 피드 관련 구현을 마무리 할 예정이다.

 

끝!