Integration with openapi-core #344
Replies: 1 comment 2 replies
-
|
Here is my final integration with openapi-core. import re
from typing import Any, TypeVar, Optional, Dict, List
from functools import partial
from box import Box, BoxList, exceptions as BoxExceptions
from blacksheep.server.bindings import Binder, BoundValue
from blacksheep.messages import Request
from blacksheep import FormPart
from openapi_core import Spec, unmarshal_request, validate_request
from openapi_core.validation.request.exceptions import (
InvalidRequestBody,
MissingRequiredParameter,
ParameterValidationError,
RequestBodyValidationError,
)
from openapi_core.validation.schemas.exceptions import InvalidSchemaValue
from openapi_core.templating.paths.exceptions import PathNotFound
from openapi_core.casting.schemas.exceptions import CastError
from openapi_core.deserializing.media_types.exceptions import MediaTypeDeserializeError
from openapi_core.datatypes import RequestParameters
from openapi_core.templating.media_types.exceptions import MediaTypeNotFound
from openapi_core.unmarshalling.request.unmarshallers import (
V30RequestBodyUnmarshaller,
V30RequestParametersValidator,
V30RequestSecurityValidator,
)
from jsonschema import ( # type: ignore
ValidationError as PayloadError,
RefResolver,
validators
)
from werkzeug.datastructures import ImmutableMultiDict, Headers
from . import exception
SchemaType = TypeVar('SchemaType', bound=Box)
spec = Spec.from_file_path('openapi.yml')
spec_content = spec.content()
PATH_PARAMETER_PATTERN = r"[<{](?:(?:string|int|float|path|uuid):)?(\w+)[>}]"
class BlackSheepOpenAPIRequest:
path_regex = re.compile(PATH_PARAMETER_PATTERN)
def __init__(self, request: Request):
if not isinstance(request, Request):
raise TypeError(f"'request' argument is not type of {Request}")
self.request = request
self.parameters = RequestParameters(
query=ImmutableMultiDict(self.request.query),
header=Headers({ h[0].decode(): h[1].decode() for h in request.headers }),
cookie=ImmutableMultiDict(self.request.cookies),
)
@property
def host_url(self) -> str:
return self.request.host
@property
def path(self) -> str:
return self.request.path
@property
def method(self) -> str:
return self.request.method.lower()
@property
def body(self) -> Optional[str]:
return getattr(self.request, 'body', '')
@property
def mimetype(self) -> str:
return self.request.content_type().decode().split(';')[0]
class FromJSONSchema(BoundValue[SchemaType]):
"""
Custom bound value that can be used to describe a list of objects validated
using a custom schema.
"""
class JSONSchemaBinder(Binder):
"""
Binder that handles a FromSchema, returning list of objects from a custom
schema.
"""
handle = FromJSONSchema
async def get_value(self, request: Request) -> Any:
validate(request, ['application/json'])
try:
body = await request.text()
setattr(request, 'body', body)
data = unmarshal_request(BlackSheepOpenAPIRequest(request), spec, cls=V30RequestBodyUnmarshaller)
value = data.body
if isinstance(value, list):
return BoxList(value)
return Box(value)
except (InvalidSchemaValue, InvalidRequestBody) as e:
raise exception.InvalidRequestException(str(e.__cause__))
except (MediaTypeDeserializeError, RequestBodyValidationError) as e:
raise exception.InvalidRequestException(str(e.__cause__))
except (MediaTypeNotFound, RequestBodyValidationError) as e:
raise exception.InvalidRequestException(str(e.__cause__))
def oneOf_numeric(numeric, TYPE_CHECKER, instance):
if isinstance(instance, list):
return False
try:
return isinstance(numeric(instance), numeric)
except ValueError:
typestr = 'int' if numeric == int else 'float'
raise PayloadError(f"{instance!r} is not of type {typestr}")
type_checker = validators.Draft7Validator.TYPE_CHECKER.redefine_many(
{
'integer': partial(oneOf_numeric, int),
'number': partial(oneOf_numeric, float),
}
)
validator_cls = validators.extend(validators.Draft7Validator, type_checker=type_checker)
def get_validator(schema):
resolver = RefResolver.from_schema(schema)
return validator_cls(schema, resolver=resolver)
class FromMultipartSchema(BoundValue[SchemaType]):
"""
Custom bound value that can be used to describe a list of objects validated
using a custom schema.
"""
class MultipartSchemaBinder(Binder):
"""
Binder that handles a FromSchema, returning list of objects from a custom
schema.
"""
handle = FromMultipartSchema
EXPECTED_CONTENT_TYPE = ['multipart/form-data', 'multipart/mixed']
async def get_value(self, request: Request) -> Any:
validate(request, self.EXPECTED_CONTENT_TYPE)
content_type = request.content_type().decode().split(';')[0]
form, schema = await get_form(request, content_type)
if not schema:
form, schema = await get_form(request, content_type)
if not schema:
raise exception.APIException(
f'No OpenAPI schema defined for content type to "{content_type}"'
)
files: Dict[str, List[str]] = {}
for name, value in form.items():
if isinstance(value, list) and isinstance(value[0], FormPart):
for file in value:
if name in files:
files[name].append(file.data.decode(errors='ignore'))
else:
files[name] = [(file.data.decode(errors='ignore'))]
vi = get_validator(schema)
try:
vi.validate({ **form, **files }, schema)
except PayloadError as e:
message = str(e).split('\n')[0]
raise exception.InvalidRequestException(message)
return Box(form)
class FromURLEncodedSchema(BoundValue[SchemaType]):
"""
Custom bound value that can be used to describe a list of objects validated
using a custom schema.
"""
def validate(request: Request, expected_content_type=None):
if expected_content_type:
content_type = request.content_type()
if content_type is None or content_type.decode().split(';')[0] not in expected_content_type:
raise exception.APIException(
f'Content-Type header should be one of {expected_content_type}. '
'If a form is being uploaded, make sure form is not empty.'
)
try:
validate_request(BlackSheepOpenAPIRequest(request), spec=spec, cls=V30RequestParametersValidator)
validate_request(BlackSheepOpenAPIRequest(request), spec=spec, cls=V30RequestSecurityValidator)
except PathNotFound as e:
raise exception.InvalidRequestException(
str(e),
code='PathNotFound'
)
except MissingRequiredParameter as e:
raise exception.InvalidRequestException(
str(e),
'MissingRequiredParameter'
)
except (CastError, ParameterValidationError) as e:
raise exception.InvalidRequestException(
str(e) + '. ' + str(e.__cause__),
'ParameterValidationError'
)
class URLEncodedSchemaBinder(Binder):
"""
Binder that handles a FromSchema, returning list of objects from a custom
schema.
"""
handle = FromURLEncodedSchema
EXPECTED_CONTENT_TYPE = 'application/x-www-form-urlencoded'
async def get_value(self, request: Request) -> Any:
validate(request, [self.EXPECTED_CONTENT_TYPE])
form, schema = await get_form(request, self.EXPECTED_CONTENT_TYPE)
if not schema:
raise exception.APIException(
f'No OpenAPI schema defined for content type to "{request.content_type().decode()}"'
)
vi = get_validator(schema)
try:
vi.validate(form, schema)
except PayloadError as e:
message = str(e).split('\n')[0]
raise exception.InvalidRequestException(message)
return Box(form)
def get_schema(request: Request, content_type: str):
openapi_schema = spec.content()
req = openapi_schema['paths'][request.path]['post']
if not req:
req = openapi_schema['paths'][request.path]['put']
if not req:
raise exception.APIException(
f'No OpenAPI schema defined for PUT/POST request for path {request.path}'
)
return req['requestBody']['content'][content_type]['schema']
async def get_form(request: Request, content_type: str):
try:
schema = get_schema(request, content_type)
form = await request.form()
return form, schema
except BoxExceptions.BoxKeyError:
return None, NoneWhy the customization ? Due to the way openapi-core is handling
Will try to use custom unmarshaller. But as framework is providing single/array values correctly for form data, using the lib for validating form requests only. But with application/json, no customization is needed. Use of |
Beta Was this translation helpful? Give feedback.
-
|
@techntools Thank You for this example, I love this! I think it deserves a dedicated package, and being documented in the docs site. I didn´t reply earlier because I had vacation this week and spent almost no time in front of my PC. I can help with the basic structure of a package and with adding a section to the documentation. When designing my Python libraries I always try to make them extendable / customizable to cover different use cases, like in this case. That is why for example I keep "private" methods to a minimum when coding. |
Beta Was this translation helpful? Give feedback.
-
|
Do suggest basic package structure. Please elaborate further about extending/customizing and keeping the private methods to a minimum. Few suggestions: Need to modify path_regex to include '/:param' for path parameters. We should use middleware. So that we can carry the validated data from one middleware to next. For example after validating request with a middleware, another middleware such as token auth middleware can use values from header. But how to pass data from one middleware to another ? Attaching custom attibute to request such as request.body ? |
Beta Was this translation helpful? Give feedback.
-
I managed to implement openapi-core protocol to validate requests using OpenAPI spec:
Here is the result I get from unmarshal_request()
OpenAPI specs:
But one thing failed to work is multipart requests. I also followed tests in the openapi-core package but still no luck. Its failing to deserialize.
Beta Was this translation helpful? Give feedback.
All reactions