FastAPI를 이용한 멀티테넌트 동적 모듈 로딩 시스템 구현

FastAPI를 이용한 멀티테넌트 동적 모듈 로딩 시스템 구현

• 소개
• 시스템 구조
  • 0. 디렉토리 구조
  • 1. 컨텍스트 관리
  • 2. 동적 모듈 로더
  • 3. 기본 모델과 커스텀 모델
    • 기본 Goods 모델:
    • CustomerA용 커스텀 Goods 모델:
  • 4. API 구현
  • 5. 애플리케이션 설정
• 작동 방식
• 장점
• 주의사항

소개

FastAPI를 사용하여 고객별로 다른 모델과 스키마를 동적으로 로딩하는 멀티테넌트 시스템을 구현하는 방법을 설명하겠습니다.

시스템 구조

0. 디렉토리 구조

app/
├── api/
│   └── v1/
│       └── goods.py
├── core/
│   └── db.py
├── models/
│   ├── customerA/
│   │   └── goods.py
│   └── goods.py
├── schemas/
│   ├── customerA/
│   │   └── goods.py
│   └── goods.py
└── utils/
    ├── context.py
    └── custom_importer.py
main.py

1. 컨텍스트 관리

먼저 현재 요청의 고객 ID를 관리하기 위한 컨텍스트 시스템이 필요합니다.

app/utils/context.py

import contextvars
from typing import Optional

_customer_context = contextvars.ContextVar('customer_id', default=None)

def set_customer_context(customer_id: Optional[str]) -> None:
    _customer_context.set(customer_id)

def get_customer_context() -> Optional[str]:
    return _customer_context.get()

contextvars를 사용하여 각 요청별로 고객 ID를 격리된 방식으로 저리합니다.

2. 동적 모듈 로더

고객별 커스텀 모듈을 동적으로 로드하는 핵심 기능입니다.

app/utils/custom_importer.py

def get_all_modules(package_name: str) -> Set[str]:
    """지정된 패키지 내의 모든 모듈 이름을 반환"""
    try:
        package = importlib.import_module(package_name)
        return {
            f"{package_name}.{name}"
            for _, name, _ in pkgutil.iter_modules(package.__path__)
        }
    except Exception as e:
        print(f"[ERROR] Failed to get modules for package {package_name}: {e}")
        return set()
def create_customer_module(customer_id: str):
    """고객별 모듈이 존재하는 경우에만 sys.modules를 대체"""
    print(f"[DEBUG] Creating customer modules for: {customer_id}")

    # app.models와 app.schemas 패키지의 모든 모듈 찾기
    base_modules = get_all_modules("app.models")
    base_modules.update(get_all_modules("app.schemas"))

    for base_name in base_modules:
        try:
            # 모듈 이름에서 마지막 부분 추출 (예: app.models.goods -> goods)
            module_parts = base_name.split('.')
            module_type = module_parts[1]  # models 또는 schemas
            module_name = module_parts[-1]

            # 고객별 모듈 경로 생성
            customer_name = f"app.{module_type}.{customer_id}.{module_name}"

            # 고객별 모듈이 존재하는지 확인
            try:
                customer_module = importlib.import_module(customer_name)
                print(f"[DEBUG] Found customer module: {customer_name}")

                # 기존 모듈이 있다면 제거
                if base_name in sys.modules:
                    del sys.modules[base_name]

                # 기본 모듈 이름으로 고객별 모듈 등록
                sys.modules[base_name] = customer_module
                print(f"[DEBUG] Replaced {base_name} with {customer_name}")

            except ImportError:
                print(f"[DEBUG] No customer module found for {customer_name}, using default")
                # 고객별 모듈이 없는 경우 기본 모듈 사용
                if base_name not in sys.modules:
                    importlib.import_module(base_name)

        except Exception as e:
            print(f"[ERROR] Error processing module {base_name}: {e}")
            # 오류 발생 시 기본 모듈 사용 시도
            try:
                if base_name not in sys.modules:
                    importlib.import_module(base_name)
            except Exception as e:
                print(f"[ERROR] Failed to load default module {base_name}: {e}")

이 코드는 다음과 같은 주요 기능을 수행합니다:

• 패키지 내의 모든 모듈을 검색
• 고객별 커스텀 모듈이 있는지 확인
• 있다면 기본 모듈 대신 커스텀 모듈을 로드

3. 기본 모델과 커스텀 모델

기본 Goods 모델:

app/models/goods.py

class Goods(SQLModel, table=True):
    __tablename__ = "goods"

    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    price: float
    description: Optional[str] = None
    created_at: datetime = Field(default_factory=datetime.utcnow)
    updated_at: datetime = Field(default_factory=datetime.utcnow)

CustomerA용 커스텀 Goods 모델:

app/models/customerA/goods.py

class Goods(SQLModel, table=True):
    __tablename__ = "goods_customerA"

    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    price: float
    description: Optional[str] = None
    created_at: datetime = Field(default_factory=datetime.utcnow)
    updated_at: datetime = Field(default_factory=datetime.utcnow)

    # CustomerA 전용 필드
    category_code: str = Field(index=True)
    vendor_code: str
    tax_rate: float = Field(default=0.1)

CustomerA 모델은 기본 모델을 확장하여 추가 필드를 포함합니다.

4. API 구현

FastAPI 라우터를 사용하여 API 엔드포인트를 구현합니다:

app/api/v1/goods.py

@router.get("/test/")
async def test_goods():
    # 동적으로 로드된 Goods 클래스 사용
    goods = Goods()
    print(f"Goods class: {Goods}")
    return goods

5. 애플리케이션 설정

마지막으로 FastAPI 애플리케이션 설정과 미들웨어 구현:

main.py

app = FastAPI()

@app.middleware("http")
async def customer_context_middleware(request: Request, call_next):
    customer_id = "customerA"
    set_customer_context(customer_id)
    response = await call_next(request)
    set_customer_context(None)
    return response

미들웨어는 각 요청마다 고객 컨텍스트를 설정하고 해제합니다.

작동 방식

1. 클라이언트가 요청을 보내면 미들웨어가 고객 ID를 컨텍스트에 설정
2. 요청 처리 중에 필요한 모델이나 스키마를 임포트하면 동적 모듈 로더가 작동
3. 고객별 커스텀 모듈이 있으면 그것을 사용하고, 없으면 기본 모듈을 사용
4. 응답이 완료되면 고객 컨텍스트가 초기화

장점

• 고객별로 다른 비즈니스 로직 구현 가능
• 코드 중복 최소화
• 런타임에 모듈 교체 가능
• 기존 코드 수정 없이 새로운 고객 지원 가능

주의사항

• 동적 모듈 로딩은 성능에 영향을 줄 수 있음
• 적절한 에러 처리와 로깅이 중요
• 테스트 케이스 작성이 복잡해질 수 있음

이 구현을 통해 단일 코드베이스로 여러 고객의 요구사항을 효율적으로 지원할 수 있습니다.