import uuid from collections.abc import Callable from datetime import datetime from typing import Annotated, Any, Literal, Optional, Union from django.utils.timezone import now from ninja import Field, Schema from pydantic import ValidationError, WrapValidator def invalid_to_none(v: Any, handler: Callable[[Any], Any]) -> Any: try: return handler(v) except ValidationError: return None class TagKeyValue(Schema): key: str value: str class Signal(Schema): number: int code: Optional[int] name: Optional[str] code_name: Optional[str] class MachException(Schema): number: int code: int subcode: int name: Optional[str] class NSError(Schema): code: int domain: str class Errno(Schema): number: int name: Optional[str] class MechanismMeta(Schema): signal: Optional[Signal] = None match_exception: Optional[MachException] = None ns_error: Optional[NSError] = None errno: Optional[Errno] = None class ExceptionMechanism(Schema): type: str description: Optional[str] = None help_link: Optional[str] = None handled: Optional[bool] = None synthetic: Optional[bool] = None meta: Optional[dict] = None data: Optional[dict] = None class StackTraceFrame(Schema): filename: Optional[str] = None function: Optional[str] = None raw_function: Optional[str] = None module: Optional[str] = None lineno: Optional[int] = None colno: Optional[int] = None abs_path: Optional[str] = None context_line: Optional[str] = None pre_context: Optional[list[str]] = None post_context: Optional[list[str]] = None source_link: Optional[str] = None in_app: Optional[bool] = None stack_start: Optional[bool] = None vars: Optional[dict[str, str]] = None instruction_addr: Optional[str] = None addr_mode: Optional[str] = None symbol_addr: Optional[str] = None image_addr: Optional[str] = None package: Optional[str] = None platform: Optional[str] = None class StackTrace(Schema): frames: list[StackTraceFrame] registers: Optional[dict[str, str]] = None class EventException(Schema): type: str value: Annotated[Optional[str], WrapValidator(invalid_to_none)] module: Optional[str] = None thread_id: Optional[str] = None mechanism: Optional[ExceptionMechanism] = None stacktrace: Optional[StackTrace] = None class ValueEventException(Schema): values: list[EventException] class EventMessage(Schema): formatted: str = Field( max_length=10, default="" ) # When None, attempt to generate it message: Optional[str] = None params: Optional[list[str]] = None class EventTemplate(Schema): lineno: int abs_path: Optional[str] = None filename: str context_line: str pre_context: Optional[list[str]] = None post_context: Optional[list[str]] = None Level = Literal["fatal", "error", "warning", "info", "debug"] class EventBreadcrumb(Schema): type: Optional[str] = None category: Optional[str] = None message: Optional[str] = None data: Optional[dict[str, Any]] = None level: Optional[Level] = None timestamp: Optional[datetime] = None class ValueEventBreadcrumb(Schema): values: list[EventBreadcrumb] class BaseEventIngestSchema(Schema): timestamp: datetime = Field(default_factory=now) platform: Optional[str] = None level: Optional[str] = "error" logger: Optional[str] = None transaction: Optional[str] = None server_name: Optional[str] = None release: Optional[str] = None dist: Optional[str] = None tags: Optional[Union[dict[str, str], list[TagKeyValue]]] = None environment: Optional[str] = None modules: Optional[dict[str, str]] = None extra: Optional[Any] = None fingerprint: Optional[list[str]] = None errors: Optional[list[Any]] = None exception: Optional[Union[list[EventException], ValueEventException]] = None message: Optional[EventMessage] = None template: Optional[EventTemplate] = None breadcrumbs: Optional[Union[list[EventBreadcrumb], ValueEventBreadcrumb]] = None class EnvelopeEventIngestSchema(BaseEventIngestSchema): type: str class EventIngestSchema(BaseEventIngestSchema): event_id: uuid.UUID class EnvelopeHeaderSchema(Schema): event_id: uuid.UUID dsn: Optional[str] = None sdk: Optional[Any] = None sent_at: datetime = Field(default_factory=now) class ItemHeaderSchema(Schema): content_type: Optional[str] type: Literal["transaction", "event"] length: Optional[int] EnvelopeSchema = list[ Union[EnvelopeHeaderSchema, ItemHeaderSchema, EnvelopeEventIngestSchema] ]