from __future__ import annotations from dataclasses import dataclass, field from typing import TYPE_CHECKING, Any, ClassVar, TypedDict import pydantic if TYPE_CHECKING: from pydantic.error_wrappers import Loc # noqa class OwsError(Exception): default_code: ClassVar[str] = "ows_error" default_message: ClassVar[str] = "Ows error" def __init__( self, message: str | None = None, *, code: str | None = None, context: dict[str, Any] | None = None, ) -> None: self.code = code or self.default_code self.message = message or self.default_message self.context = context or {} def __str__(self) -> str: return self.message class ObjectDoesNotExist(OwsError): default_code = "object_does_not_exist" default_message = "Object does not exist" def __init__( self, message: str | None = None, code: str | None = None, context: dict[str, Any] | None = None, ) -> None: context = context or {} context.setdefault("status_code", 404) super().__init__(message, code=code, context=context) @dataclass(frozen=True) class Error: code: str message: str source: list[str] = field(default_factory=list) def field_error(self, field: str) -> FieldError: # noqa return FieldError( field=field, message=self.message, code=self.code, ) class Errors: @classmethod def all(cls) -> list[Error]: return [error for error in cls.__dict__.values() if isinstance(error, Error)] @dataclass(frozen=True) class FieldError: default_code: ClassVar[str] = "invalid" field: str message: str code: str = default_code @classmethod def required(cls, field_name: str) -> FieldError: return cls( field=field_name, code="value_error.missing", message="Field required", ) class FieldErrorDict(TypedDict): field: str message: str class ValidationError(OwsError): default_code = "validation_error" default_message = "Validation error" def __init__(self, errors: list[FieldError] | None = None): super().__init__() self.errors = errors or [] def has_errors(self) -> bool: return bool(self.errors) def add_error( self, error: FieldError | FieldErrorDict, code: str | None = None, raise_exception: bool = False, ) -> None: if isinstance(error, dict): error = FieldError( field=error["field"], message=error["message"], code=code or FieldError.default_code, ) self.errors.append(error) if raise_exception: raise self @classmethod def from_pydantic_error( cls, pydantic_error: pydantic.ValidationError, loc_pos: int = 0 ) -> ValidationError: return cls( errors=[ FieldError( field=cls._flatten_loc(detail["loc"][loc_pos:]), message=detail["msg"], code=detail["type"], ) for detail in pydantic_error.errors() ] ) @staticmethod def _flatten_loc(loc: Loc) -> str: return ".".join(map(str, loc))