Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
Parse the yaml description from the documented methods
"""
api_doc = {}
obj_doc = str(inspect.getdoc(object))
raw_doc = obj_doc.split(DOC_DELIMITER)[0]
yaml_doc = None
try:
yaml_doc = yaml.safe_load(raw_doc)
except (SyntaxError, yaml.scanner.ScannerError) as exc:
safrs.log.error("Failed to parse documentation {} ({})".format(raw_doc, exc))
yaml_doc = {"description": raw_doc}
except Exception as exc:
raise ValidationError("Failed to parse api doc")
if isinstance(yaml_doc, dict):
api_doc.update(yaml_doc)
return api_doc
del links["last"]
if first_args == self_args:
del links["first"]
if next_args == last_args:
del links["next"]
if prev_args == first_args:
del links["prev"]
if isinstance(object_query, (list, sqlalchemy.orm.collections.InstrumentedList)):
instances = object_query[page_offset : page_offset + limit]
else:
try:
res_query = object_query.offset(page_offset).limit(limit)
instances = res_query.all()
except OverflowError:
raise ValidationError("Pagination Overflow Error")
except Exception as exc:
raise GenericError("Pagination Error {}".format(exc))
return links, instances, count
# Delete the item from the many-to-one relationship
delattr(parent, self.rel_name)
else:
safrs.log.warning("child not in relation")
else:
# https://jsonapi.org/format/#crud-updating-to-many-relationships
children = data
if not isinstance(data, list) or not children:
raise ValidationError("Invalid data payload")
for child in children:
child_id = child.get("id", None)
child_type = child.get("type", None)
if not child_id or not child_type:
raise ValidationError("Invalid data payload", HTTPStatus.FORBIDDEN)
if child_type != self.target._s_type:
raise ValidationError("Invalid type", HTTPStatus.FORBIDDEN)
child = self.target.get_instance(child_id)
if child in relation:
relation.remove(child)
else:
safrs.log.warning("Item with id {} not in relation".format(child_id))
return {}, HTTPStatus.NO_CONTENT
result = cls
response = SAFRSFormattedResponse()
try:
instances = result.query
links, instances, count = paginate(instances)
data = [item for item in instances]
meta = {}
errors = None
response.response = jsonapi_format_response(data, meta, links, errors, count)
except Exception as exc:
raise GenericError("Failed to execute query {}".format(exc))
for key, value in kwargs.items():
column = getattr(cls, key, None)
if not column:
raise ValidationError('Invalid Column "{}"'.format(key))
try:
instances = result.query.filter(column.like(value + "%"))
links, instances, count = paginate(instances)
data = [item for item in instances]
meta = {}
errors = None
response.response = jsonapi_format_response(data, meta, links, errors, count)
except Exception as exc:
raise GenericError("Failed to execute query {}".format(exc))
return response
409 Conflict
A server MUST return 409 Conflict when processing a POST request
to create a resource with a client-generated ID that already exists.
A server MUST return 409 Conflict when processing a POST request
in which the resource object’s type is not among the type(s) that
constitute the collection represented by the endpoint.
A server SHOULD include error details and provide enough
information to recognize the source of the conflict.
"""
payload = request.get_jsonapi_payload()
method_name = payload.get("meta", {}).get("method", None)
id = kwargs.get(self.object_id, None)
if id is not None:
# POSTing to an instance isn't jsonapi-compliant (https://jsonapi.org/format/#crud-creating-client-ids)
raise ValidationError("POSTing to instance is not allowed {}".format(self), status_code=HTTPStatus.METHOD_NOT_ALLOWED)
# Create a new instance of the SAFRSObject
data = payload.get("data")
if data is None:
raise ValidationError("Request contains no data")
if isinstance(data, list):
# http://springbot.github.io/json-api/extensions/bulk/
if not request.is_bulk:
safrs.log.warning("Client sent a bulk POST but did not specify the bulk extension")
for item in data:
instance = self._create_instance(item)
resp_data = {}
location = None
else:
instance = self._create_instance(data)
# object_id is the endpoint parameter, for example "UserId" for a User SAFRSObject
def parse_args(self, **kwargs):
"""
Parse relationship args
An error is raised if the parent doesn't exist.
An error is raised if the child doesn't exist and the
"require_child" argument is set in kwargs,
:return: parent, child, relation
"""
parent_id = kwargs.get(self.parent_object_id, None)
if parent_id is None:
raise ValidationError("Invalid Parent Id")
parent = self.parent_class.get_instance(parent_id)
relation = getattr(parent, self.rel_name)
return parent, relation
201 :
description: Created
204 :
description : No Content
403:
description : Forbidden
404 :
description : Not Found
409 :
description : Conflict
---
Update the object with the specified id
"""
id = kwargs.get(self.object_id, None)
if id is None:
raise ValidationError("Invalid ID")
payload = request.get_jsonapi_payload()
if not isinstance(payload, dict):
raise ValidationError("Invalid Object Type")
data = payload.get("data")
if not data or not isinstance(data, dict):
raise ValidationError("Invalid Data Object")
# Check that the id in the body is equal to the id in the url
body_id = data.get("id", None)
if body_id is None:
raise ValidationError("No ID in body")
path_id = self.SAFRSObject.id_type.validate_id(id)
body_id = self.SAFRSObject.id_type.validate_id(body_id)
if path_id != body_id:
def process_bind_param(value, dialect):
try:
uuid.UUID(value, version=4)
except Exception as exc:
raise ValidationError("UUID Validation Error {} ({})".format(value, exc))
return value
def get_jsonapi_payload(self):
"""
:return: jsonapi request payload
"""
if not self.is_jsonapi:
safrs.log.warning('Invalid Media Type! "{}"'.format(self.content_type))
#raise GenericError('Unsupported Media Type', 415)
if self.method == "OPTIONS":
return None
if self.method not in HTTP_METHODS:
abort(500)
result = self.get_json()
if not isinstance(result, dict):
raise ValidationError("Invalid JSON Payload : {}".format(result))
return result
def _parse_target_data(self, child_data):
"""
Validate the jsonapi payload in child_data, which should contain "id" and "type" keys
"""
if not isinstance(child_data, dict):
raise ValidationError("Invalid data type {}".format(child_data))
child_id = child_data.get("id", None)
if child_id is None:
raise ValidationError("no child id {}".format(child_data))
child_type = child_data.get("type")
if not child_id:
raise ValidationError("Invalid id in data", HTTPStatus.FORBIDDEN)
if not child_type:
raise ValidationError("Invalid type in data", HTTPStatus.FORBIDDEN)
if child_type != self.target._s_type:
raise ValidationError("Invalid type {} != {}".format(child_type, self.target._s_type), HTTPStatus.FORBIDDEN)
child = self.target.get_instance(child_id)
if not child:
raise ValidationError("invalid child id {}".format(child_id))
return child