FastAPI의 예외 처리, blinker, SQLAlchemy의 listens_for 정리

1. FastAPI의 예외 처리

   1) 기본 예외 처리

      1. HTTPException

         - HTTP 상태 코드와 추가 정보를 담은 예외를 발생시킬 수 있다.

from fastapi import FastAPI, HTTPException

app = FastAPI()

@app.get("/")
def read_root():
    raise HTTPException(status_code=404, detail="Item not found")

 

      2. RequestValidationError

         - 요청 데이터의 유효성 검사가 실패했을 때 발생한다. FastAPI가 자동으로 처리하지만, 커스터마이징이 가능하다.

         - 잘못된 JSON 데이터를 전송하면 RequestValidationError가 발생한다.

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float

@app.post("/items/")
def create_item(item: Item):
    return item

 

      3. Starlette's WebSocketDisconnect
         - WebSocket 연결이 중단되었을 때 발생한다.

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        data = await websocket.receive_text()
        await websocket.send_text(f"Message: {data}")
    except WebSocketDisconnect:
        print("Client disconnected")

 

   2) 커스텀 예외 처리

      - FastAPI에서는 기본 제공 예외 외에도 사용자 정의 예외를 만들 수 있다. 커스텀 예외를 정의하고 이를 처리하는 핸들러를 등록할 수 있다.

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse

app = FastAPI()

class CustomException(Exception):
    def __init__(self, name: str, detail: str):
        self.name = name
        self.detail = detail

@app.exception_handler(CustomException)
async def custom_exception_handler(request: Request, exc: CustomException):
    return JSONResponse(
        status_code=400,
        content={"error": {"name": exc.name, "detail": exc.detail}},
    )

@app.get("/custom-error")
def custom_error():
    raise CustomException(name="CustomError", detail="This is a custom error")

 

   3) 기본 예외 재정의하기

      - FastAPI와 Starlette는 둘 다 HTTPException을 가지고 있다.

         - FastAPI는 FastAPI의 HTTP exception을 가지고 있고, Startlette's의 Exception을 상속 받는다.
         - 차이점은 FastAPI's HTTPException은 response에 헤더를 추가할 수 있다는 것이다.

 

      - 기존 HTTPException을 재정의 할때 Starlette의 HTTPException에다가 등록해야한다.

      - Starlette와 FastAPI는 같은 HTTPException 클래스를 사용하지만, Starlette의 내부 코드나 확장 플러그인이 발생시키는 예외를 처리하기 위해서는 Starlette의 HTTPException에 핸들러를 등록하는 것이 필요하다.

from fastapi import FastAPI, HTTPException, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from starlette.exceptions import HTTPException as StarletteHTTPException

app = FastAPI()

# Starlette HTTPException 핸들러
# 모든 HTTPException(Starlette 및 FastAPI의 HTTPException 포함)을 처리하도록 등록
@app.exception_handler(StarletteHTTPException)
async def starlette_http_exception_handler(request: Request, exc: StarletteHTTPException):
    """
    Starlette의 HTTPException을 처리하는 핸들러.
    """
    return JSONResponse(
        status_code=exc.status_code,
        content={"error": "HTTPException occurred", "detail": exc.detail},
    )

# FastAPI RequestValidationError 핸들러
# 요청 데이터 유효성 검사가 실패했을 때 사용자 친화적인 메시지를 반환
@app.exception_handler(RequestValidationError)
async def request_validation_exception_handler(request: Request, exc: RequestValidationError):
    """
    RequestValidationError를 처리하는 핸들러.
    """
    return JSONResponse(
        status_code=422,
        content={"error": "Validation Error", "details": exc.errors()},
    )

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    """
    아이템 ID를 받아 처리하는 엔드포인트.
    특정 ID(3)에서 커스텀 HTTPException을 발생시킵니다.
    """
    if item_id == 3:
        # FastAPI HTTPException을 발생시킴
        raise HTTPException(status_code=418, detail="Nope! I don't like 3.")

    return {"item_id": item_id}

 

2. blinker 사용 방법 정리

   - Blinker는 Python의 경량 이벤트 시스템 라이브러리로, 간단한 방식으로 이벤트를 정의하고 연결할 수 있도록 돕는다.

   - Django의 signal 시스템과 유사한 방식으로 동작하며, 이벤트와 이벤트 리스너를 통해 비동기 작업이나 상태 변화를 처리할 수 있다.

 

   1) 주요 개념 및 특징 :

      1. Signal
         - 이벤트를 정의하고 호출한다.
         - 특정 이벤트가 발생했을 때, 연결된 리스너를 실행한다.
      2. Subscriber

         - Signal에 연결된 함수로, 이벤트가 발생했을 때 실행된다.
      3. Sender
         - Signal을 호출하는 주체로, 이벤트와 관련된 데이터를 제공할 수 있다.
      4. Lightweight
         - 의존성이 거의 없고, 간단한 API를 제공하여 빠르게 학습하고 적용할 수 있다.

 

   2) Blinker 사용 방법

      - 설치 :

pip install blinker

 

      - 기본 사용 예제

         1. Signal 정의 및 호출

from blinker import Signal

# Signal 생성
event_signal = Signal('example_event')

# 리스너 함수 정의
def event_handler(sender, **kwargs):
    print(f"Event received from {sender} with data: {kwargs}")

# Signal에 리스너 연결
event_signal.connect(event_handler)

# 이벤트 호출
event_signal.send('source', data="Hello, Blinker!")

 

         1-1. 여러개의 이벤트 연결과 실행 순서

            - 하나의 시그널에 여러개의 이벤트를 연결할 수 있으며, 연결된 이벤트는 모두 실행된다.

            - 실행 순서는 리스너가 연결된 순서대로 호출된다.

from fastapi import FastAPI
from blinker import Signal

app = FastAPI()

# Signal 정의
user_registered = Signal('user_registered')

# 리스너 1
def send_welcome_email(sender, user_email, **kwargs):
    print(f"Welcome email sent to {user_email}!")

# 리스너 2
def log_registration(sender, user_email, **kwargs):
    print(f"User {user_email} registered from {sender}.")

# 리스너 3
def send_discount_offer(sender, user_email, **kwargs):
    print(f"Discount offer sent to {user_email}!")

# Signal에 리스너 연결
user_registered.connect(send_welcome_email)   # 첫 번째로 연결
user_registered.connect(log_registration)     # 두 번째로 연결
user_registered.connect(send_discount_offer)  # 세 번째로 연결

@app.post("/register")
async def register_user(email: str):
    print(f"User {email} registered.")
    # 이벤트 호출
    user_registered.send('register_user_endpoint', user_email=email)
    return {"message": f"User {email} successfully registered."}

 

         2. 한 번만 실행되는 리스너

def one_time_handler(sender, **kwargs):
    print("This will run only once.")

# Signal에 연결
event_signal.connect(one_time_handler, weak=False)

# 이벤트 호출
event_signal.send('source')  # 실행됨
event_signal.send('source')  # 실행되지 않음

 

         3. 익명 리스너

# Signal에 익명 함수 연결
event_signal.connect(lambda sender: print("Anonymous listener triggered."))
event_signal.send('source')

 

         4. Signal 제거

# 연결 해제
event_signal.disconnect(event_handler)

 

      - 실무 기반 가상 시나리오 프로젝트

         1. 사용자 가입 이벤트 처리

from fastapi import FastAPI
from blinker import Signal

app = FastAPI()

# Signal 정의
user_registered = Signal('user_registered')

# 리스너: 이메일 알림 발송
def send_welcome_email(sender, user_email, **kwargs):
    print(f"Welcome email sent to {user_email}!")

user_registered.connect(send_welcome_email)

@app.post("/register")
async def register_user(email: str):
    # 사용자 가입 로직 처리 (DB 저장 등)
    print(f"User {email} registered.")
    # 이벤트 호출
    user_registered.send('register_user_endpoint', user_email=email)
    return {"message": f"User {email} successfully registered."}

 

         3. 주문 상태 변경 알림

from fastapi import FastAPI
from blinker import Signal

app = FastAPI()

# Signal 정의
order_status_updated = Signal('order_status_updated')

# 리스너: 상태 변경 기록
def log_order_status_change(sender, order_id, new_status, **kwargs):
    print(f"Order {order_id} status changed to {new_status}.")

# 리스너: 고객 알림
def notify_customer(sender, order_id, new_status, **kwargs):
    print(f"Customer notified: Order {order_id} is now {new_status}.")

order_status_updated.connect(log_order_status_change)
order_status_updated.connect(notify_customer)

@app.post("/update_order_status")
async def update_order_status(order_id: int, status: str):
    # 주문 상태 업데이트 로직 (DB 업데이트 등)
    print(f"Updating order {order_id} to status {status}.")
    # 이벤트 호출
    order_status_updated.send('order_service', order_id=order_id, new_status=status)
    return {"message": f"Order {order_id} updated to status {status}."}

 

   3) SQLAlchemy의 listens_for 사용 방법

  - SQLAlchemy의의 @listens_for는 이벤트 리스너를 등록할 때 사용하는 데코레이터이다. SQLAlchemy에서 모델 객체의 특정 이벤트가 발생했을 때 이를 감지하고 사용자 정의 동작을 실행할 수 있다. 이 기능은 ORM과 Core 모두에서 사용할 수 있으며, 다양한 이벤트에 대한 훅(hook)을 제공한다.

 

      1. 기본 개념

         - @listens_for 데코레이터는 특정 이벤트가 발생했을 때 호출될 함수를 등록한다.

@sqlalchemy.event.listens_for(target, identifier[, propagate=False])

 

         - target: 이벤트를 청취할 대상이다. 예를 들어, Table, Mapper, Session, Engine 등이 될 수 있다.
         - identifier: 이벤트 이름이다. 예: 'before_insert', 'after_update'.
         - propagate: 기본값은 False이며, True로 설정하면 상속된 모든 클래스에서도 이벤트가 호출된다.

 

      2. 주요 이벤트와 사용 가능한 옵션

         - ORM 이벤트

            1. before_insert

               - 설명: INSERT 실행 전에 호출된다.

from sqlalchemy import event
from sqlalchemy.orm import Session

@event.listens_for(Session, 'before_insert')
def before_insert_listener(mapper, connection, target):
    print(f"Before insert: {target}")

 

           2. after_insert

               - 설명: INSERT 실행 후 호출된다.

@event.listens_for(Session, 'after_insert')
def after_insert_listener(mapper, connection, target):
    print(f"After insert: {target}")

 

           3. before_update

               - 설명: UPDATE 실행 전에 호출된다.

@event.listens_for(Session, 'before_update')
def before_update_listener(mapper, connection, target):
    print(f"Before update: {target}")

 

           4. after_update

               - 설명: UPDATE 실행 후 호출된다.

@event.listens_for(Session, 'after_update')
def after_update_listener(mapper, connection, target):
    print(f"After update: {target}")

 

           5. before_delete

               - 설명: DELETE 실행 전에 호출된다.

@event.listens_for(Session, 'before_delete')
def before_delete_listener(mapper, connection, target):
    print(f"Before delete: {target}")

 

           6. after_delete

               - 설명: DELETE 실행 후 호출된다.

@event.listens_for(Session, 'after_delete')
def after_delete_listener(mapper, connection, target):
    print(f"After delete: {target}")

 

         - Core 이벤트

            1. before_execute

               - 설명: SQL 실행 전에 호출된다.

@event.listens_for(Engine, 'before_execute')
def before_execute_listener(conn, clause, multiparams, params):
    print(f"Before execute: {clause}")

 

            2. after_execute

               - 설명: SQL 실행 후 호출된다.

@event.listens_for(Engine, 'after_execute')
def after_execute_listener(conn, clause, multiparams, params, result):
    print(f"After execute: {clause}")

 

            3. connect

               - 설명: 데이터베이스 연결이 생성될 때 호출됩니다.

@event.listens_for(Engine, 'connect')
def connect_listener(conn, record):
    print("Database connected")

 

         - 관계형 이벤트

            1. load

               - 설명: 객체가 데이터베이스에서 로드될 때 호출된다.

@event.listens_for(MyModel, 'load')
def load_listener(target, context):
    print(f"Loaded: {target}")

 

            2. refresh

               - 설명: 객체가 새로 고침될 때 호출된다.

@event.listens_for(MyModel, 'refresh')
def refresh_listener(target, context, attrs):
    print(f"Refreshed: {target}")

 

         - 복합적인 사용 예

            - 트랜잭션 내 로깅

from sqlalchemy import event
from sqlalchemy.orm import Session

@event.listens_for(Session, 'after_begin')
def after_begin_listener(session, transaction, connection):
    print("Transaction started")

@event.listens_for(Session, 'before_commit')
def before_commit_listener(session):
    print("Before commit")

@event.listens_for(Session, 'after_commit')
def after_commit_listener(session):
    print("After commit")

@event.listens_for(Session, 'after_rollback')
def after_rollback_listener(session):
    print("Transaction rolled back")

 

         - 옵션과 특징

            - 이벤트 전파: propagate=True로 설정하면, 상위 클래스의 이벤트 리스너를 하위 클래스에서도 사용할 수 있다.

            - propagate=True 설정은 SQLAlchemy에서 이벤트 리스너가 부모 클래스의 모든 하위 클래스에도 전파되도록 하는 옵션이다.

            - 이 설정을 통해 부모 클래스에 한 번만 이벤트 리스너를 등록하면, 그 부모 클래스를 상속받는 모든 하위 클래스에서 해당 리스너가 적용된다.

from sqlalchemy import create_engine, Column, Integer, String, event
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# 기본 설정
Base = declarative_base()
engine = create_engine('sqlite:///:memory:')
Session = sessionmaker(bind=engine)
session = Session()

# 공통 이벤트 리스너 정의
@event.listens_for(Base, 'before_insert', propagate=True)
def before_insert_for_all(mapper, connection, target):
    print(f"Before insert called for: {target}")

# 첫 번째 자식 클래스
class User(Base):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True)
    name = Column(String)

# 두 번째 자식 클래스
class Product(Base):
    __tablename__ = 'product'
    id = Column(Integer, primary_key=True)
    name = Column(String)

# 테이블 생성
Base.metadata.create_all(engine)

# User 인스턴스 추가
user = User(name="Alice")
session.add(user)
session.commit()  # "Before insert called for: <User(name='Alice')>" 출력

# Product 인스턴스 추가
product = Product(name="Laptop")
session.add(product)
session.commit()  # "Before insert called for: <Product(name='Laptop')>" 출력

 

         - 객체별 필터링

            - 특정 조건에서만 실행되도록 사용자 정의 로직을 추가할 수 있다.

@event.listens_for(MyModel, 'before_insert')
def conditional_listener(mapper, connection, target):
    if target.some_field == "value":
        print("Special condition met")

 

 - reference : 

https://lucky516.tistory.com/101

Fast API 배우기 16부 - Handling Errors

https://rudaks.tistory.com/entry/fastapi-Custom-Exception-%EC%B2%98%EB%A6%AC

[fastapi] Custom Exception 처리

https://velog.io/@power0080/FastAPI-X-Python-%EC%8B%A4%EB%AC%B4%EC%97%90%EC%84%9C-%EB%B0%94%EB%A1%9C-%EC%93%B0%EB%8A%94-FastAPI-%EC%A0%84%EC%97%AD-%EC%97%90%EB%9F%AC-%EC%B2%98%EB%A6%AC

https://mopil.tistory.com/entry/FastAPI-%EC%9D%B4%EB%B2%A4%ED%8A%B8-%EA%B8%B0%EB%B0%98-%EC%95%84%ED%82%A4%ED%85%8D%EC%B2%98-%EA%B5%AC%EC%B6%95%ED%95%98%EA%B8%B0-w-fastapi-events

[FastAPI] 이벤트 기반 아키텍처 구축하기 (w. fastapi-events)