티스토리 뷰

Python/FastAPI

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

Vagabund.Gni 2023. 10. 13. 12:16
728x90
반응형

목차

     

    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

     

    지난 글까지 해서 피드와 그에 딸린 사진 업로드 기능을 구현했다.

     

    이번 글에서는 간단하게 역시 부가기능이라 할 수 있는 코멘트 CRUD를 구현할 생각인데,

     

    여기서 간단하다는 의미는 코멘트에는 사진 업로드 기능을 포함하지 않겠다는 뜻이다.

     

    계속해서 오늘 구현을 마치면 아키텍처는 다음과 같은 모양이 된다.

     

    그럼 늘 하던 대로 시작해 보자.

     

    /models

     

    가장 먼저 코멘트를 사용하기 위한 모델 클래스를 꾸며준다.

     

    의존성을 위해 피드와 유저 모델도 바꾸어주어야 한다.

     

    comment.py

     

    코멘트 모델은 여타 모델과 마찬가지로 두 개의 모델로 나누어 구현한다.

     

    내가 잊어버릴까 봐 다시 언급하자면 SQLAlchemy 모델과 Pydantic 모델이 그것이다.

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


class Comment(Base):
    __tablename__ = "comments"

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

    author = relationship("User", back_populates="comments")
    feed = relationship("Feed", back_populates="comments")


class CommentCreate(BaseModel):
    content: str
    feed_id: int


class CommentUpdate(BaseModel):
    content: Optional[str]


class CommentInDB(CommentCreate):
    pass


class CommentResponse(CommentCreate):
    id: int
    author_email: str
    author_nickname: str

    코드도 역시 반으로 잘라 그 의미를 살펴보자. 임포트 문은 생략한다.

    class Comment(Base):
    __tablename__ = "comments"

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

    author = relationship("User", back_populates="comments")
    feed = relationship("Feed", back_populates="comments")

    먼저 SQLAlchemy의 Declarative Base 클래스를 상속받아 Comment 클래스를 만든다.

     

    Base 클래스를 상속함으로써 Comment 클래스는 SQLAlchemy가 관리하는 모델 클래스가 된다.

     

    간결하게 아이디와 내용을 필드로 갖고 두 개의 외래 키를 가지고 있으며

     

    relationship을 통해 특정 사용자, 특정 피드와의 관계를 설명한다.

    class CommentCreate(BaseModel):
    content: str
    feed_id: int


class CommentUpdate(BaseModel):
    content: Optional[str]


class CommentInDB(CommentCreate):
    pass


class CommentResponse(CommentCreate):
    id: int
    author_email: str
    author_nickname: str

    계속해서 Pydantic의 기본 클래스 BaseModel을 상속받아 두 개의 모델 클래스를 정의한다.

     

    BaseModel을 상속함으로써 두 클래스는 Pydantic에서 관리하는 모델 클래스가 되며, 유효성 검사를 쉽게 할 수 있게 된다.

     

    각 모델의 역할은 이름 그대로 코멘트 작성과 수정을 담당하고 있으며,

     

    이후의 두 클래스는 CommentCreate 자체를 다시 상속해 content와 feed_id 필드를 재사용한다.

     

    user.py

     

    사용자 모델에는 관계만 추가해 주면 된다.

    class User(Base):
    __tablename__ = "users"

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

    feeds = relationship("Feed", back_populates="author")
    comments = relationship("Comment", back_populates="author", post_update=True)

    코드의 위아래를 불가피하게 생략했으나 이전 글을 보면 전부 나와 있다!

     

    feed.py

     

    피드 모델도 마찬가지이다.

    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"))
    image_urls = Column(JSON, nullable=True)

    author = relationship("User", back_populates="feeds")
    comments = relationship("Comment", back_populates="feed", post_update=True)

     

    comment_service.py

     

    계속해서 비즈니스 로직 부분이다. 크게 어려운 부분은 없지만 나를 위해 전체를 공유 후 토막 내 설명한다.

    from fastapi import HTTPException
from sqlalchemy.orm import Session
from models.comment import Comment, CommentCreate, CommentUpdate
from models.user import User
from models.feed import Feed
import logging

logging.basicConfig(level=logging.DEBUG)


def create_comment(db: Session, comment: CommentCreate, author_email: str):
    comment_dict = comment.model_dump()
    comment_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

    feed = db.query(Feed).filter(Feed.id == comment.feed_id).first()
    if feed is None:
        raise HTTPException(status_code=404, detail="Feed Not Found")

    db_comment = Comment(**comment_dict)
    db.add(db_comment)
    db.commit()
    db.refresh(db_comment)

    return {
        "id": db_comment.id,
        "content": db_comment.content,
        "author_email": db_comment.author_email,
        "author_nickname": author_nickname,
        "feed_id": db_comment.feed_id,
    }


def get_comment_by_feed_id(db: Session, feed_id: int):
    comments = db.query(Comment).filter(Comment.feed_id == feed_id).all()
    comment_responses = []

    for comment in comments:
        author = db.query(User).filter(User.email == comment.author_email).first()

        author_nickname = author.nickname

        comment_responses.append(
            {
                "id": comment.id,
                "content": comment.content,
                "author_email": comment.author_email,
                "author_nickname": author_nickname,
                "feed_id": comment.feed_id,
            }
        )

    return comment_responses


def update_comment(db: Session, comment_id: int, comment_update: CommentUpdate, email: str):
    db_comment = db.query(Comment).filter(Comment.id == comment_id).first()

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

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

    db_comment.content = comment_update.content or db_comment.content

    db.commit()
    db.refresh(db_comment)

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

    author_nickname = author.nickname

    return {
        "id": db_comment.id,
        "content": db_comment.content,
        "author_email": db_comment.author_email,
        "author_nickname": author_nickname,
        "feed_id": db_comment.feed_id,
    }


def delete_comment(db: Session, comment_id: int, email: str):
    db_comment = db.query(Comment).filter(Comment.id == comment_id).first()

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

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

    db.delete(db_comment)
    db.commit()

    임포트문은 생략하고 다음부터 본다.

    def create_comment(db: Session, comment: CommentCreate, author_email: str):
    comment_dict = comment.model_dump()
    comment_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

    feed = db.query(Feed).filter(Feed.id == comment.feed_id).first()
    if feed is None:
        raise HTTPException(status_code=404, detail="Feed Not Found")

    db_comment = Comment(**comment_dict)
    db.add(db_comment)
    db.commit()
    db.refresh(db_comment)

    return {
        "id": db_comment.id,
        "content": db_comment.content,
        "author_email": db_comment.author_email,
        "author_nickname": author_nickname,
        "feed_id": db_comment.feed_id,
    }

    이 코드는 SQLAlchemy의 Session과 앞서 정의한 모델을 사용해 코멘트를 작성하는 로직을 담고 있다.

    def create_comment(db: Session, comment: CommentCreate, author_email: str):

    먼저 함수를 정의한다. Session과 CommentCreate모델, 그리고 라우터에서 전달할 회원 이메일을 받는다.

        comment_dict = comment.model_dump()

    가장 먼저 CommentCreate 객체를 .model_dump() 메서드를 이용해 딕셔너리로 바꿔준다.

        comment_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

    사용자를 찾았으면 해당 닉네임을 author_nickname에 할당한다.

        feed = db.query(Feed).filter(Feed.id == comment.feed_id).first()
    if feed is None:
        raise HTTPException(status_code=404, detail="Feed Not Found")

    계속해서 CreateComment의 입력값으로 주어질 feed_id와 일치하는 피드를 찾고,

     

    없으면 역시 예외를 던져준다.

        db_comment = Comment(**comment_dict)

    완성된 딕셔너리를 사용해 새로운 Comment 객체를 생성한다.

        db.add(db_comment)
    db.commit()
    db.refresh(db_comment)

    해당 객체를 디비 세션에 추가하고, 커밋한 뒤 최신 정보로 코멘트 객체를 업데이트한다.

        return {
        "id": db_comment.id,
        "content": db_comment.content,
        "author_email": db_comment.author_email,
        "author_nickname": author_nickname,
        "feed_id": db_comment.feed_id,
    }

    마지막으로 author_nickname까지 사용해서 코멘트의 정보를 딕셔너리 형태로 반환한다.

     

    피드에 이어 코멘트까지 구현하니 이제 좀 알듯 말듯한 기분이 든다.

    def get_comment_by_feed_id(db: Session, feed_id: int):
    comments = db.query(Comment).filter(Comment.feed_id == feed_id).all()
    comment_responses = []

    for comment in comments:
        author = db.query(User).filter(User.email == comment.author_email).first()

        author_nickname = author.nickname

        comment_responses.append(
            {
                "id": comment.id,
                "content": comment.content,
                "author_email": comment.author_email,
                "author_nickname": author_nickname,
                "feed_id": comment.feed_id,
            }
        )

    return comment_responses

    계속해서 피드 아이디로 해당 피드의 코멘트를 전부 조회하는 로직이다.

    def get_comment_by_feed_id(db: Session, feed_id: int):

    함수를 호출한다. 라우터로부터 Session 객체와 feed_id를 매개변수로 받는다.

        comments = db.query(Comment).filter(Comment.feed_id == feed_id).all()

    먼저 feed_id에 해당하는 코멘트를 디비에서 모두 조회한다.

        comment_responses = []

    반환값으로 넘겨줄 코멘트를 담을 빈 리스트를 생성한다.

        for comment in comments:

    각 코멘트에 대해 다음의 작업을 수행한다.

            author = db.query(User).filter(User.email == comment.author_email).first()

    먼저 코멘트 작성자의 정보를 디비에서 가져와 author에 할당한다.

            author_nickname = author.nickname

    해당 작성자의 닉네임을 author_nickname에 할당한다. 이 부분은 생략하고 바로 반환값에 넣어도 된다.

            comment_responses.append(
            {
                "id": comment.id,
                "content": comment.content,
                "author_email": comment.author_email,
                "author_nickname": author_nickname,
                "feed_id": comment.feed_id,
            }
        )

    위에서 만든 빈 배열에 반환값에 필요한 딕셔너리를 추가한다.

        return comment_responses

    배열을 반환한다.

     

    솔직히 말해서 너무 편리하고 간결하고 직관적이라 적응이 안 될 정도다.

     

    계속 진행하자.

    def update_comment(db: Session, comment_id: int, comment_update: CommentUpdate, email: str):
    db_comment = db.query(Comment).filter(Comment.id == comment_id).first()

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

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

    db_comment.content = comment_update.content or db_comment.content

    db.commit()
    db.refresh(db_comment)

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

    author_nickname = author.nickname

    return {
        "id": db_comment.id,
        "content": db_comment.content,
        "author_email": db_comment.author_email,
        "author_nickname": author_nickname,
        "feed_id": db_comment.feed_id,
    }

    다음은 업데이트 로직이다.

    def update_comment(db: Session, comment_id: int, comment_update: CommentUpdate, email: str):

    함수를 선언한다. Session 객체와 comment_id, 그리고 CommentUpdate 객체와 email을 매개변수로 받는다.

     

    이렇게 써놓고 보니 create_comment 메서드와 일관성이 조금 떨어지는 것이 눈에 띄지만

     

    일단 넘어가기로 한다..

        db_comment = db.query(Comment).filter(Comment.id == comment_id).first()

    먼저 디비에서 해당 아이디를 가진 코멘트를 조회한다.

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

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

    코멘트가 없거나 작성자의 이메일이 일치하지 않을 경우 예외를 던진다.

     

    403 에러는 사용자가 인증은 되었으나 수정할 권한이 없다고 판단될 때를 위함이다.

        db_comment.content = comment_update.content or db_comment.content

    comment_update.content 값이 존재하면 해당 코멘트를 갱신해 준다.

        db.commit()
    db.refresh(db_comment)

    계속해서 디비에 변경사항을 반영하고 새로고침 해준다.

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

    author_nickname = author.nickname

    이어서 반환 값에 사용할 사용자 닉네임을 가져온 뒤

        return {
        "id": db_comment.id,
        "content": db_comment.content,
        "author_email": db_comment.author_email,
        "author_nickname": author_nickname,
        "feed_id": db_comment.feed_id,
    }

    딕셔너리로 만들어 반환해 준다.

    def delete_comment(db: Session, comment_id: int, email: str):
    db_comment = db.query(Comment).filter(Comment.id == comment_id).first()

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

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

    db.delete(db_comment)
    db.commit()

    마지막으로 삭제 로직이다.

    def delete_comment(db: Session, comment_id: int, email: str):

    Session 객체와 comment_id, email을 매개변수로 받는다.

        db_comment = db.query(Comment).filter(Comment.id == comment_id).first()

    이어서 해당 아이디와 일치하는 코멘트를 디비에서 조회하고,

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

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

    코멘트가 없거나 작성자가 다를 경우 각각 예외를 던져준다.

        db.delete(db_comment)
    db.commit()

    마지막으로 디비에서 삭제하고 변경사항을 반영.

     

    간단하게 마무리된다.

     

    역시 CRUD는 코드가 길어 글이 늘어진다.

     

    여기서 자르고 다음 글에서 마무리 해야겠다.

    반응형
    저작자표시 비영리 변경금지

    'Python > FastAPI' 카테고리의 다른 글

    [FastAPI]SQLAlchemy + Pydantic를 이용한 게시판 만들기-(5)async  (0) 2023.10.23
    [FastAPI]SQLAlchemy + Pydantic를 이용한 게시판 만들기-(4-2)Comment CRUD - Routes, main.py  (0) 2023.10.17
    [FastAPI]Base, BaseModel과 상속, 의존성 주입에 대하여  (0) 2023.10.16
    [FastAPI]SQLAlchemy + Pydantic을 이용한 게시판 만들기-(3)S3 버킷을 이용한 사진 업로드  (0) 2023.10.12
    [FastAPI]SQLAlchemy + Pydantic을 이용한 게시판 만들기-(2-2)Feed CRUD - Routes, main.py  (0) 2023.10.10
    [FastAPI]SQLAlchemy + Pydantic을 이용한 게시판 만들기-(2-1)Feed CRUD - Models, Services  (1) 2023.10.06
    댓글
    공지사항
    최근에 올라온 글
    최근에 달린 댓글
    Total
    Today
    Yesterday
    링크
    TAG more
    «   2024/11   »
    일 월 화 수 목 금 토
    1 2
    3 4 5 6 7 8 9
    10 11 12 13 14 15 16
    17 18 19 20 21 22 23
    24 25 26 27 28 29 30
    글 보관함