35 lines
1.7 KiB
Python
35 lines
1.7 KiB
Python
from typing import Literal, Union, Annotated
|
|
from pydantic import BaseModel, Field
|
|
|
|
# --- События ---
|
|
class HttpOkEvent(BaseModel):
|
|
"""Успешное HTTP-событие (2xx)"""
|
|
kind: Literal["ok"] = Field(description="Тип события: успешный ответ")
|
|
status: Literal[200] = Field(description="HTTP статус-код успешного ответа")
|
|
path: str = Field(description="Путь запроса (endpoint)")
|
|
duration_ms: float = Field(description="Длительность запроса в миллисекундах")
|
|
|
|
def __repr__(self) -> str:
|
|
return f"HttpOkEvent(kind='ok', status=200, path='{self.path}', duration_ms={self.duration_ms})"
|
|
|
|
|
|
class HttpErrorEvent(BaseModel):
|
|
"""Ошибочное HTTP-событие (4xx/5xx)"""
|
|
kind: Literal["error"] = Field(description="Тип события: ошибка")
|
|
status: int = Field(description="HTTP статус-код ошибки (4xx или 5xx)", ge=400, le=599)
|
|
path: str = Field(description="Путь запроса (endpoint)")
|
|
error_message: str = Field(description="Сообщение об ошибке")
|
|
|
|
def __repr__(self) -> str:
|
|
return f"HttpErrorEvent(kind='error', status={self.status}, path='{self.path}', error='{self.error_message[:30]}...')"
|
|
|
|
|
|
# --- Discriminated Union (способ 1: через аннотацию) ---
|
|
ApiEvent = Annotated[
|
|
Union[HttpOkEvent, HttpErrorEvent],
|
|
Field(discriminator="kind")
|
|
]
|
|
|
|
# Альтернативный способ через TypeAdapter (для прямого использования в Pydantic)
|
|
from pydantic import TypeAdapter
|
|
ApiEventAdapter = TypeAdapter(ApiEvent) |