Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def on_patch(self, req, resp):
if req.json is None or len(req.json) == 0:
raise falcon.HTTPBadRequest('BodyRequired', 'Expecting not empty JSON body')
include_prerelease = req.json.get('include_prerelease')
if include_prerelease is not None and not isinstance(include_prerelease, bool):
raise falcon.HTTPBadRequest('WrongValue', '"include_prerelease" have to be bool')
enabled = req.json.get('enabled')
if enabled is not None and not isinstance(enabled, bool):
raise falcon.HTTPBadRequest('WrongValue', '"enabled" have to be bool')
check_interval = req.json.get('interval')
if check_interval is not None and not isinstance(check_interval, int):
raise falcon.HTTPBadRequest('WrongValue', '"interval" have to be int')
if include_prerelease is not None:
if self.settings_manager.get_new_version_check_include_prerelease() != include_prerelease:
self.settings_manager.set_new_version_check_include_prerelease(include_prerelease)
def validate_confirm_password_reset(req, res, resource, params):
schema = {
'code': FIELDS['code'],
'password': FIELDS['password_create']
}
v = Validator(schema)
if not v.validate(req.context['data']):
raise falcon.HTTPBadRequest('Bad request', v.errors)
def _on_search_error(ex, _request, _response, _params):
raise falcon.HTTPBadRequest(title='Search error', description=str(ex))
def process_request(self, req, res):
if req.content_type == 'application/json':
body = req.stream.read().decode('utf-8')
try:
req.context['data'] = json.loads(body)
except ValueError:
message = "Request body is not valid 'application/json'"
raise falcon.HTTPBadRequest('Bad request', message)
def process_request(self, req, resp):
if req.content_length in (None, 0):
# Nothing to do
return
body = req.stream.read()
if not body:
falcon_log.exception('Empty request body')
raise falcon.HTTPBadRequest('Empty request body',
'A valid JSON document is required.')
try:
req.context['doc'] = ujson.loads(body.decode('utf-8'))
except (ValueError, UnicodeDecodeError):
falcon_log.exception('Malformed JSON, could not decode')
raise falcon.HTTPError(falcon.HTTP_753,
'Malformed JSON',
'Could not decode the request body. The '
'JSON was incorrect or not encoded as '
def on_post(self, req, resp):
model = models.UserScores(
username=req.media.get('username'),
company=req.media.get('company'),
score=req.media.get('score')
)
try:
model.save(self.db.session)
except IntegrityError:
raise falcon.HTTPBadRequest(
'Username exists',
'Could not create user due to username already existing'
)
resp.status = falcon.HTTP_201
resp.media = {
'id': model.id
}
"""
Return total number of results in a query.
:param queryset: queryset object from :func:`get_queryset`
:param totals: a list of dicts with aggregate function as key and column as value
:type totals: list
:return: dict with totals calculated in this query, ex. total_count with number of results
:rtype: dict
"""
if not totals:
return {}
for total in totals:
if len(total) > 1 or 'count' not in total or total['count'] is not None:
raise falcon.HTTPBadRequest('Invalid attribute', 'Only _count_ is supported in the _totals_ param')
return {'total_count': queryset.count()}
def process_request(self, req, resp):
if not req.content_length:
raise falcon.HTTPBadRequest('Not empty')
try:
req.context['data'] = json.loads(req.stream.read())
except (ValueError, UnicodeDecodeError):
raise falcon.HTTPBadRequest('Not valid JSON')
def _on_patch_by_pool_list(self, request, response, project_id,
flavor, pool_list):
if len(pool_list) == 0:
response.status = falcon.HTTP_400
response.location = request.path
raise falcon.HTTPBadRequest(_('Unable to create'), 'Bad Request')
# NOTE(gengchc2): If the flavor does not exist, return
try:
self._ctrl.get(flavor, project=project_id)
except errors.FlavorDoesNotExist as ex:
LOG.debug(ex)
raise wsgi_errors.HTTPNotFound(six.text_type(ex))
flavor_obj = {}
flavor_obj['name'] = flavor
# NOTE(gengchc2): Get the pools list with flavor.
pool_list_old = list(self._pools_ctrl.get_pools_by_flavor(
flavor=flavor_obj))
# NOTE(gengchc2): Check if the new pool in the pool_list exist.
try:
self._check_pools_exists(pool_list)
except errors.PoolDoesNotExist as ex:
context_json_str = self.create_context(alert_params)
app_template_count = session.execute('''
SELECT EXISTS (
SELECT 1 FROM
`plan_notification`
JOIN `template` ON `template`.`name` = `plan_notification`.`template`
JOIN `template_content` ON `template_content`.`template_id` = `template`.`id`
WHERE `plan_notification`.`plan_id` = :plan_id
AND `template_content`.`application_id` = :app_id
)
''', {'app_id': app['id'], 'plan_id': plan_id}).scalar()
if not app_template_count:
logger.warn('no plan template exists for this app')
raise HTTPBadRequest('No plan template actions exist for this app')
data = {
'plan_id': plan_id,
'created': datetime.datetime.utcnow(),
'application_id': app['id'],
'context': context_json_str,
'current_step': 0,
'active': True,
}
incident_id = session.execute(
'''INSERT INTO `incident` (`plan_id`, `created`, `context`,
`current_step`, `active`, `application_id`)
VALUES (:plan_id, :created, :context, 0, :active, :application_id)''',
data).lastrowid