Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_load_yaml_from_docstring():
def f():
"""
Foo
bar
baz quux
---
herp: 1
derp: 2
"""
result = yaml_utils.load_yaml_from_docstring(f.__doc__)
assert result == {"herp": 1, "derp": 2}
description="FlowKit Analytical API",
license=dict(name="MPLv2", url="https://www.mozilla.org/en-US/MPL/2.0/"),
contact=dict(email="flowkit@flowminder.org"),
),
)
spec.components._schemas = flowmachine_query_schemas
spec.components.security_scheme(
"token", dict(type="http", scheme="bearer", bearerFormat="JWT")
)
# Loop over all the registered views and try to parse a yaml
# openapi spec from their docstrings
for rule in current_app.url_map.iter_rules():
try:
func = current_app.view_functions[rule.rule]
operations = yaml_utils.load_operations_from_docstring(func.__doc__)
if len(operations) > 0:
for method, op in operations.items():
op["operationId"] = f"{rule.rule}.{method}"
spec.path(
path=rule[
0
].rule, # In theory, could have multiple rules that match but will only be a single one here
operations=operations,
)
except Exception as e:
pass # Don't include in API
return spec
def operation_helper(
self, path=None, operations=None, methods=None, func=None, **kwargs
):
"""May mutate operations.
:param str path: Path to the resource
:param dict operations: A `dict` mapping HTTP methods to operation object. See
:param list methods: A list of methods registered for this path
"""
for method in methods:
yaml_doc_string = yaml_utils.load_operations_from_docstring(func.__doc__)
yaml_doc_string = yaml.safe_load(
str(yaml_doc_string).replace(
"{{self.__class__.__name__}}", self.__class__.__name__
)
)
if yaml_doc_string:
operation_spec = yaml_doc_string.get(method.lower(), {})
if self.get_method_permission(func.__name__):
operation_spec["security"] = [{"jwt": []}]
operations[method.lower()] = operation_spec
else:
operations[method.lower()] = {}
def path_helper(self, operations, view, **kwargs):
"""Path helper that allows passing a bottle view function."""
operations.update(yaml_utils.load_operations_from_docstring(view.__doc__))
app = kwargs.get('app', _default_app)
route = self._route_for_view(app, view)
return self.bottle_path_to_openapi(route.rule)
def path_helper(self, operations, resource, base_path=None, **kwargs):
"""Path helper that allows passing a Falcon resource instance."""
resource_uri_mapping = self._generate_resource_uri_mapping(self._app)
if resource not in resource_uri_mapping:
raise APISpecError("Could not find endpoint for resource {0}".format(resource))
operations.update(yaml_utils.load_operations_from_docstring(resource.__doc__) or {})
path = resource_uri_mapping[resource]["uri"]
if base_path is not None:
# make sure base_path accept either with or without leading slash
# swagger 2 usually come with leading slash but not in openapi 3.x.x
base_path = '/' + base_path.strip('/')
path = re.sub(base_path, "", path, 1)
methods = resource_uri_mapping[resource]["methods"]
for method_name, method_handler in methods.items():
docstring_yaml = yaml_utils.load_yaml_from_docstring(method_handler.__doc__)
operations[method_name] = docstring_yaml or dict()
return path
def path_helper(self, operations, view, **kwargs):
"""Path helper that allows passing a Flask view function."""
rule = self._rule_for_view(view)
operations.update(yaml_utils.load_operations_from_docstring(view.__doc__))
if hasattr(view, 'view_class') and issubclass(view.view_class, MethodView):
for method in view.methods:
if method in rule.methods:
method_name = method.lower()
method = getattr(view.view_class, method_name)
operations[method_name] = yaml_utils.load_yaml_from_docstring(method.__doc__)
path = self.flaskpath2openapi(rule.rule)
app_root = current_app.config['APPLICATION_ROOT'] or '/'
return urljoin(app_root.rstrip('/') + '/', path.lstrip('/'))
HTTP_METHODS = ("get", "put", "post", "delete", "options", "head", "patch")
handlers = [handler for handler in handlers if not
isinstance(handler, URLSpec) and len(handler) == 2]
for (endpoint, handler) in handlers:
for http_method in HTTP_METHODS:
method = getattr(handler, http_method)
if method.__doc__ is None:
continue
path_template = endpoint
path_template = re.sub('\(.*?\)\??', '{}', path_template)
path_template = re.sub('(?=[^/]{1}){}', '/{}', path_template)
path_parameters = path_template.count('{}')
spec = yaml_utils.load_yaml_from_docstring(method.__doc__)
parameters = list(inspect.signature(method).parameters.keys())[1:]
parameters = parameters + (path_parameters - len(parameters)) * ['',]
multiple_spec = spec.pop('multiple', {})
single_spec = spec.pop('single', {})
other_spec = spec
for subspec in [single_spec, other_spec]:
if subspec:
path = path_template.format(*parameters)
openapi_spec.path(
path=path,
operations={
http_method: subspec
}
)
if self.contact is not None:
info["contact"] = self.contact
if self.license is not None:
info["license"] = self.license
spec = APISpec(
title=self.title,
version=self.version,
openapi_version=self.openapi_version,
plugins=self.plugins,
info=info,
)
for route in self.app.router.routes:
if route.description:
operations = yaml_utils.load_operations_from_docstring(
route.description
)
spec.path(path=route.route, operations=operations)
for name, schema in self.schemas.items():
spec.components.schema(name, schema=schema)
return spec
def _extensions_from_handler(handler_class):
"""Returns extensions dict from handler docstring
:param handler_class:
:type handler_class: RequestHandler descendant
"""
return yaml_utils.load_yaml_from_docstring(handler_class.__doc__)
def _operations_from_methods(handler_class):
"""Generator of operations described in handler's http methods
:param handler_class:
:type handler_class: RequestHandler descendant
"""
for httpmethod in yaml_utils.PATH_KEYS:
method = getattr(handler_class, httpmethod)
operation_data = yaml_utils.load_yaml_from_docstring(method.__doc__)
if operation_data:
operation = {httpmethod: operation_data}
yield operation