728x90

FastAPI Hot Reload - Sqlmodel Metadata Reload Issue

문제 분석
정리

FastAPI에서 sys.modules 조작이 잘 유지된 상태라고 가정하더라도, SQLModel의 metadata 초기화 문제가 리로드 시 발생할 가능성이 있습니다.
이는 SQLModel(SQLAlchemy 기반)이 FastAPI 애플리케이션의 재시작과 함께 metadata 객체를 초기화하기 때문입니다.

문제 분석

문제의 원인: SQLModel의 `metadata` 초기화

SQLModel은 기본적으로 sqlmodel.sqlalchemy.MetaData 객체를 사용해 테이블 스키마를 관리합니다. Uvicorn의 --reload가 작동하면 Python 프로세스가 재시작되고, 이에 따라 다음 문제가 발생할 수 있습니다:

테이블 등록 정보 손실: 리로드로 인해 SQLModel.metadata가 초기화되면서 기존에 등록된 테이블 정보가 사라집니다.
재등록 문제: 애플리케이션이 다시 시작되면서 동일한 테이블이 재등록되거나, 기존 테이블과 충돌이 발생할 수 있습니다.
세션 관리 문제: 기존 Session 또는 engine 객체가 유효하지 않게 되어, 데이터베이스 작업이 실패할 가능성이 있습니다.

문제 확인 방법

다음 사항을 점검하여 리로드 문제인지 확인하세요:

metadata 초기화 확인
리로드 시 SQLModel.metadata 객체가 초기화되는지 확인합니다.
```
from sqlmodel import SQLModel

print(f"Tables in metadata: {SQLModel.metadata.tables.keys()}")
```
리로드 후 테이블 목록이 빈 값으로 출력된다면 metadata가 초기화된 것입니다.
리로드 시 에러 확인
리로드 후 로그에서 다음과 같은 에러를 확인합니다:
- sqlalchemy.exc.NoSuchTableError
- sqlalchemy.exc.InvalidRequestError: Table already exists
테이블 중복 정의 확인
리로드 이후 테이블이 중복으로 정의되거나, 데이터베이스에 충돌이 발생하는지 확인합니다.

문제 해결 방안

리로드 시 SQLModel.metadata가 초기화되는 문제를 방지하거나 복구하기 위해 다음 방법을 사용할 수 있습니다.

1. `metadata` 재등록 로직 추가

리로드 시 SQLModel.metadata가 초기화되었다면, 이를 다시 설정해주는 로직을 추가합니다. FastAPI 애플리케이션 초기화 시, metadata를 재등록하도록 설정합니다.

from sqlmodel import SQLModel, create_engine

# 엔진 생성
DATABASE_URL = "sqlite:///example.db"
engine = create_engine(DATABASE_URL)

# 테이블 재등록
def register_metadata():
    # 모든 모델에서 테이블을 재등록
    SQLModel.metadata.create_all(bind=engine)
    print("[INFO] Metadata registered!")

# FastAPI 앱 초기화
from fastapi import FastAPI

app = FastAPI()

@app.on_event("startup")
def on_startup():
    register_metadata()

@app.get("/")
def read_root():
    return {"message": "Metadata reloaded successfully!"}

2. `metadata` 충돌 방지

SQLAlchemy가 동일한 테이블을 두 번 정의하지 않도록 방지하려면, 테이블 정의 전에 metadata 객체를 초기화하는 조건을 추가합니다.

from sqlmodel import SQLModel

# 초기화 방지
if not SQLModel.metadata.tables:
    SQLModel.metadata.clear()

이 코드로 테이블 중복 정의를 방지할 수 있습니다.

3. `sys.modules` 활용하여 상태 유지

리로드 시에도 SQLModel.metadata를 유지하기 위해 sys.modules를 활용해 테이블 상태를 저장할 수 있습니다.

import sys
from sqlmodel import SQLModel

# 기존 metadata 유지
if "app_metadata" in sys.modules:
    SQLModel.metadata = sys.modules["app_metadata"]
else:
    sys.modules["app_metadata"] = SQLModel.metadata

정리

리로드로 인해 SQLModel.metadata가 초기화되는 문제는 다음과 같이 해결할 수 있습니다:

metadata 재등록: FastAPI의 startup 이벤트에서 SQLModel.metadata.create_all()을 호출.
테이블 중복 방지: 리로드 전에 metadata를 초기화하거나, sys.modules를 활용하여 상태를 유지.

이 접근 방식으로 리로드 시 발생하는 SQLModel의 문제를 안정적으로 해결할 수 있습니다.

728x90

'IT Best Practise > FastAPI' 카테고리의 다른 글

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

728x90

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

소개
시스템 구조
작동 방식
장점
주의사항

소개

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

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

작동 방식

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

장점

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

주의사항

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

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

728x90

'IT Best Practise > FastAPI' 카테고리의 다른 글

FastAPI Hot Reload - Sqlmodel Metadata Reload Issue (0)	2024.11.21

GilliLab - TechLog

IT Best Practise/FastAPI