Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
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
def method_wrapper(*args, **kwargs):
""" Wrap the method and perform error handling
:param *args:
:param **kwargs:
:return: result of the wrapped method
"""
try:
if not request.is_jsonapi and fun.__name__ not in ["get"]:
# reuire jsonapi content type for requests to these routes
raise GenericError('Unsupported Content-Type', 415)
result = fun(*args, **kwargs)
safrs.DB.session.commit()
return result
except (ValidationError, GenericError, NotFoundError) as exc:
safrs.log.exception(exc)
status_code = getattr(exc, "status_code")
message = exc.message
except werkzeug.exceptions.NotFound:
status_code = 404
message = "Not Found"
except Exception as exc:
status_code = getattr(exc, "status_code", 500)
traceback.print_exc()
safrs.DB.Model.__init__(self, **db_args)
except Exception as exc:
# OOPS .. things are going bad, this might happen using sqla automap
safrs.log.error("Failed to instantiate {}".format(self))
safrs.log.debug("db args: {}".format(db_args))
safrs.log.exception(exc)
safrs.DB.Model.__init__(self)
if self._s_auto_commit:
# Add the object to the database if specified by the class parameters
safrs.DB.session.add(self)
try:
safrs.DB.session.commit()
except sqlalchemy.exc.SQLAlchemyError as exc:
# Exception may arise when a DB constrained has been violated (e.g. duplicate key)
raise GenericError(exc)
# pylint: disable=not-callable
instance = self.SAFRSObject(**attributes)
if not instance._s_auto_commit:
#
# The item has not yet been added/commited by the SAFRSBase,
# in that case we have to do it ourselves
#
safrs.DB.session.add(instance)
try:
safrs.DB.session.commit()
except sqlalchemy.exc.SQLAlchemyError as exc:
# Exception may arise when a db constrained has been violated
# (e.g. duplicate key)
safrs.log.warning(str(exc))
raise GenericError(str(exc))
return instance
"""
pageable: True
description : Regex search all matching objects (works only in MySQL!!!)
args:
name: thom.*
"""
result = cls
for key, value in kwargs.items():
column = getattr(cls, key, None)
if not column:
raise ValidationError('Invalid Column "{}"'.format(key))
try:
result = result.query.filter(column.op("regexp")(value))
except Exception as exc:
raise GenericError("Failed to execute query {}".format(exc))
return SAFRSFormattedResponse(result.all())
if not isinstance(payload, dict):
raise ValidationError("Invalid Object Type")
data = payload.get("data")
relation = getattr(parent, self.rel_name)
obj_args = {self.parent_object_id: parent.jsonapi_id}
if isinstance(data, dict):
# https://jsonapi.org/format/#crud-updating-to-one-relationships
# server MUST respond to PATCH requests to a URL from a to-one relationship link as described below.
# The PATCH request MUST include a top-level member named data containing one of:
# a resource identifier object corresponding to the new related resource.
# null, to remove the relationship.
if self.SAFRSObject.relationship.direction != MANYTOONE:
raise GenericError("Provide a list o PATCH a TOMANY relationship")
child = self._parse_target_data(data)
setattr(parent, self.rel_name, child)
obj_args[self.child_object_id] = child.jsonapi_id
elif isinstance(data, list) and not self.SAFRSObject.relationship.direction == MANYTOONE:
"""
http://jsonapi.org/format/#crud-updating-to-many-relationships
If a client makes a PATCH request to a URL from a to-many relationship link,
the server MUST either completely replace every member of the relationship,
return an appropriate error response if some resourcescan not be found
or accessed, or return a 403 Forbidden response if complete
replacement is not allowed by the server.
"""
# first remove all items, then append the new items
# if the relationship has been configured with lazy="dynamic"
pageable: True
description : lookup column names
args:
column: value
"""
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