python+fastAPI+postgresql 로 구성된 환경에서 간단한 CRUD 구현2

2025.03.28 - [reference page/Python] - python+fastAPI+postgresql 로 구성된 환경에서 간단한 CRUD 구현1

 

지난번 글을 이어서 기능적인 코드를 남겨보려고 합니다.

보통 각 역할에 따라 폴더를 생성해서 관리를 하게 되는데..

먼저 RESTful API 엔드포인트를 관리하는 controller 에 있는 파일입니다.

 

user_controller.py

"""
사용자 관련 API 엔드포인트 정의
이 모듈은 사용자 관리를 위한 RESTful API 엔드포인트들을 정의합니다.
"""

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from typing import List, Dict
from database.database import get_db
from service.user_service import UserService
from service.login_info_service import LoginInfoService
from schemas.user import UserTest, UserTestCreate, UserTestUpdate
from schemas.login_info import LoginInfoCreate, LoginInfo

# API 라우터 설정
router = APIRouter(
    prefix="/users",  # URL 접두사
    tags=["users"]    # Swagger 문서 태그
)

@router.post("/", response_model=UserTest)
def create_user(user: UserTestCreate, db: Session = Depends(get_db)):
    """
    새로운 사용자를 생성합니다.
    
    Args:
        user (UserTestCreate): 생성할 사용자 정보
        db (Session): 데이터베이스 세션
    
    Returns:
        UserTest: 생성된 사용자 정보
    
    Raises:
        HTTPException: 이메일이 이미 등록된 경우 400 에러
    """
    return UserService.create_user(db=db, user=user)

@router.get("/", response_model=List[UserTest])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    """
    사용자 목록을 조회합니다.
    
    Args:
        skip (int): 건너뛸 레코드 수 (기본값: 0)
        limit (int): 조회할 최대 레코드 수 (기본값: 100)
        db (Session): 데이터베이스 세션
    
    Returns:
        List[UserTest]: 사용자 목록
    """
    return UserService.get_users(db=db, skip=skip, limit=limit)

@router.get("/{user_id}", response_model=UserTest)
def read_user(user_id: int, db: Session = Depends(get_db)):
    """
    특정 사용자의 정보를 조회합니다.
    
    Args:
        user_id (int): 조회할 사용자의 ID
        db (Session): 데이터베이스 세션
    
    Returns:
        UserTest: 사용자 정보
    
    Raises:
        HTTPException: 사용자를 찾을 수 없는 경우 404 에러
    """
    return UserService.get_user_by_id(db=db, user_id=user_id)

@router.put("/{user_id}", response_model=UserTest)
def update_user(user_id: int, user: UserTestUpdate, db: Session = Depends(get_db)):
    """
    사용자 정보를 수정합니다.
    
    Args:
        user_id (int): 수정할 사용자의 ID
        user (UserTestUpdate): 수정할 정보
        db (Session): 데이터베이스 세션
    
    Returns:
        UserTest: 수정된 사용자 정보
    
    Raises:
        HTTPException: 사용자를 찾을 수 없는 경우 404 에러
    """
    return UserService.update_user(db=db, user_id=user_id, user=user)

@router.delete("/{user_id}")
def delete_user(user_id: int, db: Session = Depends(get_db)):
    """
    사용자를 삭제합니다.
    
    Args:
        user_id (int): 삭제할 사용자의 ID
        db (Session): 데이터베이스 세션
    
    Returns:
        dict: 삭제 성공 메시지
    
    Raises:
        HTTPException: 사용자를 찾을 수 없는 경우 404 에러
    """
    return UserService.delete_user(db=db, user_id=user_id)

 

여기에는 기본적인 CRUD 할수 있는 rest api 를 구성해둡니다.

API 라우터를 정의하고 정의한 라우터로 get/post/put/delete 로 호출해서 사용하면 됩니다.

 

 

그리고 이 기능을 실행하는 service 에는

user_service.py

"""
사용자 관련 비즈니스 로직
이 모듈은 사용자 관리에 필요한 모든 비즈니스 로직을 구현합니다.
"""

from sqlalchemy.orm import Session
from fastapi import HTTPException
from model.user import UserTest
from schemas.user import UserTestCreate, UserTestUpdate
from typing import List, Optional

class UserService:
    """
    사용자 관리 서비스
    데이터베이스 작업과 비즈니스 로직을 처리합니다.
    """
    
    @staticmethod
    def create_user(db: Session, user: UserTestCreate) -> UserTest:
        """
        새로운 사용자를 생성합니다.
        
        Args:
            db (Session): 데이터베이스 세션
            user (UserTestCreate): 생성할 사용자 정보
        
        Returns:
            UserTest: 생성된 사용자 객체
        
        Raises:
            HTTPException: 이메일이 이미 등록된 경우 발생
        """
        # 이메일 중복 검사
        db_user = db.query(UserTest).filter(UserTest.email == user.email).first()
        if db_user:
            raise HTTPException(status_code=400, detail="이미 등록된 이메일입니다")
        
        # 새 사용자 생성
        db_user = UserTest(**user.model_dump())
        db.add(db_user)
        db.commit()
        db.refresh(db_user)
        return db_user

    @staticmethod
    def get_users(db: Session, skip: int = 0, limit: int = 100) -> List[UserTest]:
        """
        사용자 목록을 조회합니다.
        
        Args:
            db (Session): 데이터베이스 세션
            skip (int): 건너뛸 레코드 수
            limit (int): 조회할 최대 레코드 수
        
        Returns:
            List[UserTest]: 사용자 목록
        """
        return db.query(UserTest).offset(skip).limit(limit).all()

    @staticmethod
    def get_user_by_id(db: Session, user_id: int) -> Optional[UserTest]:
        """
        특정 사용자의 정보를 조회합니다.
        
        Args:
            db (Session): 데이터베이스 세션
            user_id (int): 조회할 사용자의 ID
        
        Returns:
            UserTest: 사용자 객체
        
        Raises:
            HTTPException: 사용자를 찾을 수 없는 경우 발생
        """
        db_user = db.query(UserTest).filter(UserTest.id == user_id).first()
        if db_user is None:
            raise HTTPException(status_code=404, detail="사용자를 찾을 수 없습니다")
        return db_user

    @staticmethod
    def get_user_by_email(db: Session, email: str) -> Optional[UserTest]:
        """
        이메일로 사용자를 조회합니다.
        
        Args:
            db (Session): 데이터베이스 세션
            email (str): 조회할 사용자의 이메일
            
        Returns:
            Optional[UserTest]: 찾은 사용자 객체 또는 None
        """
        return db.query(UserTest).filter(UserTest.email == email).first()

    @staticmethod
    def update_user(db: Session, user_id: int, user: UserTestUpdate) -> UserTest:
        """
        사용자 정보를 수정합니다.
        
        Args:
            db (Session): 데이터베이스 세션
            user_id (int): 수정할 사용자의 ID
            user (UserTestUpdate): 수정할 정보
        
        Returns:
            UserTest: 수정된 사용자 객체
        
        Raises:
            HTTPException: 사용자를 찾을 수 없는 경우 발생
        """
        # 사용자 존재 확인
        db_user = db.query(UserTest).filter(UserTest.id == user_id).first()
        if db_user is None:
            raise HTTPException(status_code=404, detail="사용자를 찾을 수 없습니다")
        
        # 제공된 필드만 업데이트
        update_data = user.model_dump(exclude_unset=True)
        for field, value in update_data.items():
            setattr(db_user, field, value)
        
        db.commit()
        db.refresh(db_user)
        return db_user

    @staticmethod
    def delete_user(db: Session, user_id: int) -> dict:
        """
        사용자를 삭제합니다.
        
        Args:
            db (Session): 데이터베이스 세션
            user_id (int): 삭제할 사용자의 ID
        
        Returns:
            dict: 삭제 성공 메시지
        
        Raises:
            HTTPException: 사용자를 찾을 수 없는 경우 발생
        """
        # 사용자 존재 확인
        db_user = db.query(UserTest).filter(UserTest.id == user_id).first()
        if db_user is None:
            raise HTTPException(status_code=404, detail="사용자를 찾을 수 없습니다")
        
        # 사용자 삭제
        db.delete(db_user)
        db.commit()
        return {"message": "사용자가 성공적으로 삭제되었습니다"}

 

이 코드를 통해서 crud 하는 방법을 익혀나가면 될거 같습니다.

 

마지막 파일로는 controller 이나 service 에서 import 해서 사용하는 schemas 입니다.

user.py

"""
사용자 관련 Pydantic 모델 정의
이 모듈은 API 요청/응답에 사용되는 데이터 검증 및 직렬화 스키마를 정의합니다.
"""

from pydantic import BaseModel
from typing import Optional

class UserTestBase(BaseModel):
    """
    사용자 기본 정보 스키마
    모든 사용자 관련 스키마의 기본이 되는 클래스입니다.
    
    Attributes:
        username (str): 사용자 이름
        email (str): 이메일 주소
    """
    username: str
    email: str

class UserTestCreate(UserTestBase):
    """
    사용자 생성 요청 스키마
    UserTestBase를 상속받아 사용자 생성에 필요한 필드만 정의합니다.
    """
    pass

class UserTestUpdate(BaseModel):
    """
    사용자 정보 수정 요청 스키마
    모든 필드가 선택적(Optional)이어서 부분 업데이트가 가능합니다.
    
    Attributes:
        username (Optional[str]): 변경할 사용자 이름 (선택)
        email (Optional[str]): 변경할 이메일 주소 (선택)
    """
    username: Optional[str] = None
    email: Optional[str] = None

class UserTest(UserTestBase):
    """
    사용자 정보 응답 스키마
    데이터베이스에서 조회한 사용자 정보를 반환할 때 사용됩니다.
    
    Attributes:
        id (int): 사용자 고유 식별자
        created_at (str): 계정 생성 시간 ('yyyymmddhhmmss' 형식)
        updated_at (str): 정보 수정 시간 ('yyyymmddhhmmss' 형식)
    """
    id: int
    created_at: str
    updated_at: str

    class Config:
        """Pydantic 설정"""
        from_attributes = True  # ORM 모델을 Pydantic 모델로 변환 허용

 

 

이상으로 fastAPI 를 이용해서 CRUD 하는 방법을 코드를 통해서 봤는데요..

아직은 익숙하지 않지만, 계속 코드를 작성해보면 코드가 어렵지 않겠다는 생각을 해봅니다.