Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def parse_object_doc(object):
"""
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
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)
safrs.DB.session.rollback()
safrs.log.error(message)
errors = dict(detail=message)
abort(status_code, errors=[errors])
if getattr(rel_query, "limit", False):
count = rel_query.count()
rel_query = rel_query.limit(limit)
if rel_query.count() >= get_config("BIG_QUERY_THRESHOLD"):
warning = 'Truncated result for relationship "{}",consider paginating this request'.format(rel_name)
safrs.log.warning(warning)
meta["warning"] = warning
items = rel_query.all()
else:
items = list(rel_query)
count = len(items)
meta["count"] = count
meta["limit"] = limit
data = [{"id": i.jsonapi_id, "type": i._s_type} for i in items]
else: # shouldn't happen!!
safrs.log.error("Unknown relationship direction for relationship {}: {}".format(rel_name, relationship.direction))
# add the relationship direction, for debugging purposes.
if is_debug():
# meta["direction"] = relationship.direction.name
pass
rel_link = urljoin(self_link, rel_name)
links = dict(self=rel_link)
rel_data = dict(links=links)
rel_data["data"] = data
if meta:
rel_data["meta"] = meta
relationships[rel_name] = rel_data
attributes = self.to_dict()
# extract the required fieldnames from the request args, eg. Users/?Users[name] => [name]
def __init__(self, message, status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value):
Exception.__init__(self)
self.status_code = status_code
if safrs.log.getEffectiveLevel() <= logging.DEBUG:
self.message = str(message)
else:
self.message += HIDDEN_LOG
safrs.log.debug(traceback.format_exc())
safrs.log.error("Generic Error: %s", message)
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
def _documented_api_method(method):
"""
:param method:
add metadata to the method:
REST_DOC: swagger documentation
HTTP_METHODS: the http methods (GET/POST/..) used to call this method
"""
USE_API_METHODS = get_config("USE_API_METHODS")
if USE_API_METHODS:
try:
api_doc = parse_object_doc(method)
except yaml.scanner.ScannerError:
safrs.log.error("Failed to parse documentation for %s", method)
setattr(method, REST_DOC, api_doc)
setattr(method, HTTP_METHODS, http_methods)
return method