Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def get(self):
"""Return the schema for get requests"""
return jsonify(self.api.__schema__), 200
def error_response(status_code, kom_error=None, error_msg=""):
# TODO: I think we need to unify these error types to make the API
# easier. Perhaps use protocol a error codes as they are, and
# add our own httpkom error codes on 1000 and above?
if kom_error is not None:
# The error should exist in the dictionary, but we use .get() to be safe
response = jsonify(error_code=_kom_servererror_to_error_code(kom_error),
error_status=str(kom_error),
error_type="protocol-a",
error_msg=str(kom_error.__class__.__name__))
else:
# We don't have any fancy error codes for httpkom yet.
response = jsonify(error_type="httpkom",
error_msg=error_msg)
response.status_code = status_code
return response
async def authors():
assert quart.current_app.db is db
await Author.create()
author = Author(name='at7h', email='g@test.com', password='xxxx')
await author.save()
author_list = await Author.select().all(False)
await Author.drop(safe=True)
return quart.jsonify(author_list)
async def sessions_who_am_i():
"""TODO
"""
try:
session_no = await g.ksession.who_am_i()
if await g.ksession.is_logged_in():
pers_no = await g.ksession.get_current_person_no()
person = await to_dict(KomPerson(pers_no), g.ksession)
else:
person = None
return jsonify(dict(person=person, session_no=session_no))
except komerror.Error as ex:
return error_response(400, kom_error=ex)
return (
jsonify(
{
"status": "Error",
"msg": "Missing query parameter: 'aggregation_unit'",
}
),
500,
)
# Check aggregation claims
if aggregation_unit not in aggregation_claims:
current_app.query_run_logger.error(
"SPATIAL_AGGREGATION_LEVEL_NOT_ALLOWED_BY_TOKEN", **log_dict
)
return (
jsonify(
{
"status": "Error",
"msg": f"'get_result' access denied for '{aggregation_unit}' "
f"aggregated result of '{query_kind}' query",
}
),
401,
)
else:
pass
else:
pass
current_app.query_run_logger.info("Authorised", **log_dict)
return await func(*args, **kwargs)
return (
jsonify(
{
"status": "Error",
"msg": f"'{claim_type}' access denied for '{query_kind}' query",
}
),
401,
)
# Check aggregation claims
elif aggregation_unit not in aggregation_claims:
current_app.query_run_logger.error(
"SPATIAL_AGGREGATION_LEVEL_NOT_ALLOWED_BY_TOKEN", **log_dict
)
return (
jsonify(
{
"status": "Error",
"msg": f"'{claim_type}' access denied for '{aggregation_unit}' "
f"aggregated result of '{query_kind}' query",
}
),
401,
)
else:
pass
current_app.query_run_logger.info("Authorised", **log_dict)
return await func(*args, **kwargs)
slots_dir.mkdir(parents=True, exist_ok=True)
# Write slots
total_length = 0
for slot_name, slot_values in slots_dict.items():
slot_file_path = slots_dir / slot_name
with open(slot_file_path, "w") as slot_file:
for value in slot_values:
value = value.strip()
print(value, file=slot_file)
total_length += len(value)
# Return length of written text
return str(total_length)
else:
return jsonify(slots_to_dict(slots_dir))
def handle_api_error(error):
error_id = handle_and_print_error()
app.logger.error(error.message)
data = error.to_dict()
data['error_id'] = error_id
response = jsonify(data)
response.status_code = error.status
return response
"""Function to handle validation errors
The constructor will register this function to handle a :exc:`~jsonschema.exceptions.ValidationError`
which is thrown by the :mod:`jsonschema` validation routines.
:param error: The exception that was raised
:return: Json message with the validation error
.. seealso::
:meth:`quart.Quart.register_error_handler`
Registering error handlers with quart
"""
LOGGER.error('request body validation failed, returning error: msg: %s, instance: %r',
error.message, error.instance)
return jsonify({
'message': 'Request Body failed validation',
'error': {
'msg': error.message,
'value': error.instance,
'schema': error.schema
}
}), HTTPStatus.BAD_REQUEST.value
async def index():
_servers = dict()
for i, server in enumerate(current_app.config['HTTPKOM_LYSKOM_SERVERS']):
_servers[server[0]] = Server(server[0], i, server[1], server[2], server[3])
servers = dict([ (s.id, s.to_dict()) for s in _servers.values() ])
return jsonify(servers)