[FastAPI] 객체지향과 FastAPI (16)

 

 

FastAPI에서 객체지향 개념을 적용해 보자.

 

1. FastAPI에서 객체지향 적용의 필요성

캡슐화(Encapsulation), 상속(Inheritance), 다형성(Polymorphism)

• 코드가 커지다 보면, 비슷한 기능을 하는 클래스를 여러개 만들 때가 많다. 이럴때 추상클래스와 상속, 다형성 개념을 적용하면 코드를 깔끔하게 정리할 수 있다.
• 예를 들어 삭제 동작을 정의한다고 할 때, 프로젝트의 다양한 곳에서 delete, remove, erase, eliminate 등 다양한 이름으로 사용된다면 상당히 관리가 어려울 것이기 때문이다. 이럴 때, 추상클래스를 상속해 사용한다면 코드 관리가 아주 편리해 질 수 있을 것이다.

 

의존성 주입을 통한 결합도 낮추기

• fastapi의 Depends를 활용한 객체 주입할 수 있다.
• 이는 서비스 객체를 주입하여 비즈니스 로직 분리한다.

 

---

2. 서비스 레이어(Service Layer)

설명

• 서비스 클래스는 비즈니스 로직을 캡슐화하는 역할
• 라우터 부분은 FastAPI 관련 코드로 구성하고, 비즈니스 로직은 서비스로 분리
• 데이터베이스 접근은 리포지토리(repository)에서 처리하고, 서비스에서는 비즈니스 로직만 담당

장점

• 엔드포인트를 깔끔하게 유지
• 서비스 레이어를 독립적으로 테스트 가능

 

코드 설명

• Depends 를 이용해 서비스(Service) 레이어에서 정의한 클래스를 가져온다.
• 주입할 객체에 필요한 인자도 넣어줄 수 있다.

router = APIRouter()

def get_user_service(db: Session = Depends(get_db)):    
    return UserService(db)

@router.post("/users/")
def create_user(
    user: UserCreate,
    user_service: UserService = Depends(get_user_service)
):
    """유저 생성 시 예외 처리"""
    try:
        new_user = user_service.create_user(user)
        return new_user
    except Exception as e:  
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"User creation failed: {str(e)}"
        )

 

---

3. 리포지토리 레이어(Repository Layer)

리포지토리 인터페이스(추상 클래스)를 만들어서 리포지토리의 표준을 정의한다.

표준이 정해졌으니, 여러명의 개발자가 함께 개발을 한다고 해도

코드가 깔끔하게 정리될 것이다.

from abc import ABC, abstractmethod
from typing import Generic, TypeVar, List, Optional
from sqlalchemy.orm import Session

T = TypeVar("T")  # 제네릭 타입 변수 (모든 엔터티에 사용 가능)

class IRepository(ABC, Generic[T]):
    """모든 리포지토리에서 공통적으로 사용할 인터페이스"""
    
    @abstractmethod
    def create(self, db: Session, obj: T) -> T:
        pass

    @abstractmethod
    def get_by_id(self, db: Session, obj_id: int) -> Optional[T]:
        pass

    @abstractmethod
    def get_all(self, db: Session) -> List[T]:
        pass

    @abstractmethod
    def update(self, db: Session, obj_id: int, obj: T) -> Optional[T]:
        pass

    @abstractmethod
    def delete(self, db: Session, obj_id: int) -> bool:
        pass

 

레포지토리 상속받아 구현

class UserRepository(IRepository[User]):
    """User 엔터티에 특화된 리포지토리"""

    def create(self, db: Session, obj: UserCreate) -> User:
        new_user = User(name=obj.name, email=obj.email)
        db.add(new_user)
        db.commit()
        db.refresh(new_user)
        return new_user

    def get_by_id(self, db: Session, obj_id: int) -> Optional[User]:
        return db.query(User).filter(User.id == obj_id).first()

    def get_all(self, db: Session) -> List[User]:
        return db.query(User).all()

    def update(self, db: Session, obj_id: int, obj: UserCreate) -> Optional[User]:
        user = db.query(User).filter(User.id == obj_id).first()
        if not user:
            return None
        user.name = obj.name
        user.email = obj.email
        db.commit()
        db.refresh(user)
        return user

    def delete(self, db: Session, obj_id: int) -> bool:
        user = db.query(User).filter(User.id == obj_id).first()
        if not user:
            return False
        db.delete(user)
        db.commit()
        return True

 

서비스 레이어에서 레포지토리 사용

class UserService:
    def __init__(self, db: Session, repository: IRepository[User]):
        self.db = db
        self.repository = repository  # 상위 인터페이스에 의존

    def create_user(self, user: UserCreate):
        return self.repository.create(self.db, user)

    def get_user(self, user_id: int):
        return self.repository.get_by_id(self.db, user_id)

    def get_all_users(self):
        return self.repository.get_all(self.db)

    def update_user(self, user_id: int, user: UserCreate):
        return self.repository.update(self.db, user_id, user)

    def delete_user(self, user_id: int):
        return self.repository.delete(self.db, user_id)

 

---

 

4. 싱글톤 패턴(Singleton Pattern) 활용 in FastAPI

싱글톤 패턴은 객체의 인스턴스를 하나만 생성하여 재사용하는 디자인 패턴

다음에 싱글톤 패턴을 적용하면 성능 최적화와 리소스 절약을 할 수 있다

 

단, 다음과 같은 경우에는 싱글톤 패턴을 쓰지 않도록 한다.

• DB 세션을 직접 가지고 있음
• 데이터가 오염될 가능성 있음
• 자원을 많이 가지는 경우

 

FastAPI의 lifespan을 이용한 싱글톤 적용

from fastapi import FastAPI
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    db = next(get_db())  # DB 세션을 미리 생성
    user_service = UserService(db)  # 싱글톤 객체 생성
    app.state.user_service = user_service  # FastAPI의 state에 저장
    yield
    db.close()  # 앱 종료 시 DB 세션 닫기

app = FastAPI(lifespan=lifespan)

@router.post("/users/")
def create_user(user: UserCreate):
    user_service = app.state.user_service  # 싱글톤 객체 가져오기
    return user_service.create_user(user)

 

---

5. 기타

pydantic 모델을 DTO로 활용

sqlalchemy 모델

설정(config) 객체를 싱글톤으로 관리

 

등등...