[FastAPI]SQLAlchemy + Pydantic를 이용한 게시판 만들기-(6)Follow, Like

목차

 

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를 이용한 게시판 만들기-(5)async

[FastAPI]SQLAlchemy + Pydantic를 이용한 게시판 만들기-(6)Follow, Like

 

이전 글에서 사용하는 DB를 바꾸고, 2.0 스타일로 바꾼 쿼리를 이용해 비동기처리를 마쳤다.

 

이제 대충 게시판이나 인스타그램 정도 되는 것 같은데, 빠진 기능이 있어서 메우고 가려고 한다.

 

바로 팔로우와 좋아요.

 

시작해 보자.

 

user.py

 

그런데 이 구현을 시작하자마자 뭔가 잘못되었다는 것을 깨달았다.

 

바로 기존의 유저 모델에서 자동 증가하는 번호가 아닌 이메일을 기본 키로 사용하고 있었던 것.

 

그동안 회원 가입, 로그인, 피드와 코멘트에는 별 무리 없이 사용해 왔는데

 

팔로우 기능을 구현하려고 하니 문제가 생겼다.

 

인식한 김에 코드를 살짝 뒤집어엎었고, 그렇게 재구성한 유저 모델이 아래와 같다.

from sqlalchemy import Column, String, Integer, Sequence
from sqlalchemy.orm import relationship
from config.db import Base
from pydantic import BaseModel, validator
import re


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    email = Column(String, 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)
    likes = relationship("Like", back_populates="user")


class UserCreate(BaseModel):
    email: str
    password: str
    nickname: str


class UserInDB(UserCreate):
    pass


class UserLogin(BaseModel):
    email: str
    password: str

    @validator("email")
    def validate_email(cls, value):
        if not re.match(r"[^@]+@[^@]+\.[^@]+", value):
            raise ValueError("Invalid Email Format")
        return value


class UserResponse(BaseModel):
    email: str
    nickname: str

이외에도 작은 부분에서 여기저기 많은 수정이 있었는데,

 

이걸 다 올리는 건 생산적이지 않은 것 같아 구현을 모아놓은 레포지토리 주소로 대체한다.

 

https://github.com/gnidinger/FastAPI_SQLAlchemy_Pydantic

GitHub - gnidinger/FastAPI_SQLAlchemy_Pydantic

이곳에 가면 그간 구현했던 모든 코드가 별개의 브랜치로 잘 정리되어 있다.

 

당연히 07.like_follow를 보면 된다.

 

Follow

 

follow.py

 

이제 진짜로 시작해 보자. 팔로우와 라이크 기능 모두 각각의 토글 버튼으로 해결할 생각이다.

 

그러기 위해서 팔로우 모델은 아주 단순하게만 만들면 된다.

from sqlalchemy import Column, Integer, ForeignKey, UniqueConstraint
from config.db import Base


class Follow(Base):
    __tablename__ = "follows"

    follower_id = Column(Integer, ForeignKey("users.id"), primary_key=True)
    following_id = Column(Integer, ForeignKey("users.id"), primary_key=True)

    # 두 필드의 조합이 고유해야 함
    __table_args__ = (UniqueConstraint("follower_id", "following_id", name="unique_follow"),)

Pydantic모델은 필요하지 않다.

 

그 이유는 너무나 당연하게도, Pydantic 모델 자체가 데이터 검증 및 직렬화/역직렬화를 위한 라이브러리이기 때문이다.

 

팔로우는 데이터 유효성 검사나 직렬화/역직렬화를 사용할 일이 없는 간단한 기능이라는 뜻이다.

 

어쨌거나 먼저 두 칼럼을 모두 정수타입이자 유저 테이블의 id를 참조하는 외래키이자 follows테이블의 기본 키로 설정한다.

 

이렇게 두 칼럼을 모두 기본키로 설정함으로써 각 행은 고유한 팔로워와 팔로잉의 조합을 나타낼 수 있다.

 

그리고 마지막 줄이 중요한데,

    # 두 필드의 조합이 고유해야 함
    __table_args__ = (UniqueConstraint("follower_id", "following_id", name="unique_follow"),)

주석에 달아놓은 바와 같이 특정 칼럼의 조합이 유일하게 보장되어야 한다는 제약 조건이다.

 

'__table_args__'는 이와 같은 옵션을 설정하기 위해 SQLAlchemy에서 사용된다.

 

위와 같은 조건 때문에 특정 사용자가 동일한 사람을 여러 번 팔로우할 수 없게 된다.

 

follow_service.py

 

계속해서 서비스 로직을 보자. 위에서 예고했듯 하나의 함수를 이용해 토글버튼처럼 사용한다.

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from models.follow import Follow
from services import auth_service
from fastapi import HTTPException


async def toggle_follow(db: AsyncSession, follower_id: int, following_id: int):
    if follower_id == following_id:
        raise HTTPException(status_code=400, detail="You cannot follow yourself")

    following = await auth_service.get_user_by_id(db, following_id)

    if not following:
        raise HTTPException(status_code=404, detail="User not found")

    result = await db.execute(
        select(Follow).where(Follow.follower_id == follower_id, Follow.following_id == following_id)
    )
    existing_follow = result.scalar_one_or_none()

    if existing_follow:
        await db.delete(existing_follow)
        action = "unfollowed"
    else:
        new_follow = Follow(follower_id=follower_id, following_id=following_id)
        db.add(new_follow)
        action = "followed"

    await db.commit()
    return {"action": action}

지난 글에서 구현했던 비동기 처리를 계속 이어간다.

 

스스로를 팔로우하는 것을 막고, 대상 사용자가 없는 경우에 대해 예외처리를 해준다.

 

이어서 2.0 스타일 쿼리를 이용해 요청을 보낸 사용자가 해당 사용자를 팔로우하고 있는지,

 

즐 팔로우 객체가 테이블에 있는지 확인한 뒤에 언팔로우/팔로우 여부를 결정한다.

 

사실 이 부분이 존재하기 때문에 모델에서 제약조건을 걸어줄 필요는 없다고 볼 수도 있다.

 

마지막으로 디비에 커밋해 주고, 방금 이루어진 액션이 어느 것인지를 리턴으로 보내준다.

 

follow_router.py

 

다음으로 라우팅 함수이다.

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.future import select
from sqlalchemy.ext.asyncio import AsyncSession
from services.follow_service import toggle_follow
from models.user import User
from services import auth_service
from config.db import get_db

router = APIRouter()


@router.patch("/{following_id}")
async def toggle_follow_route(
    following_id: int,
    email: str = Depends(auth_service.get_current_user_authorization),
    db: AsyncSession = Depends(get_db),
):
    # 이메일을 이용해 사용자 정보를 가져옴
    result = await db.execute(select(User).filter_by(email=email))
    current_user = result.scalar_one_or_none()

    # 사용자 정보가 없으면 에러 처리
    if current_user is None:
        raise HTTPException(status_code=404, detail="User not found")

    current_user_id = current_user.id

    return await toggle_follow(db, current_user_id, following_id)

import 부분은 건너뛰고, 인증 정보를 가져온 이메일을 통해 현재 사용자의 아이디를 가져온다.

 

앞서 언급했듯 이 부분에서 뭔가 잘못되었다는 것을 깨닫고 코드를 뒤집었다.

 

어쨌거나 아이디를 가져와 비동기로 토글 요청을 보낸다. 간단한 로직이다.

 

이어서 좋아요 부분의 구현을 살펴보자.

 

Like

 

like.py

 

좋아요는 어떤 식으로 구현을 할까 고민을 했다.

 

피드 좋아요와 코멘트 좋아요를 분리해서 가능한 빈 값을 제거할까 하다가

 

그냥 하나로 합치고 빈 값을 허용하기로 했다.

 

이렇게 고민해서 나온 결과가 아래 구현이다.

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


class Like(Base):
    __tablename__ = "likes"

    id = Column(Integer, primary_key=True, index=True)
    user_email = Column(String, ForeignKey("users.email"), index=True)
    feed_id = Column(Integer, ForeignKey("feeds.id"), index=True, nullable=True)
    comment_id = Column(Integer, ForeignKey("comments.id"), index=True, nullable=True)

    user = relationship("User", back_populates="likes")
    feed = relationship("Feed", back_populates="likes")
    comment = relationship("Comment", back_populates="likes")

역시 Pydantic 모델은 필요하지 않다.

 

자동 증가하는 기본키를 별개로 설정하고, 유저, 피드, 코멘트의 값을 참조할 외래키를 설정한다.

 

피드 아이디와 코멘트 아이디는 위에서 이야기했듯 null 값을 허용한다.

 

이어서 역시 유저, 피드, 코멘트와 관계를 맺어주면 끝이다.

 

like_service.py

 

계속해서 비즈니스 로직이다.

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from fastapi import HTTPException
from models.like import Like
import logging

logging.basicConfig(level=logging.DEBUG)


async def toggle_like(
    db: AsyncSession, user_email: str, feed_id: int = None, comment_id: int = None
):
    # feed_id와 comment_id 둘 다 없거나 둘 다 있을 경우 에러
    if (feed_id is None and comment_id is None) or (feed_id is not None and comment_id is not None):
        raise HTTPException(
            status_code=400, detail="Either feed_id or comment_id must be provided."
        )

    # 이미 좋아요가 있는지 찾기
    query = select(Like).where(
        (Like.user_email == user_email)
        & (Like.feed_id == feed_id)
        & (Like.comment_id == comment_id)
    )
    result = await db.execute(query)
    existing_like = result.scalar_one_or_none()

    # 좋아요가 이미 있다면 삭제, 없다면 추가
    if existing_like:
        await db.delete(existing_like)
    else:
        new_like = Like(user_email=user_email, feed_id=feed_id, comment_id=comment_id)
        db.add(new_like)

    await db.commit()

피드 아이디와 코멘트 아이디로 좋아요 여부를 판단할 것이기 때문에

 

요청에 둘 다 포함된다면 예외를 발생시킨다.

 

이후의 로직은 팔로우와 비슷하게 이어지므로 생략한다.

 

like_router.py

 

다음은 라우팅 함수다.

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from config.db import get_db
from services import auth_service
from services.like_service import toggle_like

router = APIRouter()


@router.patch("/")
async def toggle(
    user_email: str = Depends(auth_service.get_current_user_authorization),
    feed_id: int = None,
    comment_id: int = None,
    db: AsyncSession = Depends(get_db),
):
    try:
        await toggle_like(db, user_email, feed_id=feed_id, comment_id=comment_id)
    except HTTPException as e:
        raise e

    return {"message": "Like toggled successfully"}

메서드를 patch로 한 것은 팔로우와 똑같지만,

 

하나의 라우터에서 두 개의 좋아요를 처리하기 위해 PathVariable이 아닌 입력값으로

 

대상의 아이디를 받도록 구현했다.

 

해놓고 보니 이게 베스트 초이스였는지는 확실하지 않다..

 

어쨌거나 처음 의도에 맞게 잘 구현이 되었다.

 

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 models.comment import Base as CommentBase  # Comment의 Base
from models.like import Base as LikeBase  # Like의 Base 클래스
from models.follow import Base as FollowBase  # Follow의 Base 클래스
from routers import auth_router, feed_router, comment_router, like_router, follow_router
from config.cors_config import setup_cors

DATABASE_URL = "postgresql://postgres:rjslrjsl333@localhost:5432/postgres"
engine = create_engine(DATABASE_URL)

UserBase.metadata.create_all(bind=engine)
FeedBase.metadata.create_all(bind=engine)
CommentBase.metadata.create_all(bind=engine)
LikeBase.metadata.create_all(bind=engine)
FollowBase.metadata.create_all(bind=engine)

app = FastAPI()

setup_cors(app)

app.include_router(auth_router.router, prefix="/api/auth", tags=["auth"])
app.include_router(feed_router.router, prefix="/api/feed", tags=["feed"])
app.include_router(comment_router.router, prefix="/api/comment", tags=["comment"])
app.include_router(like_router.router, prefix="/api/like", tags=["like"])
app.include_router(follow_router.router, prefix="/api/follow", tags=["follow"])


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

이렇게 해서 간단하게 팔로우와 좋아요 구현이 끝났다.

 

실은 유저 테이블과 해당되는 로직들을 고치느라 나 개인적으로는 시간이 많이 들었지만..

 

어찌어찌해놓고 나니 뿌듯하기는 하다.

 

다음은 무슨 구현을 이어가 볼까,

 

일단은 끝!