from __future__ import annotations
from dataclasses import MISSING, Field, _process_class, dataclass, field, fields # type: ignore[attr-defined]
from enum import Enum, auto
from typing import TYPE_CHECKING, Any, Callable, TypeVar, overload
from edri.config.setting import API_RESPONSE_WRAPPED
from edri.dataclass.directive import ResponseDirective
from edri.utility.function import snake2camel
__all__ = ["Response", "response", "ResponseStatus"]
if TYPE_CHECKING:
from edri.api.dataclass.file import File
[docs]
class ResponseStatus(Enum):
NONE = auto()
OK = auto()
FAILED = auto()
[docs]
@dataclass
class Response:
_status: ResponseStatus = field(init=False, default=ResponseStatus.NONE)
_changed: bool | None = field(init=False, default=None)
_directives: list[ResponseDirective] = field(init=False, default_factory=list)
_is_file_only: bool = field(default=False, init=False)
[docs]
def as_dict(self, /, *, transform: bool, keep_concealed: bool, wrapped: bool = API_RESPONSE_WRAPPED) -> dict[str, Any]:
response_full: dict[str, Any] = {}
raw = {
f.name: getattr(self, f.name)
for f in fields(self)
if keep_concealed or not f.name.startswith("_")
}
if transform:
data = {
snake2camel(key, exceptions=("_", "id_")): value
for key, value in raw.items()
}
else:
data = raw
if wrapped:
if data:
response_full["data"] = data
else:
response_full = data
response_full["status"] = self._status.name
return response_full
[docs]
def get_status(self) -> ResponseStatus:
return self._status
[docs]
def set_status(self, status: ResponseStatus) -> None:
self._status = status
def __setattr__(self, key: str, value: Any) -> None:
if self._changed is not None and not self._changed:
super().__setattr__("_changed", True)
super().__setattr__("_status", ResponseStatus.OK)
super().__setattr__(key, value)
def __post_init__(self) -> None:
self._directives = []
self._changed = False # Has to by the last
[docs]
def has_changed(self) -> bool:
return self._changed if self._changed else False
[docs]
def add_directive(self, directive: ResponseDirective, /, *, ignore_status: bool = False) -> None:
if self._changed is not None and not self._changed and not ignore_status:
super().__setattr__("_changed", True)
super().__setattr__("_status", ResponseStatus.OK)
self._directives.append(directive)
[docs]
def get_directives(self) -> list[ResponseDirective]:
return self._directives
@property
def is_file_only(self) -> bool:
return self._is_file_only
_R = TypeVar("_R", bound=Response)
def _response(cls: type[_R], init: bool, repr: bool, eq: bool, order: bool, unsafe_hash: bool, frozen: bool,
match_args: bool, kw_only: bool, slots: bool, weakref_slot: bool | None) -> type[_R]:
for name, value in cls.__annotations__.items():
attribute = getattr(cls, name, False)
if isinstance(attribute, Field):
# Preserve caller-supplied defaults, factories and metadata. EDRI
# response fields remain output-only even when declared with
# dataclasses.field(...).
attribute.init = False
if attribute.default is MISSING and attribute.default_factory is MISSING:
# An output-only field with neither a default nor a factory has
# no way to be assigned: _process_class drops the class
# attribute and no __init__ sets it, so getattr() would raise.
# Match the implicit None the other branches already give.
attribute.default = None
setattr(cls, name, attribute)
elif attribute:
setattr(cls, name, field(init=False, default=attribute))
else:
setattr(cls, name, field(init=False, default=None))
dataclass: type[_R] = _process_class(cls, init=init, repr=repr, eq=eq, order=order, unsafe_hash=unsafe_hash, frozen=frozen,
match_args=match_args,
kw_only=kw_only, slots=slots, weakref_slot=weakref_slot)
if len(cls.__annotations__) == 1:
from edri.api.dataclass.file import File
field_file = cls.__annotations__.get("file")
if field_file == File:
dataclass._is_file_only = True
return dataclass
@overload
def response(cls: type[_R], /) -> type[_R]: ...
@overload
def response(cls: None = ..., /, *, init: bool = ..., repr: bool = ..., eq: bool = ..., order: bool = ...,
unsafe_hash: bool = ..., frozen: bool = ..., match_args: bool = ..., kw_only: bool = ...,
slots: bool = ..., weakref_slot: bool | None = ...) -> Callable[[type[_R]], type[_R]]: ...
[docs]
def response(cls: type[_R] | None = None, /, *, init: bool = True, repr: bool = True, eq: bool = True, order: bool = False,
unsafe_hash: bool = False, frozen: bool = False, match_args: bool = True, kw_only: bool = False,
slots: bool = False, weakref_slot: bool | None = None) -> type[_R] | Callable[[type[_R]], type[_R]]:
def wrapper(cls: type[_R]) -> type[_R]:
return _response(cls, init=init, repr=repr, eq=eq, order=order, unsafe_hash=unsafe_hash,
frozen=frozen, match_args=match_args, kw_only=kw_only, slots=slots, weakref_slot=weakref_slot)
if cls is None:
return wrapper
return wrapper(cls)