[Python] ORM을 활용하여 fastAPI 기반 게시판 구현해보기

JaehyoJJAng2024년 06월 06일

이전 게시글(ORM 활용해보기)와 연결되는 내용입니다.

디렉토리 구조

해당 프로젝트의 디렉토리 구조는 다음과 같다.

tree -L 2 .

.
├── DB
│   └── database.py
├── Models
│   └── model.py
├── Validation
│   └── pydantic_models.py
├── crud.py
└── main.py

1. Model 정의

Models/model.py

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, DateTime
from sqlalchemy.orm import relationship
from datetime import datetime

Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True)
    name = Column(String(30), nullable=False)
    fullname = Column(String(40), nullable=False)
    nickname = Column(String(20))
    
    posts = relationship("Post", back_populates="author")
    comments = relationship("Comment", back_populates="author")
    
    def __repr__(self) -> str:
        return f"<User(name='{self.name}', fullname='{self.fullname}', nickname='{self.nickname}')"


class Post(Base):
    __tablename__ = "posts"
    
    id = Column(Integer, primary_key=True)
    title = Column(String(40), nullable=False)
    content = Column(String(1000), nullable=False)
    user_id = Column(Integer, ForeignKey('users.id'))
    created_at = Column(DateTime, default=datetime.now())
    
    author = relationship("User", back_populates="posts")
    comments = relationship("Comment", back_populates="post")
    
    def __repr__(self):
        return f"<Post(title='{self.title}', content='{self.content}')>"

class Comment(Base):
    __tablename__ = 'comments'

    id = Column(Integer, primary_key=True)
    content = Column(String(150), nullable=False)
    user_id = Column(Integer, ForeignKey('users.id'))
    post_id = Column(Integer, ForeignKey('posts.id'))
    created_at = Column(DateTime, default=datetime.now())

    author = relationship("User", back_populates="comments")
    post = relationship("Post", back_populates="comments")

    def __repr__(self):
        return f"<Comment(content='{self.content}')>"

back_populates="author"
- Post 모델의 author 속성을 통해 양방향 관계를 설정함
- 이를 통해 User와 Post 간의 양방향 참조가 가능

relationship 함수에 들어가는 인자 설명

back_populates
- 양방향 관계를 정의합니다. 한쪽 모델에서 back_populates로 정의된 속성은 반대쪽 모델에서도 동일하게 설정되어야 합니다.
- 예를 들어, User 모델의 posts와 Post 모델의 author 속성이 서로를 참조하도록 설정할 수 있습니다.
cascade
- 관계된 객체에 대한 일괄 작업을 정의합니다.
- 예를 들어, 부모 객체가 삭제될 때 자식 객체도 함께 삭제되도록 설정할 수 있습니다.
- 사용 예: cascade="all, delete-orphan"
lazy
- 관계된 객체를 언제 로드할지를 설정합니다.
- select: 기본값으로, 관계된 객체를 사용할 때 쿼리를 실행하여 로드합니다.
- joined: 부모 객체를 로드할 때 조인을 사용하여 관계된 객체를 함께 로드합니다.
- subquery: 부모 객체를 로드할 때 서브쿼리를 사용하여 관계된 객체를 함께 로드합니다.
- dynamic: 쿼리를 반환하여 나중에 필터링하거나 실행할 수 있게 합니다.
primaryjoin
- 관계를 정의하는 두 테이블 간의 조인 조건을 명시적으로 설정합니다.
- 기본적으로 ForeignKey를 기반으로 자동 설정되지만, 복잡한 조건이 필요할 경우 명시적으로 지정할 수 있습니다.

2. 데이터베이스 설정

DB/database.py

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from Models.model import Base

engine = create_engine(url="mysql+pymysql://orm:orm@192.168.219.113/orm")
SessionLocal = sessionmaker(bind=engine)

# 데이터베이스 테이블 생성
def init_db() -> None:
    Base.metadata.create_all(bind=engine)

3. Pydantic 정의

Validation/pydantic_models.py

from pydantic import BaseModel, field_validator

class UserModel(BaseModel):
    name: str
    fullname: str
    nickname: str | None
    
    @field_validator('name', 'fullname')
    def name_must_not_be_empty(cls, value: str) -> str:
        if not value or value.strip() == '':
            raise ValueError("name and fullname must not be empty")
        return value

class PostModel(BaseModel):
    title: str
    content: str
    user_id: int
    
    @field_validator('title', 'content')
    def title_and_content_must_not_be_empty(cls, value: str) -> str:
        if not value or value.strip() == '':
            raise ValueError('Title and content must not be empty')
        return value

class CommentModel(BaseModel):
    content: str
    user_id: int
    post_id: int

    @field_validator('content')
    def content_must_not_be_empty(cls, value: str) -> str:
        if not value or value.strip() == '':
            raise ValueError('content must not be empty')
        return value

4. 데이터 삽입 및 조회 로직 구현

crud.py

from sqlalchemy.orm import Session
from Models.model import User, Post, Comment
from Validation.pydantic_models import UserModel, PostModel, CommentModel

# Create Functions
def create_user(db: Session, user_data: dict) -> User:
    user = UserModel(**user_data)
    db_user = User(name=user.name, fullname=user.fullname, nickname=user.nickname)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

def create_post(db: Session, post_data: dict) -> Post:
    post = PostModel(**post_data)
    db_post = Post(title=post.title, content=post.content, user_id=post.user_id)
    db.add(db_post)
    db.commit()
    db.refresh(db_post)
    return db_post

def create_comment(db: Session, comment_data: dict) -> Comment:
    comment = CommentModel(**comment_data)
    db_comment = Comment(content=comment.content, user_id=comment.user_id, post_id=comment.post_id)
    db.add(db_comment)
    db.commit()
    db.refresh(db_comment)
    return db_comment


# Read Functions
def get_user_by_id(db: Session, user_id: int) -> User:
    return db.query(User).filter(User.id == user_id).first()

def get_users(db: Session, skip: int = 0, limit: int = 10) -> list[User]:
    return db.query(User).offset(skip).limit(limit=limit).all()

def get_post_by_id(db: Session, post_id: int) -> Post:
    return db.query(Post).filter(Post.id == post_id).first()

def get_posts(db: Session, skip: int = 0, limit: int = 10) -> list[Post]:
    return db.query(Post).offset(skip).limit(limit).all()

def get_comment_by_id(db: Session, comment_id: int) -> Comment:
    return db.query(Comment).filter(Comment.id == comment_id).first()

def get_comments(db: Session, skip: int = 0, limit: int = 10) -> list[Comment]:
    return db.query(Comment).offset(skip).limit(limit).all()

# Update functions
def update_user(db: Session, user_id: int, user_data: dict):
    db_user = get_user_by_id(db, user_id)
    if not db_user:
        return None
    for key, value in user_data.items():
        setattr(db_user, key, value)
    db.commit()
    db.refresh(db_user)
    return db_user

def update_post(db: Session, post_id: int, post_data: dict):
    db_post = get_post_by_id(db, post_id)
    if not db_post:
        return None
    for key, value in post_data.items():
        setattr(db_post, key, value)
    db.commit()
    db.refresh(db_post)
    return db_post

def update_comment(db: Session, comment_id: int, comment_data: dict):
    db_comment = get_comment_by_id(db, comment_id)
    if not db_comment:
        return None
    for key, value in comment_data.items():
        setattr(db_comment, key, value)
    db.commit()
    db.refresh(db_comment)
    return db_comment

# Delete Functions
def delete_user(db: Session, user_id: int):
    db_user = get_user_by_id(db, user_id)
    if not db_user:
        return None
    db.delete(db_user)
    db.commit()
    return db_user

def delete_post(db: Session, post_id: int):
    db_post = get_post_by_id(db, post_id)
    if not db_post:
        return None
    db.delete(db_post)
    db.commit()
    return db_post

def delete_comment(db: Session, comment_id: int):
    db_comment = get_comment_by_id(db, comment_id)
    if not db_comment:
        return None
    db.delete(db_comment)
    db.commit()
    return db_comment

5. 메인 애플리케이션 로직

main.py

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from database import SessionLocal, init_db
import crud
import models
import pydantic_models

# FastAPI 인스턴스 생성
app = FastAPI()

# 데이터베이스 초기화
init_db()

# 의존성 주입
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# 사용자 관련 엔드포인트
@app.post("/users/", response_model=pydantic_models.UserModel)
def create_user(user: pydantic_models.UserModel, db: Session = Depends(get_db)):
    return crud.create_user(db, user.dict())

@app.get("/users/{user_id}", response_model=pydantic_models.UserModel)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_id(db, user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

@app.get("/users/", response_model=list[pydantic_models.UserModel])
def read_users(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    return crud.get_users(db, skip=skip, limit=limit)

# 게시물 관련 엔드포인트
@app.post("/posts/", response_model=pydantic_models.PostModel)
def create_post(post: pydantic_models.PostModel, db: Session = Depends(get_db)):
    return crud.create_post(db, post.dict())

@app.get("/posts/{post_id}", response_model=pydantic_models.PostModel)
def read_post(post_id: int, db: Session = Depends(get_db)):
    db_post = crud.get_post_by_id(db, post_id)
    if db_post is None:
        raise HTTPException(status_code=404, detail="Post not found")
    return db_post

@app.get("/posts/", response_model=list[pydantic_models.PostModel])
def read_posts(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    return crud.get_posts(db, skip=skip, limit=limit)

# 댓글 관련 엔드포인트
@app.post("/comments/", response_model=pydantic_models.CommentModel)
def create_comment(comment: pydantic_models.CommentModel, db: Session = Depends(get_db)):
    return crud.create_comment(db, comment.dict())

@app.get("/comments/{comment_id}", response_model=pydantic_models.CommentModel)
def read_comment(comment_id: int, db: Session = Depends(get_db)):
    db_comment = crud.get_comment_by_id(db, comment_id)
    if db_comment is None:
        raise HTTPException(status_code=404, detail="Comment not found")
    return db_comment

@app.get("/comments/", response_model=list[pydantic_models.CommentModel])
def read_comments(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    return crud.get_comments(db, skip=skip, limit=limit)

마무리

유저 목록 조회해보기

curl 192.168.219.113:8000/users/1

# 출력
{"name":"jaehyo","fullname":"Jaehyo Lee","nickname":"jaehyojjang"}

python