Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
:return: a sqla query object
"""
# First check if a filter= URL query parameter has been used
# the SAFRSObject should've implemented a filter method or
# overwritten the _s_filter method to implement custom filtering
filter_args = get_request_param("filter")
if filter_args:
safrs_object_filter = getattr(safrs_object, "filter", None)
if callable(safrs_object_filter):
result = safrs_object_filter(filter_args)
else:
result = safrs_object._s_filter(filter_args)
return result
expressions = []
filters = get_request_param("filters", {})
for attr_name, val in filters.items():
if not attr_name in safrs_object._s_jsonapi_attrs + ["id"]:
safrs.log.warning("Invalid filter {}".format(attr_name))
continue
attr = getattr(safrs_object, attr_name)
expressions.append((attr, val))
if isinstance(safrs_object, (list, sqlalchemy.orm.collections.InstrumentedList)):
# todo: filter properly
result = safrs_object
elif expressions:
expressions_ = [column.in_(val.split(",")) for column, val in expressions]
result = safrs_object._s_query.filter(*expressions_)
else:
result = safrs_object._s_query
offset is the number of records to offset by prior to returning resources
:param object_query: SQLAalchemy query object
:param SAFRSObject: optional
:return: links, instances, count
"""
def get_link(count, limit):
result = SAFRSObject._s_url if SAFRSObject else ""
result += "?" + "&".join(
["{}={}".format(k, v) for k, v in request.args.items()] + ["page[offset]={}&page[limit]={}".format(count, limit)]
)
return result
try:
page_offset = int(get_request_param("page_offset"))
limit = int(get_request_param("page_limit", get_config("MAX_PAGE_LIMIT")))
except ValueError:
raise ValidationError("Pagination Value Error")
page_base = int(page_offset / limit) * limit
# Counting may take > 1s for a table with millions of records, depending on the storage engine :|
# Make it configurable
# With mysql innodb we can use following to retrieve the count:
# select TABLE_ROWS from information_schema.TABLES where TABLE_NAME = 'TableName';
if isinstance(object_query, (list, sqlalchemy.orm.collections.InstrumentedList)):
count = len(object_query)
elif SAFRSObject is None: # for backwards compatibility, ie. when not passed as an arg to paginate()
count = object_query.count()
else:
count = SAFRSObject._s_count()
def jsonapi_filter(safrs_object):
"""
https://jsonapi.org/recommendations/#filtering
Apply the request.args filters to the object
:param safrs_object:
:return: a sqla query object
"""
# First check if a filter= URL query parameter has been used
# the SAFRSObject should've implemented a filter method or
# overwritten the _s_filter method to implement custom filtering
filter_args = get_request_param("filter")
if filter_args:
safrs_object_filter = getattr(safrs_object, "filter", None)
if callable(safrs_object_filter):
result = safrs_object_filter(filter_args)
else:
result = safrs_object._s_filter(filter_args)
return result
expressions = []
filters = get_request_param("filters", {})
for attr_name, val in filters.items():
if not attr_name in safrs_object._s_jsonapi_attrs + ["id"]:
safrs.log.warning("Invalid filter {}".format(attr_name))
continue
attr = getattr(safrs_object, attr_name)
expressions.append((attr, val))
def jsonapi_format_response(data=None, meta=None, links=None, errors=None, count=None, include=None):
"""
Create a response dict according to the json:api schema spec
:param data : the objects that will be serialized
:return: jsonapi formatted dictionary
"""
limit = get_request_param("page_limit", get_config("MAX_PAGE_LIMIT"))
try:
limit = int(limit)
except ValueError:
raise ValidationError("page[limit] error")
if meta is None:
meta = {}
if include is None:
include = request.args.get("include", safrs.SAFRS.DEFAULT_INCLUDED)
meta["limit"] = limit
meta["count"] = count
jsonapi = dict(version="1.0")
included = list(get_included(data, limit, include=include))
"""if count >= 0:
:param object_query: SQLAalchemy query object
:param SAFRSObject: optional
:return: links, instances, count
"""
def get_link(count, limit):
result = SAFRSObject._s_url if SAFRSObject else ""
result += "?" + "&".join(
["{}={}".format(k, v) for k, v in request.args.items()] + ["page[offset]={}&page[limit]={}".format(count, limit)]
)
return result
try:
page_offset = int(get_request_param("page_offset"))
limit = int(get_request_param("page_limit", get_config("MAX_PAGE_LIMIT")))
except ValueError:
raise ValidationError("Pagination Value Error")
page_base = int(page_offset / limit) * limit
# Counting may take > 1s for a table with millions of records, depending on the storage engine :|
# Make it configurable
# With mysql innodb we can use following to retrieve the count:
# select TABLE_ROWS from information_schema.TABLES where TABLE_NAME = 'TableName';
if isinstance(object_query, (list, sqlalchemy.orm.collections.InstrumentedList)):
count = len(object_query)
elif SAFRSObject is None: # for backwards compatibility, ie. when not passed as an arg to paginate()
count = object_query.count()
else:
count = SAFRSObject._s_count()
if count is None: