calculate-utils-4-lib/calculate/server/utils/responses.py

from fastapi import HTTPException, status
from fastapi.encoders import jsonable_encoder

from starlette.responses import JSONResponse
from starlette.requests import Request

from pydantic.main import ModelMetaclass
from pydantic import parse_obj_as

from typing import Dict, Union, _GenericAlias, Optional


class ResponseStructure:
    '''Класс структуры данных, для получения HATEOAS ответа на запросы
    клиента.'''
    def __init__(self, base_url: str):
        self._base_url: str = base_url.strip("/")

        self._data: dict = dict()
        self._links: Dict[str, Dict[str, str]] = dict()
        self._embedded: Dict[str, dict] = dict()

    def get_dict(self) -> dict:
        dictionary = dict()
        if self._data:
            dictionary["data"] = self._data
        if self._links:
            dictionary["_links"] = self._links
        if self._embedded:
            dictionary["_embedded"] = self._embedded
        return dictionary

    def add_data(self, *args, **kwargs) -> "ResponseStructure":
        if args:
            if isinstance(self._data, dict):
                if len(args) > 1:
                    self._data = list(args)
                else:
                    self._data = args[0]
            elif isinstance(self._data, list):
                if len(args) > 1:
                    self._data.extend(args)
                else:
                    self._data.append(args[0])
            else:
                self._data = [self._data, *args]
        elif kwargs:
            if not isinstance(self._data, dict):
                self._data = {}
            self._data.update(kwargs)
        return self

    def embed(self, resource: str,
              content: Union[dict, list]) -> "ResponseStructure":
        self._embedded[resource] = content
        return self

    def add_link(self, action: str, uri: str,
                 templated: bool = False) -> "ResponseStructure":
        uri = self._get_uri(uri)

        if action in self._links:
            link_section = self._links[action]
            if link_section["href"] != uri:
                link_section["href"] = uri
        else:
            link_section = {"href": uri}
            self._links[action] = link_section

        if templated:
            print("TEMPLATED:", action)
            link_section["templated"] = templated

        return self

    def _get_uri(self, uri: str) -> str:
        if uri.startswith(self._base_url):
            return uri
        return f"{self._base_url}/{uri.strip('/')}"


def get_base_url(request: Request) -> str:
    base_url = f"{request.base_url.scheme}://{request.base_url.netloc}"
    return base_url


def get_command_not_found(command_id: str) -> HTTPException:
    return HTTPException(status_code=status.HTTP_404_NOT_FOUND,
                         detail=(f'Command with id "{command_id}"'
                                 ' is not found.'))


def get_cl_command_not_found(console_command: str) -> HTTPException:
    return HTTPException(status_code=status.HTTP_404_NOT_FOUND,
                         detail=(f'Console command "{console_command}" is not'
                                 ' found.'))


def get_configuration_not_found(wid: int) -> HTTPException:
    return HTTPException(status_code=status.HTTP_404_NOT_FOUND,
                         detail=f"Configuration with id={wid} is not found.")


def validate_response(data: Union[dict, list],
                      schema: Union[ModelMetaclass, _GenericAlias, type],
                      media_type: Optional[str] = None,
                      status_code: int = 200
                      ) -> JSONResponse:
    """Функция для валидации данных ответа сервера по указанной схеме с учетом
    наличия псевдонимов полей и возможности того, что схема задана обычным
    типом, типом из typing или BaseModel с __root__. Возвращает объект ответа.
    """
    if isinstance(schema, ModelMetaclass):
        if "__root__" in schema.__annotations__:
            validated = schema(__root__=data)
        else:
            validated = schema(**data)
    elif isinstance(schema, (_GenericAlias, type)):
        validated = parse_obj_as(schema, data)

    return JSONResponse(content=jsonable_encoder(validated, by_alias=True),
                        media_type=media_type, status_code=status_code)