Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
# Data is optional, it's also really slow for large sets!!!!!
rel_query = getattr(self, rel_name)
limit = request.page_limit
if not get_config("ENABLE_RELATIONSHIPS"):
meta["warning"] = "ENABLE_RELATIONSHIPS set to false in config.py"
elif rel_query:
# todo: chekc if lazy=dynamic
# In order to work with the relationship as with Query,
# you need to configure it with lazy='dynamic'
# "limit" may not be possible !
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)
# convert single instance to a list so we can generically add the includes
included = [included]
elif isinstance(included, sqlalchemy.orm.collections.InstrumentedList):
pass
elif not included or included in result:
continue
try:
# This works on sqlalchemy.orm.dynamic.AppenderBaseQuery
included = included[:limit]
except Exception as exc:
safrs.log.debug("Failed to add included for {} (included: {} - {}): {}".format(relationship, type(included), included, exc))
try:
result = result.union(included)
except Exception as exc:
safrs.log.warning(
"Failed to unionize included for {} (included: {} - {}): {}".format(relationship, type(included), included, exc)
)
result.add(included)
if INCLUDE_ALL in includes:
for nested_included in [get_included(result, limit, level=level + 1) for obj in result]:
# Removed recursion with get_included(result, limit, INCLUDE_ALL)
result = result.union(nested_included)
elif nested_rel:
for nested_included in [get_included(result, limit, nested_rel, level=level + 1) for obj in result]:
result = result.union(nested_included)
return result
def __init__(self, message="", status_code=HTTPStatus.NOT_FOUND.value):
Exception.__init__(self)
self.status_code = status_code
if safrs.log.getEffectiveLevel() <= logging.DEBUG:
self.message += message
safrs.log.error("Not found: %s", message)
else:
self.message += HIDDEN_LOG
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
elif isinstance(object, dict):
for k, v in object.items():
if isinstance(v, str):
properties[k] = {"example": v, "type": "string"}
elif isinstance(v, int):
properties[k] = {"example": v, "type": "integer"}
elif isinstance(v, (dict, list)):
if isinstance(v, dict):
v = encode_schema(v)
properties[k] = {"example": v, "type": "string"}
elif v is None:
properties[k] = {"example": "", "type": "string"}
else: # isinstance(object, datetime.datetime):
properties = {"example": str(k), "type": "string"}
safrs.log.warning("Invalid schema object type %s", type(object))
else:
raise ValidationError("Invalid schema object type {}".format(type(object)))
# generate a unique name to be used as a reference
idx = _references.count(name)
if idx:
name = name + str(idx)
# name = urllib.parse.quote(name)
_references.append(name)
return SchemaClassFactory(name, properties)
def get_legacy(param, default=0):
""" Retrieve request parameters
Used for backwards compatibility (with safrs < 2.x)
:param param: parameter to retrieve
:param default:
:return: prequest parameter or None
:rtype: Boolean
"""
result = getattr(request, param, None)
if result is None:
safrs.log.error('Legacy Request parameter "{}", consider upgrading'.format(param))
result = default
return result
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)
safrs.log.info("Exposing {} relationship {} on {}, endpoint: {}".format(parent_name, rel_name, url, endpoint))