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

from fastapi import status, HTTPException
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, List, _GenericAlias, Optional

from commands import Command
# from pprint import pprint


class ResponseStructure:
    '''Класс структуры данных, для получения HATEAOS ответа на запросы
    клиента.'''
    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:
            self._data = args[0]
        if kwargs:
            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('/')}"


class Commands:
    '''Предварительная реализация контейнера описаний команд.'''
    def __init__(self, *commands: List[Command]):
        self._by_id: dict = dict()
        self._by_command: dict = dict()
        for command in commands:
            self._by_id[command.id] = command
            self._by_command[command.command] = command

    def get_commands(self, base_url: str) -> Dict[str, dict]:
        response = dict()

        for command in self._by_id.values():
            data = ResponseStructure(base_url)
            data.add_data(id=command.id,
                          title=command.title,
                          category=command.category,
                          command=command.command)
            data.add_link("self", f"/commands/{command.id}?by_id=true")
            data.add_link("parameters", f"/commands/{command.id}/parameters")
            data.add_link("configure", f"/configs/{command.id}")
            response[command.id] = data.get_dict()

        return response

    def get_by_id(self, command_id: str, base_url: str) -> Dict[str, Dict]:
        if command_id in self._by_id:
            command = self._by_id[command_id]
            data = ResponseStructure(base_url)
            data.add_data(id=command.id,
                          title=command.title,
                          category=command.category,
                          command=command.command)
            data.add_link("self", f"/commands/{command.id}?by_id=1")
            data.add_link("parameters", f"/commands/{command.id}/parameters")
            data.add_link("configure", f"/configs/{command.id}")
            parameters_data = self._get_parameters_data(command.parameters,
                                                        command.id, base_url)
            data.embed("parameters", parameters_data)
            return data.get_dict()

        return None

    def find_command(self, console_command: str,
                     base_url: str) -> Union[dict, None]:
        if console_command in self._by_command:
            command = self._by_command[console_command]

            data = ResponseStructure(base_url)
            data.add_data(id=command.id,
                          title=command.title,
                          category=command.category,
                          command=command.command)
            data.add_link("self", f"/commands/{command.id}?by_id=1")
            data.add_link("parameters", f"/commands/{command.id}/parameters")
            data.add_link("configure", f"/configs/{command.id}")
            parameters_data = self._get_parameters_data(command.parameters,
                                                        command.id, base_url)
            data.embed("parameters", parameters_data)

            return data.get_dict()
        else:
            return None

    def get_parameters(self, command_id: str, base_url: str
                       ) -> Union[dict, None]:
        if command_id in self._by_id:
            command = self._by_id[command_id]
            parameters_data = self._get_parameters_data(command.parameters,
                                                        command.id, base_url)
            return parameters_data
        else:
            return None

    def get_command_object(self, command_id: str) -> Command:
        if command_id in self._by_id:
            return self._by_id[command_id]
        return None

    def _get_parameters_data(self, parameters: list, command_id: str,
                             base_url: str) -> List[dict]:
        data = ResponseStructure(base_url)
        data.add_data(parameters)
        data.add_link("self", f"/commands/{command_id}/parameters")
        return data.get_dict()


def get_base_url(request: Request):
    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)