Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
# This makes things really complicated!!!
# TODO: simplify this by creating a proper superclass
"custom_decorators": decorators,
"parent": parent_class,
"_target": safrs_object,
},
)
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
# 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 _s_filter(cls, filter_args):
"""
Apply a filter to this model
:param filter_args: filter to apply, passed as a request URL parameter
:return: sqla query object
"""
safrs.log.info("_s_filter args: {}".format(filter_args))
safrs.log.info("override the {}._s_filter classmethod to implement your filtering".format(cls.__name__))
return cls._s_query
if cls.relationship.direction in (ONETOMANY, MANYTOMANY):
data = [data]
rel_del_schema = schema_from_object("{}_Relationship".format(class_name), {"data": data})
parameters.append(
{
"name": "{} body".format(class_name),
"in": "body",
"description": "{} POST model".format(class_name),
"schema": rel_del_schema,
"required": True,
}
)
else:
# one of 'options', 'head', 'patch'
safrs.log.info('no documentation for "{}" '.format(http_method))
doc["parameters"] = parameters
if doc.get("responses"):
responses.update({str(val): desc for val, desc in doc["responses"].items()})
doc["responses"] = responses
direction = "to-many" if cls.relationship.direction in (ONETOMANY, MANYTOMANY) else "to-one"
parent_name = parent_class.__name__ # to be used by f-string
child_name = child_class.__name__ # to be used by f-string
apply_fstring(doc, locals())
return swagger.doc(doc)(func)
def _s_filter(cls, filter_args):
"""
Apply a filter to this model
:param filter_args: filter to apply, passed as a request URL parameter
:return: sqla query object
"""
safrs.log.info("_s_filter args: {}".format(filter_args))
safrs.log.info("override the {}._s_filter classmethod to implement your filtering".format(cls.__name__))
return cls._s_query
self.expose_methods(url_prefix, tags=tags)
RESOURCE_URL_FMT = get_config("RESOURCE_URL_FMT")
url = RESOURCE_URL_FMT.format(url_prefix, safrs_object._s_collection_name)
endpoint = safrs_object.get_endpoint()
properties["SAFRSObject"] = safrs_object
properties["http_methods"] = safrs_object.http_methods
swagger_decorator = swagger_doc(safrs_object)
# Create the class and decorate it
api_class = api_decorator(type(api_class_name, (SAFRSRestAPI,), properties), swagger_decorator)
# Expose the collection
safrs.log.info("Exposing {} on {}, endpoint: {}".format(safrs_object._s_type, url, endpoint))
self.add_resource(api_class, url, endpoint=endpoint, methods=["GET", "POST"])
INSTANCE_URL_FMT = get_config("INSTANCE_URL_FMT")
url = INSTANCE_URL_FMT.format(url_prefix, safrs_object._s_collection_name, safrs_object.__name__)
endpoint = safrs_object.get_endpoint(type="instance")
# Expose the instances
safrs.log.info("Exposing {} instances on {}, endpoint: {}".format(safrs_object._s_collection_name, url, endpoint))
self.add_resource(api_class, url, endpoint=endpoint)
object_doc = parse_object_doc(safrs_object)
object_doc["name"] = safrs_object._s_collection_name
self._swagger_object["tags"].append(object_doc)
for relationship in safrs_object._s_relationships:
self.expose_relationship(relationship, url, tags=tags)