313 lines
9.4 KiB
Python
313 lines
9.4 KiB
Python
import logging
|
|
import traceback
|
|
from functools import wraps
|
|
from typing import Any, Callable, TypeVar
|
|
|
|
from fastapi import FastAPI, HTTPException, Request, Response, status
|
|
from fastapi.responses import JSONResponse
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
|
|
# 로거 설정
|
|
logger = logging.getLogger(__name__)
|
|
|
|
T = TypeVar("T")
|
|
|
|
|
|
class FastShipError(Exception):
|
|
"""Base exception for all exceptions in fastship api"""
|
|
# status_code to be returned for this exception
|
|
# when it is handled
|
|
status = status.HTTP_400_BAD_REQUEST
|
|
|
|
|
|
class EntityNotFound(FastShipError):
|
|
"""Entity not found in database"""
|
|
|
|
status = status.HTTP_404_NOT_FOUND
|
|
|
|
|
|
class BadPassword(FastShipError):
|
|
"""Password is not strong enough or invalid"""
|
|
|
|
status = status.HTTP_400_BAD_REQUEST
|
|
|
|
|
|
class ClientNotAuthorized(FastShipError):
|
|
"""Client is not authorized to perform the action"""
|
|
|
|
status = status.HTTP_401_UNAUTHORIZED
|
|
|
|
|
|
class ClientNotVerified(FastShipError):
|
|
"""Client is not verified"""
|
|
|
|
status = status.HTTP_401_UNAUTHORIZED
|
|
|
|
|
|
class NothingToUpdate(FastShipError):
|
|
"""No data provided to update"""
|
|
|
|
|
|
class BadCredentials(FastShipError):
|
|
"""User email or password is incorrect"""
|
|
|
|
status = status.HTTP_401_UNAUTHORIZED
|
|
|
|
|
|
class InvalidToken(FastShipError):
|
|
"""Access token is invalid or expired"""
|
|
|
|
status = status.HTTP_401_UNAUTHORIZED
|
|
|
|
|
|
class DeliveryPartnerNotAvailable(FastShipError):
|
|
"""Delivery partner/s do not service the destination"""
|
|
|
|
status = status.HTTP_406_NOT_ACCEPTABLE
|
|
|
|
|
|
class DeliveryPartnerCapacityExceeded(FastShipError):
|
|
"""Delivery partner has reached their max handling capacity"""
|
|
|
|
status = status.HTTP_406_NOT_ACCEPTABLE
|
|
|
|
|
|
# =============================================================================
|
|
# 데이터베이스 관련 예외
|
|
# =============================================================================
|
|
|
|
|
|
class DatabaseError(FastShipError):
|
|
"""Database operation failed"""
|
|
|
|
status = status.HTTP_503_SERVICE_UNAVAILABLE
|
|
|
|
|
|
class DatabaseConnectionError(DatabaseError):
|
|
"""Database connection failed"""
|
|
|
|
status = status.HTTP_503_SERVICE_UNAVAILABLE
|
|
|
|
|
|
class DatabaseTimeoutError(DatabaseError):
|
|
"""Database operation timed out"""
|
|
|
|
status = status.HTTP_504_GATEWAY_TIMEOUT
|
|
|
|
|
|
# =============================================================================
|
|
# 외부 서비스 관련 예외
|
|
# =============================================================================
|
|
|
|
|
|
class ExternalServiceError(FastShipError):
|
|
"""External service call failed"""
|
|
|
|
status = status.HTTP_502_BAD_GATEWAY
|
|
|
|
|
|
class GPTServiceError(ExternalServiceError):
|
|
"""GPT API call failed"""
|
|
|
|
status = status.HTTP_502_BAD_GATEWAY
|
|
|
|
|
|
class CrawlingError(ExternalServiceError):
|
|
"""Web crawling failed"""
|
|
|
|
status = status.HTTP_502_BAD_GATEWAY
|
|
|
|
|
|
class BlobStorageError(ExternalServiceError):
|
|
"""Azure Blob Storage operation failed"""
|
|
|
|
status = status.HTTP_502_BAD_GATEWAY
|
|
|
|
|
|
class CreatomateError(ExternalServiceError):
|
|
"""Creatomate API call failed"""
|
|
|
|
status = status.HTTP_502_BAD_GATEWAY
|
|
|
|
|
|
# =============================================================================
|
|
# 예외 처리 데코레이터
|
|
# =============================================================================
|
|
|
|
|
|
def handle_db_exceptions(
|
|
error_message: str = "데이터베이스 작업 중 오류가 발생했습니다.",
|
|
):
|
|
"""데이터베이스 예외를 처리하는 데코레이터.
|
|
|
|
Args:
|
|
error_message: 오류 발생 시 반환할 메시지
|
|
|
|
Example:
|
|
@handle_db_exceptions("사용자 조회 중 오류 발생")
|
|
async def get_user(user_id: int):
|
|
...
|
|
"""
|
|
|
|
def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
|
|
@wraps(func)
|
|
async def wrapper(*args: Any, **kwargs: Any) -> Any:
|
|
try:
|
|
return await func(*args, **kwargs)
|
|
except HTTPException:
|
|
# HTTPException은 그대로 raise
|
|
raise
|
|
except SQLAlchemyError as e:
|
|
logger.error(f"[DB Error] {func.__name__}: {e}")
|
|
logger.error(traceback.format_exc())
|
|
print(f"[DB Error] {func.__name__}: {e}")
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail=error_message,
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"[Unexpected Error] {func.__name__}: {e}")
|
|
logger.error(traceback.format_exc())
|
|
print(f"[Unexpected Error] {func.__name__}: {e}")
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail="서비스 처리 중 예기치 않은 오류가 발생했습니다.",
|
|
)
|
|
|
|
return wrapper
|
|
|
|
return decorator
|
|
|
|
|
|
def handle_external_service_exceptions(
|
|
service_name: str = "외부 서비스",
|
|
error_message: str | None = None,
|
|
):
|
|
"""외부 서비스 호출 예외를 처리하는 데코레이터.
|
|
|
|
Args:
|
|
service_name: 서비스 이름 (로그용)
|
|
error_message: 오류 발생 시 반환할 메시지
|
|
|
|
Example:
|
|
@handle_external_service_exceptions("GPT")
|
|
async def call_gpt():
|
|
...
|
|
"""
|
|
|
|
def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
|
|
@wraps(func)
|
|
async def wrapper(*args: Any, **kwargs: Any) -> Any:
|
|
try:
|
|
return await func(*args, **kwargs)
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
msg = error_message or f"{service_name} 서비스 호출 중 오류가 발생했습니다."
|
|
logger.error(f"[{service_name} Error] {func.__name__}: {e}")
|
|
logger.error(traceback.format_exc())
|
|
print(f"[{service_name} Error] {func.__name__}: {e}")
|
|
raise HTTPException(
|
|
status_code=status.HTTP_502_BAD_GATEWAY,
|
|
detail=msg,
|
|
)
|
|
|
|
return wrapper
|
|
|
|
return decorator
|
|
|
|
|
|
def handle_api_exceptions(
|
|
error_message: str = "요청 처리 중 오류가 발생했습니다.",
|
|
):
|
|
"""API 엔드포인트 예외를 처리하는 데코레이터.
|
|
|
|
Args:
|
|
error_message: 오류 발생 시 반환할 메시지
|
|
|
|
Example:
|
|
@handle_api_exceptions("가사 생성 중 오류 발생")
|
|
async def generate_lyric():
|
|
...
|
|
"""
|
|
|
|
def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
|
|
@wraps(func)
|
|
async def wrapper(*args: Any, **kwargs: Any) -> Any:
|
|
try:
|
|
return await func(*args, **kwargs)
|
|
except HTTPException:
|
|
raise
|
|
except SQLAlchemyError as e:
|
|
logger.error(f"[API DB Error] {func.__name__}: {e}")
|
|
logger.error(traceback.format_exc())
|
|
print(f"[API DB Error] {func.__name__}: {e}")
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail="데이터베이스 연결에 문제가 발생했습니다.",
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"[API Error] {func.__name__}: {e}")
|
|
logger.error(traceback.format_exc())
|
|
print(f"[API Error] {func.__name__}: {e}")
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=error_message,
|
|
)
|
|
|
|
return wrapper
|
|
|
|
return decorator
|
|
|
|
|
|
def _get_handler(status: int, detail: str):
|
|
# Define
|
|
def handler(request: Request, exception: Exception) -> Response:
|
|
# DEBUG PRINT STATEMENT 👇
|
|
from rich import print, panel
|
|
print(
|
|
panel.Panel(
|
|
exception.__class__.__name__,
|
|
title="Handled Exception",
|
|
border_style="red",
|
|
),
|
|
)
|
|
# DEBUG PRINT STATEMENT 👆
|
|
|
|
# Raise HTTPException with given status and detail
|
|
# can return JSONResponse as well
|
|
raise HTTPException(
|
|
status_code=status,
|
|
detail=detail,
|
|
)
|
|
# Return ExceptionHandler required with given
|
|
# status and detail for HTTPExcetion above
|
|
return handler
|
|
|
|
|
|
def add_exception_handlers(app: FastAPI):
|
|
# Get all subclass of 👇, our custom exceptions
|
|
exception_classes = FastShipError.__subclasses__()
|
|
|
|
for exception_class in exception_classes:
|
|
# Add exception handler
|
|
app.add_exception_handler(
|
|
# Custom exception class
|
|
exception_class,
|
|
# Get handler function
|
|
_get_handler(
|
|
status=exception_class.status,
|
|
detail=exception_class.__doc__,
|
|
),
|
|
)
|
|
|
|
@app.exception_handler(status.HTTP_500_INTERNAL_SERVER_ERROR)
|
|
def internal_server_error_handler(request, exception):
|
|
return JSONResponse(
|
|
content={"detail": "Something went wrong..."},
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
headers={
|
|
"X-Error": f"{exception}",
|
|
}
|
|
)
|
|
|