Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
required: bool = True,
converter: Optional[Callable] = None):
self.expected_type = expected_type
self.required = required
self.converter = converter
self.name = None
@abstractmethod
async def get_value(self, request: Request) -> T:
...
def __repr__(self):
return f'<{self.__class__.__name__} {self.expected_type} at {id(self)}>'
class MissingBodyError(BadRequest):
def __init__(self):
super().__init__('Missing body payload')
class MissingParameterError(BadRequest):
def __init__(self, name: str, source: str):
super().__init__(f'Missing parameter `{name}` from {source}')
class InvalidRequestBody(BadRequest):
def __init__(self, description: Optional[str] = 'Invalid body payload'):
super().__init__(description)
def _default_bool_converter(value: str):
if value in {'1', 'true'}:
return True
if value in {'0', 'false'}:
return False
# bad request: expected a bool value, but
# got something different that is not handled
raise BadRequest(f'Expected a bool value for a parameter, but got {value}.')
def __call__(self, request):
value = get_param(request, self.name)
if value is None or value == '':
if self.is_optional:
return None
raise BadRequest(f'missing parameter: {self.name}.')
if self.annotation is str:
return unwrap(value)
return value
# http://svn.tools.ietf.org/svn/wg/httpbis/specs/rfc7233.html#rfc.section.3.1
# A server must ignore a Range header field received with a request method other than GET
if request.method != 'GET':
return None
# NB: only the first Range request header is taken into consideration;
# if the HTTP contains several Range headers, only the first is used
range_header = request.get_first_header(b'range')
if not range_header:
return None
try:
value = Range.parse(range_header)
except InvalidRangeValue:
raise BadRequest('Invalid Range header')
else:
# An origin server must ignore a Range header field that contains a range unit it does not understand.
if value.unit != 'bytes':
return None
return value
@abstractmethod
async def get_value(self, request: Request) -> T:
...
def __repr__(self):
return f'<{self.__class__.__name__} {self.expected_type} at {id(self)}>'
class MissingBodyError(BadRequest):
def __init__(self):
super().__init__('Missing body payload')
class MissingParameterError(BadRequest):
def __init__(self, name: str, source: str):
super().__init__(f'Missing parameter `{name}` from {source}')
class InvalidRequestBody(BadRequest):
def __init__(self, description: Optional[str] = 'Invalid body payload'):
super().__init__(description)
class MissingConverterError(Exception):
def __init__(self, expected_type, binder_type):
super().__init__(f'A default converter for type `{str(expected_type)}` is not configured. '
f'Please define a converter method for this binder ({binder_type.__name__}).')
def parse_value(value, desired_type: type, param_name: str):
try:
if desired_type is bool:
return bool(int(value))
if desired_type in {int, float}:
return desired_type(value)
return value
except ValueError:
raise BadRequest(f'invalid parameter "{param_name}". '
f'The value cannot be parsed as {desired_type.__name__}.')
return f'<{self.__class__.__name__} {self.expected_type} at {id(self)}>'
class MissingBodyError(BadRequest):
def __init__(self):
super().__init__('Missing body payload')
class MissingParameterError(BadRequest):
def __init__(self, name: str, source: str):
super().__init__(f'Missing parameter `{name}` from {source}')
class InvalidRequestBody(BadRequest):
def __init__(self, description: Optional[str] = 'Invalid body payload'):
super().__init__(description)
class MissingConverterError(Exception):
def __init__(self, expected_type, binder_type):
super().__init__(f'A default converter for type `{str(expected_type)}` is not configured. '
f'Please define a converter method for this binder ({binder_type.__name__}).')
class FromBody(Binder):
_excluded_methods = {'GET', 'HEAD', 'TRACE'}
def __init__(self,
async def get_value(self, request: Request) -> T:
raw_value = self.get_raw_value(request)
try:
value = self.converter(raw_value)
except (ValueError, BadRequest):
raise BadRequest(f'Invalid value for parameter `{self.name}`; expected {self.expected_type}')
if value is None and self.required:
raise MissingParameterError(self.name, self.source_name)
if not self.required and self._empty_iterable(value):
return None
return value