Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
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.log.error(exc)
if safrs.log.getEffectiveLevel() > logging.DEBUG:
message = "Logging Disabled"
else:
message = str(exc)
args:
email: test email
"""
print(args)
print(kwargs)
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:
log.exception(exc)
return response
if custom_method and callable(custom_method):
decorated_method = custom_method
# keep the default method as parent_, e.g. parent_get
parent_method = getattr(cls, method_name)
cls.http_methods[method_name] = lambda *args, **kwargs: parent_method(*args, **kwargs)
# Apply custom decorators, specified as class variable list
try:
# Add swagger documentation
decorated_method = swagger_decorator(decorated_method)
except RecursionError:
# Got this error when exposing WP DB, TODO: investigate where it comes from
safrs.log.error("Failed to generate documentation for {} {} (Recursion Error)".format(cls, decorated_method))
# pylint: disable=broad-except
except Exception as exc:
safrs.log.exception(exc)
safrs.log.error("Failed to generate documentation for {}".format(decorated_method))
# Add cors
if cors_domain is not None:
decorated_method = cors.crossdomain(origin=cors_domain)(decorated_method)
# Add exception handling
decorated_method = http_method_decorator(decorated_method)
setattr(decorated_method, "SAFRSObject", cls.SAFRSObject)
for custom_decorator in getattr(cls.SAFRSObject, "custom_decorators", []):
decorated_method = custom_decorator(decorated_method)
setattr(cls, method_name, decorated_method)
return cls
)
properties["SAFRSObject"] = rel_object
properties["http_methods"] = safrs_object.http_methods
swagger_decorator = swagger_relationship_doc(rel_object, tags)
api_class = api_decorator(type(api_class_name, (SAFRSRestRelationshipAPI,), properties), swagger_decorator)
# Expose the relationship for the parent class:
# GET requests to this endpoint retrieve all item ids
safrs.log.info("Exposing relationship {} on {}, endpoint: {}".format(rel_name, url, endpoint))
self.add_resource(api_class, url, endpoint=endpoint, methods=["GET", "POST", "PATCH", "DELETE"])
try:
child_object_id = safrs_object.object_id
except Exception as exc:
safrs.log.exception(exc)
safrs.log.error("No object id for {}".format(safrs_object))
child_object_id = safrs_object.__name__
if safrs_object == parent_class:
# Avoid having duplicate argument ids in the url:
# append a 2 in case of a self-referencing relationship
# todo : test again
child_object_id += "2"
# Expose the relationship for , this lets us
# query and delete the class relationship properties for a given
# child id
# nb: this is not really documented in the jsonapi spec, remove??
url = (RELATIONSHIP_URL_FMT + "/").format(url_prefix, rel_name, child_object_id)
endpoint = "{}api.{}Id".format(url_prefix, rel_name)
# ENABLE_RELATIONSHIPS enables relationships to be included.
# This may slow down certain queries if the relationships are not properly configured!
ENABLE_RELATIONSHIPS = bool(os.environ.get("ENABLE_RELATIONSHIPS", safrs.SAFRS.ENABLE_RELATIONSHIPS))
if not ENABLE_RELATIONSHIPS:
ENABLE_RELATIONSHIPS = True
MAX_TABLE_COUNT = int(os.environ.get("MAX_TABLE_COUNT", safrs.SAFRS.MAX_TABLE_COUNT))
try:
result = current_app.config[option]
except (KeyError, RuntimeError):
#
result = getattr(safrs.SAFRS, option, None)
if result is None:
result = locals().get(option)
except Exception as exc:
safrs.log.exception(exc)
raise
return result