|
|
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)
|