def get_whois(item):
"""
Whois-like services based on utils/ips.py
"""
try:
item = {'rawItem': _get_deobfuscate_item(item)}
except AttributeError:
raise BadRequest('Invalid item')
try:
ip_addr, _, _ = _get_item_ip_hostname_url(item)
if not ip_addr:
raise BadRequest('Unable to get infos for this item')
except ValidationError:
raise BadRequest('Invalid item')
return {'ipCategory': utils.get_ip_network(ip_addr)}
python类BadRequest()的实例源码
def create(body):
""" Create provider
"""
if 'email' not in body:
raise BadRequest('Email field required')
if len(Provider.objects.filter(email=body['email'])) > 1:
raise BadRequest('Provider already exists')
try:
cat = None
if body.get('defaultCategory'):
cat = Category.objects.get(name=body['defaultCategory'])
body.pop('defaultCategory', None)
body = {k: v for k, v in body.iteritems() if k in PROVIDER_FIELDS}
provider = Provider.objects.create(defaultCategory=cat, **body)
return model_to_dict(provider)
except (FieldError, IntegrityError, ObjectDoesNotExist) as ex:
raise BadRequest(str(ex.message))
def update(prov, body):
""" Update provider infos
"""
try:
provider = Provider.objects.get(email=prov)
except (ObjectDoesNotExist, ValueError):
raise NotFound('Provider does not exist')
try:
body = {k: v for k, v in body.iteritems() if k in PROVIDER_FIELDS}
cat = None
if body.get('defaultCategory'):
cat = Category.objects.get(name=body['defaultCategory'])
body.pop('defaultCategory', None)
Provider.objects.filter(pk=provider.pk).update(defaultCategory=cat, **body)
provider = Provider.objects.get(pk=provider.pk)
except (FieldError, IntegrityError, ObjectDoesNotExist) as ex:
raise BadRequest(str(ex.message))
return model_to_dict(provider)
def add_tag(provider_email, body):
""" Add provider tag
"""
try:
tag = Tag.objects.get(**body)
provider = Provider.objects.get(email=provider_email)
if provider.__class__.__name__ != tag.tagType:
raise BadRequest('Invalid tag for provider')
provider.tags.add(tag)
provider.save()
except (KeyError, FieldError, IntegrityError, ObjectDoesNotExist, ValueError):
raise NotFound('Provider or tag not found')
return model_to_dict(provider)
def validate_body(schema_desc):
"""
Validate json body
"""
def real_decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
try:
body = request.get_json()
if not Schemas.get(func.__name__):
Schemas[func.__name__] = Schema(schema_desc, required=True)
Logger.debug(unicode('registering schema for %s' % (func.__name__)))
Schemas[func.__name__](body)
except (Invalid, MultipleInvalid) as ex:
Logger.error(unicode(ex))
msg = 'Missing or invalid field(s) in body, expecting {}'.format(schema_desc)
raise BadRequest(msg)
return func(*args, **kwargs)
return wrapper
return real_decorator
def auth():
"""
Check user/password and returns token if valid
"""
if settings.API.get('forwarded_host'):
try:
if not request.environ['HTTP_X_FORWARDED_HOST'] == settings.API['forwarded_host']:
raise BadRequest('Invalid HTTP_X_FORWARDED_HOST')
except KeyError:
raise BadRequest('Missing HTTP_X_FORWARDED_HOST')
body = request.get_json()
authenticated, ret = GeneralController.auth(body)
if authenticated:
return ret
else:
raise Unauthorized(ret)
def parse_protobuf(self, proto_type):
"""Parse the data into an instance of proto_type."""
if 'protobuf' not in self.environ.get('CONTENT_TYPE', ''):
raise BadRequest('Not a Protobuf request')
obj = proto_type()
try:
obj.ParseFromString(self.data)
except Exception:
raise BadRequest("Unable to parse Protobuf request")
# Fail if not all required fields are set
if self.protobuf_check_initialization and not obj.IsInitialized():
raise BadRequest("Partial Protobuf request")
return obj
def on_json_loading_failed(self, e):
"""Called if decoding of the JSON data failed. The return value of
this method is used by :meth:`get_json` when an error occurred. The
default implementation just raises a :class:`BadRequest` exception.
.. versionchanged:: 0.10
Removed buggy previous behavior of generating a random JSON
response. If you want that behavior back you can trivially
add it by subclassing.
.. versionadded:: 0.8
"""
ctx = _request_ctx_stack.top
if ctx is not None and ctx.app.config.get('DEBUG', False):
raise BadRequest('Failed to decode JSON object: {0}'.format(e))
raise BadRequest()
def parse_protobuf(self, proto_type):
"""Parse the data into an instance of proto_type."""
if 'protobuf' not in self.environ.get('CONTENT_TYPE', ''):
raise BadRequest('Not a Protobuf request')
obj = proto_type()
try:
obj.ParseFromString(self.data)
except Exception:
raise BadRequest("Unable to parse Protobuf request")
# Fail if not all required fields are set
if self.protobuf_check_initialization and not obj.IsInitialized():
raise BadRequest("Partial Protobuf request")
return obj
def test_trapping_of_bad_request_key_errors(self):
app = flask.Flask(__name__)
app.testing = True
@app.route('/fail')
def fail():
flask.request.form['missing_key']
c = app.test_client()
self.assert_equal(c.get('/fail').status_code, 400)
app.config['TRAP_BAD_REQUEST_ERRORS'] = True
c = app.test_client()
try:
c.get('/fail')
except KeyError as e:
self.assert_true(isinstance(e, BadRequest))
else:
self.fail('Expected exception')
def parse_protobuf(self, proto_type):
"""Parse the data into an instance of proto_type."""
if 'protobuf' not in self.environ.get('CONTENT_TYPE', ''):
raise BadRequest('Not a Protobuf request')
obj = proto_type()
try:
obj.ParseFromString(self.data)
except Exception:
raise BadRequest("Unable to parse Protobuf request")
# Fail if not all required fields are set
if self.protobuf_check_initialization and not obj.IsInitialized():
raise BadRequest("Partial Protobuf request")
return obj
def test_trapping_of_bad_request_key_errors(self):
app = flask.Flask(__name__)
app.testing = True
@app.route('/fail')
def fail():
flask.request.form['missing_key']
c = app.test_client()
self.assert_equal(c.get('/fail').status_code, 400)
app.config['TRAP_BAD_REQUEST_ERRORS'] = True
c = app.test_client()
try:
c.get('/fail')
except KeyError as e:
self.assert_true(isinstance(e, BadRequest))
else:
self.fail('Expected exception')
def parse_protobuf(self, proto_type):
"""Parse the data into an instance of proto_type."""
if 'protobuf' not in self.environ.get('CONTENT_TYPE', ''):
raise BadRequest('Not a Protobuf request')
obj = proto_type()
try:
obj.ParseFromString(self.data)
except Exception:
raise BadRequest("Unable to parse Protobuf request")
# Fail if not all required fields are set
if self.protobuf_check_initialization and not obj.IsInitialized():
raise BadRequest("Partial Protobuf request")
return obj
def parse_protobuf(self, proto_type):
"""Parse the data into an instance of proto_type."""
if 'protobuf' not in self.environ.get('CONTENT_TYPE', ''):
raise BadRequest('Not a Protobuf request')
obj = proto_type()
try:
obj.ParseFromString(self.data)
except Exception:
raise BadRequest("Unable to parse Protobuf request")
# Fail if not all required fields are set
if self.protobuf_check_initialization and not obj.IsInitialized():
raise BadRequest("Partial Protobuf request")
return obj
def on_json_loading_failed(self, e):
"""Called if decoding of the JSON data failed. The return value of
this method is used by :meth:`get_json` when an error occurred. The
default implementation just raises a :class:`BadRequest` exception.
.. versionchanged:: 0.10
Removed buggy previous behavior of generating a random JSON
response. If you want that behavior back you can trivially
add it by subclassing.
.. versionadded:: 0.8
"""
ctx = _request_ctx_stack.top
if ctx is not None and ctx.app.config.get('DEBUG', False):
raise BadRequest('Failed to decode JSON object: {0}'.format(e))
raise BadRequest()
def validation(schema):
"""function decorator"""
def dec(func):
def wrapper(self, *args, **kwargs):
try:
rawdata = request.data
enc = chardet.detect(rawdata)
data = rawdata.decode(enc['encoding'])
json_input = json.loads(data)
jsonschema.validate(json_input, schema)
json_input = prepare_input(json_input)
except:
raise BadRequest('JSON input not valid: {}'.format(format_exc()))
return func(self, json_input, *args, **kwargs)
return wrapper
return dec
def contains_payment(self, price, request_headers, **kwargs):
"""Validate the payment information received in the request headers.
Args:
price (int): The price the user must pay for the resource.
request_headers (dict): Headers sent by client with their request.
keyword args: Any other headers needed to verify payment.
Returns:
(bool): True if payment is valid,
False if no payment attached (402 initiation).
Raises:
BadRequest: If request is malformed.
"""
for method in self.allowed_methods:
if method.should_redeem(request_headers):
try:
return method.redeem_payment(price, request_headers, **kwargs)
except PaymentError as e:
raise BadRequest(str(e))
except Exception as e:
raise BadRequest(repr(e))
return False
def parse_protobuf(self, proto_type):
"""Parse the data into an instance of proto_type."""
if 'protobuf' not in self.environ.get('CONTENT_TYPE', ''):
raise BadRequest('Not a Protobuf request')
obj = proto_type()
try:
obj.ParseFromString(self.data)
except Exception:
raise BadRequest("Unable to parse Protobuf request")
# Fail if not all required fields are set
if self.protobuf_check_initialization and not obj.IsInitialized():
raise BadRequest("Partial Protobuf request")
return obj
def on_json_loading_failed(self, e):
"""Called if decoding of the JSON data failed. The return value of
this method is used by :meth:`get_json` when an error occurred. The
default implementation just raises a :class:`BadRequest` exception.
.. versionchanged:: 0.10
Removed buggy previous behavior of generating a random JSON
response. If you want that behavior back you can trivially
add it by subclassing.
.. versionadded:: 0.8
"""
ctx = _request_ctx_stack.top
if ctx is not None and ctx.app.config.get('DEBUG', False):
raise BadRequest('Failed to decode JSON object: {0}'.format(e))
raise BadRequest()
def parse_protobuf(self, proto_type):
"""Parse the data into an instance of proto_type."""
if 'protobuf' not in self.environ.get('CONTENT_TYPE', ''):
raise BadRequest('Not a Protobuf request')
obj = proto_type()
try:
obj.ParseFromString(self.data)
except Exception:
raise BadRequest("Unable to parse Protobuf request")
# Fail if not all required fields are set
if self.protobuf_check_initialization and not obj.IsInitialized():
raise BadRequest("Partial Protobuf request")
return obj
