Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
data = {"id": rel_item.jsonapi_id, "type": rel_item._s_type}
elif relationship.direction in (ONETOMANY, MANYTOMANY):
# 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
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:
count = object_query.count()
if count > get_config("MAX_TABLE_COUNT"):
safrs.log.warning("Large table count detected, performance may be impacted, consider using '_s_count'")
first_args = (0, limit)
last_args = (int(int(count / limit) * limit), limit) # round down
self_args = (page_base if page_base <= last_args[0] else last_args[0], limit)
next_args = (page_offset + limit, limit) if page_offset + limit <= last_args[0] else last_args
prev_args = (page_offset - limit, limit) if page_offset > limit else first_args
links = {
"first": get_link(*first_args),
"self": get_link(page_offset, limit), # cfr. request.url
"last": get_link(*last_args),
"prev": get_link(*prev_args),
"next": get_link(*next_args),
}
properties = {}
# if the relationship is not an sql sqlalchemy.orm.relationships.RelationshipProperty instance
# then we should have defined the _target
safrs_object = relationship.mapper.class_
rel_name = relationship.key
parent_class = relationship.parent.class_
parent_name = parent_class.__name__
# Name of the endpoint class
RELATIONSHIP_URL_FMT = get_config("RELATIONSHIP_URL_FMT")
api_class_name = API_CLASSNAME_FMT.format(parent_name, rel_name)
url = RELATIONSHIP_URL_FMT.format(url_prefix, rel_name)
ENDPOINT_FMT = get_config("ENDPOINT_FMT")
endpoint = ENDPOINT_FMT.format(url_prefix, rel_name)
# Relationship object
decorators = getattr(parent_class, "custom_decorators", []) + getattr(parent_class, "decorators", [])
rel_object = type(
"{}.{}".format(parent_name, rel_name), # Name of the class we're creating here
(SAFRSRelationshipObject,),
{
"relationship": relationship,
# Merge the relationship decorators from the classes
# This makes things really complicated!!!
# TODO: simplify this by creating a proper superclass
"custom_decorators": decorators,
"parent": parent_class,
"_target": safrs_object,
},
def get_endpoint(cls, url_prefix=None, type=None):
"""
:param url_prefix: URL prefix used by the app
:param type: endpoint type, e.g. "instance"
:return: the API endpoint
:rtype: str
"""
if url_prefix is None:
url_prefix = cls.url_prefix
if type == "instance":
INSTANCE_ENDPOINT_FMT = get_config("INSTANCE_ENDPOINT_FMT")
endpoint = INSTANCE_ENDPOINT_FMT.format(url_prefix, cls._s_type)
else: # type = 'collection'
endpoint = "{}api.{}".format(url_prefix, cls._s_type)
return endpoint
def api_decorator(cls, swagger_decorator):
""" Decorator for the API views:
- add swagger documentation ( swagger_decorator )
- add cors
- add generic exception handling
We couldn't use inheritance because the rest method decorator
references the cls.SAFRSObject which isn't known
"""
cors_domain = get_config("cors_domain")
cls.http_methods = {} # holds overridden http methods, note: cls also has the "methods" set, but it's not related to this
for method_name in ["get", "post", "delete", "patch", "put", "options"]: # HTTP methods
method = getattr(cls, method_name, None)
if not method:
continue
decorated_method = method
# if the SAFRSObject has a custom http method decorator, use it
# e.g. SAFRSObject.get
custom_method = getattr(cls.SAFRSObject, method_name, None)
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)
for api_method in api_methods:
method_name = api_method.__name__
api_method_class_name = "method_{}_{}".format(safrs_object._s_class_name, method_name)
if (
isinstance(safrs_object.__dict__.get(method_name, None), (classmethod, staticmethod))
or getattr(api_method, "__self__", None) is safrs_object
):
# method is a classmethod or static method, make it available at the class level
CLASSMETHOD_URL_FMT = get_config("CLASSMETHOD_URL_FMT")
url = CLASSMETHOD_URL_FMT.format(url_prefix, safrs_object._s_collection_name, method_name)
else:
# expose the method at the instance level
INSTANCEMETHOD_URL_FMT = get_config("INSTANCEMETHOD_URL_FMT")
url = INSTANCEMETHOD_URL_FMT.format(url_prefix, safrs_object._s_collection_name, safrs_object.object_id, method_name)
ENDPOINT_FMT = get_config("ENDPOINT_FMT")
endpoint = ENDPOINT_FMT.format(url_prefix, safrs_object._s_collection_name + "." + method_name)
swagger_decorator = swagger_method_doc(safrs_object, method_name, tags)
properties = {"SAFRSObject": safrs_object, "method_name": method_name}
properties["http_methods"] = safrs_object.http_methods
api_class = api_decorator(type(api_method_class_name, (SAFRSJSONRPCAPI,), properties), swagger_decorator)
meth_name = safrs_object._s_class_name + "." + api_method.__name__
safrs.log.info("Exposing method {} on {}, endpoint: {}".format(meth_name, url, endpoint))
self.add_resource(api_class, url, endpoint=endpoint, methods=get_http_methods(api_method), jsonapi_rpc=True)
:param url_prefix: api url prefix
:param tags: swagger tags
:return: None
"""
safrs_object = self.safrs_object
api_methods = safrs_object._s_get_jsonapi_rpc_methods()
for api_method in api_methods:
method_name = api_method.__name__
api_method_class_name = "method_{}_{}".format(safrs_object._s_class_name, method_name)
if (
isinstance(safrs_object.__dict__.get(method_name, None), (classmethod, staticmethod))
or getattr(api_method, "__self__", None) is safrs_object
):
# method is a classmethod or static method, make it available at the class level
CLASSMETHOD_URL_FMT = get_config("CLASSMETHOD_URL_FMT")
url = CLASSMETHOD_URL_FMT.format(url_prefix, safrs_object._s_collection_name, method_name)
else:
# expose the method at the instance level
INSTANCEMETHOD_URL_FMT = get_config("INSTANCEMETHOD_URL_FMT")
url = INSTANCEMETHOD_URL_FMT.format(url_prefix, safrs_object._s_collection_name, safrs_object.object_id, method_name)
ENDPOINT_FMT = get_config("ENDPOINT_FMT")
endpoint = ENDPOINT_FMT.format(url_prefix, safrs_object._s_collection_name + "." + method_name)
swagger_decorator = swagger_method_doc(safrs_object, method_name, tags)
properties = {"SAFRSObject": safrs_object, "method_name": method_name}
properties["http_methods"] = safrs_object.http_methods
api_class = api_decorator(type(api_method_class_name, (SAFRSJSONRPCAPI,), properties), swagger_decorator)
meth_name = safrs_object._s_class_name + "." + api_method.__name__
safrs.log.info("Exposing method {} on {}, endpoint: {}".format(meth_name, url, endpoint))
self.add_resource(api_class, url, endpoint=endpoint, methods=get_http_methods(api_method), jsonapi_rpc=True)
def add_resource(self, resource, *urls, **kwargs):
"""
This method is partly copied from flask_restful_swagger_2/__init__.py
I changed it because we don't need path id examples when
there's no {id} in the path.
We also have to filter out the unwanted parameters
"""
#
# This function has grown out of proportion and should be refactored, disable lint warning for now
#
# pylint: disable=too-many-nested-blocks,too-many-statements, too-many-locals
#
relationship = kwargs.pop("relationship", False) # relationship object
SAFRS_INSTANCE_SUFFIX = get_config("OBJECT_ID_SUFFIX") + "}"
path_item = collections.OrderedDict()
definitions = {}
resource_methods = kwargs.get("methods", HTTP_METHODS)
kwargs.pop("safrs_object", None)
is_jsonapi_rpc = kwargs.pop("jsonapi_rpc", False) # check if the exposed method is a jsonapi_rpc method
deprecated = kwargs.pop("deprecated", False) # TBD!!
for method in self.get_resource_methods(resource):
if deprecated:
continue
if not method.upper() in resource_methods:
continue
f = getattr(resource, method, None)
if not f:
continue