Source code for openapi.data.view
import os
from dataclasses import dataclass
from typing import Any, Dict, NoReturn, Optional, cast
from aiohttp import web
from ..types import DataType, QueryType
from ..utils import TypingInfo, as_list, compact
from .dump import dump
from .validate import ErrorType, ValidationErrors, validate
BAD_DATA_MESSAGE = os.getenv("BAD_DATA_MESSAGE", "Invalid data format")
@dataclass
class Operation:
body_schema: Optional[TypingInfo] = None
query_schema: Optional[TypingInfo] = None
response_schema: Optional[TypingInfo] = None
response: int = 200
[docs]class DataView:
"""Utility class for data with a valid :ref:`aio-openapi-schema`"""
operation: Operation = Operation()
[docs] def cleaned(
self,
schema: Any,
data: QueryType,
*,
multiple: bool = False,
strict: bool = True,
Error: Optional[type] = None,
) -> DataType:
"""Clean data using a given schema
:param schema: a valid :ref:`aio-openapi-schema` or an the name of an
attribute in :class:`.Operation`
:param data: data to validate and clean
:param multiple: multiple values for a given key are acceptable
:param strict: all required attributes in schema must be available
:param Error: optional :class:`.Exception` class
"""
type_info = self.get_schema(schema)
validated = validate(type_info, data, strict=strict, multiple=multiple)
if validated.errors:
if Error:
raise Error
elif schema == "path_schema":
raise web.HTTPNotFound
self.raise_validation_error(errors=validated.errors)
return validated.data
[docs] def dump(self, schema: Any, data: DataType) -> DataType:
"""Dump data using a given a valid :ref:`aio-openapi-schema`,
if the schema is `None` it returns the same `data` as the input.
:param schema: a schema or an the name of an attribute in :class:`.Operation`
:param data: data to clean and dump
"""
return data if schema is None else dump(self.get_schema(schema), data)
[docs] def get_schema(self, schema: Any = None) -> TypingInfo:
"""Get the :ref:`aio-openapi-schema`. If not found it raises an exception
:param schema: a schema or an the name of an attribute in :class:`.Operation`
"""
if isinstance(schema, str):
Schema = getattr(self.operation, schema, None)
else:
Schema = schema
if Schema is None:
Schema = getattr(self, str(schema), None)
if Schema is None:
raise web.HTTPNotImplemented
return cast(TypingInfo, TypingInfo.get(Schema))
[docs] def validation_error(
self, message: str = "", errors: Optional[ErrorType] = None
) -> Exception:
"""Create the validation exception used by :meth:`.raise_validation_error`"""
return ValidationErrors(self.as_errors(message, errors))
[docs] def raise_validation_error(
self, message: str = "", errors: Optional[ErrorType] = None
) -> NoReturn:
"""Raise an :class:`aiohttp.web.HTTPUnprocessableEntity`"""
raise self.validation_error(message, errors)
def raise_bad_data(
self, exc: Optional[Exception] = None, message: str = ""
) -> None:
if not message and exc:
raise exc from exc
raise TypeError(message or BAD_DATA_MESSAGE)
def as_errors(self, message: str = "", errors: Optional[ErrorType] = None) -> Dict:
if isinstance(errors, str):
message = cast(str, message or errors)
errors = None
return compact(message=message, errors=as_list(errors or ()))