Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@pytest.mark.parametrize("openapi_version", ("2.0", "3.0.0"))
def test_plugin_operation_helper_is_used(self, openapi_version):
spec = APISpec(
title="Swagger Petstore",
version="1.0.0",
openapi_version=openapi_version,
plugins=(self.test_plugin_factory(),),
)
spec.path("/path_2", operations={"post": {"responses": {"200": {}}}})
paths = get_paths(spec)
assert len(paths) == 1
assert paths["/path_2"] == {"post": {"responses": {"201": {}}}}
class TestPluginsOrder:
class OrderedPlugin(BasePlugin):
def __init__(self, index, output):
super(TestPluginsOrder.OrderedPlugin, self).__init__()
self.index = index
self.output = output
def path_helper(self, path, operations, **kwargs):
self.output.append("plugin_{}_path".format(self.index))
def operation_helper(self, path, operations, **kwargs):
self.output.append("plugin_{}_operations".format(self.index))
def test_plugins_order(self):
"""Test plugins execution order in APISpec.path
- All path helpers are called, then all operation helpers, then all response helpers.
- At each step, helpers are executed in the order the plugins are passed to APISpec.
def test_plugin_factory(return_none=False):
class TestPlugin(BasePlugin):
def schema_helper(self, name, definition, **kwargs):
if not return_none:
return {"properties": {"name": {"type": "string"}}}
def parameter_helper(self, parameter, **kwargs):
if not return_none:
return {"description": "some parameter"}
def response_helper(self, response, **kwargs):
if not return_none:
return {"description": "42"}
def path_helper(self, path, operations, parameters, **kwargs):
if not return_none:
if path == "/path_1":
operations.update({"get": {"responses": {"200": {}}}})
def test_api_extra_spec_plugins(self, app, schemas, openapi_version):
"""Test extra plugins can be passed to internal APISpec instance"""
app.config['OPENAPI_VERSION'] = openapi_version
class MyPlugin(apispec.BasePlugin):
def schema_helper(self, name, definition, **kwargs):
return {'dummy': 'whatever'}
api = Api(app, spec_kwargs={'extra_plugins': (MyPlugin(), )})
api.spec.components.schema('Pet', schema=schemas.DocSchema)
assert get_schemas(api.spec)['Pet']['dummy'] == 'whatever'
# from flask-restplus
RE_URL = re.compile(r'<(?:[^:<>]+:)?([^<>]+)>')
# From flask-apispec
DEFAULT_CONVERTER_MAPPING = {
werkzeug.routing.UnicodeConverter: ('string', None),
werkzeug.routing.IntegerConverter: ('integer', 'int32'),
werkzeug.routing.FloatConverter: ('number', 'float'),
werkzeug.routing.UUIDConverter: ('string', 'uuid'),
}
DEFAULT_TYPE = ('string', None)
class FlaskPlugin(BasePlugin):
"""Plugin to create OpenAPI paths from Flask rules"""
def __init__(self):
super().__init__()
self.converter_mapping = dict(DEFAULT_CONVERTER_MAPPING)
self.openapi_version = None
def init_spec(self, spec):
super().init_spec(spec)
self.openapi_version = spec.openapi_version
# From apispec
@staticmethod
def flaskpath2openapi(path):
"""Convert a Flask URL rule to an OpenAPI-compliant path.
except ImportError:
from urlparse import urljoin
from flask import current_app
from flask.views import MethodView
from apispec.compat import iteritems
from apispec import BasePlugin, yaml_utils
from apispec.exceptions import APISpecError
# from flask-restplus
RE_URL = re.compile(r'<(?:[^:<>]+:)?([^<>]+)>')
class FlaskPlugin(BasePlugin):
"""APISpec plugin for Flask"""
@staticmethod
def flaskpath2openapi(path):
"""Convert a Flask URL rule to an OpenAPI-compliant path.
:param str path: Flask path template.
"""
return RE_URL.sub(r'{\1}', path)
@staticmethod
def _rule_for_view(view):
view_funcs = current_app.view_functions
endpoint = None
for ept, view_func in iteritems(view_funcs):
if view_func == view:
import copy
import re
from apispec import BasePlugin, yaml_utils
from apispec.exceptions import APISpecError
class FalconPlugin(BasePlugin):
"""APISpec plugin for Falcon"""
def __init__(self, app):
super(FalconPlugin, self).__init__()
self._app = app
@staticmethod
def _generate_resource_uri_mapping(app):
routes_to_check = copy.copy(app._router._roots)
mapping = {}
for route in routes_to_check:
uri = route.uri_template
resource = route.resource
mapping[resource] = {
"uri": uri,
"""
from __future__ import absolute_import
import re
from bottle import default_app
from apispec import BasePlugin, yaml_utils
from apispec.exceptions import APISpecError
RE_URL = re.compile(r'<(?:[^:<>]+:)?([^<>]+)>')
_default_app = default_app()
class BottlePlugin(BasePlugin):
"""APISpec plugin for Bottle"""
@staticmethod
def bottle_path_to_openapi(path):
return RE_URL.sub(r'{\1}', path)
@staticmethod
def _route_for_view(app, view):
endpoint = None
for route in app.routes:
if route.callback == view:
endpoint = route
break
if not endpoint:
raise APISpecError('Could not find endpoint for route {0}'.format(view))
return endpoint
# {'/hello': {'get': {'description': 'Get a greeting',
# 'responses': {200: {'description': 'A greeting to the '
# 'client',
# 'schema': {'$ref': '#/definitions/Greeting'}}}}}}
"""
from __future__ import absolute_import
import inspect
import sys
from tornado.web import URLSpec
from apispec import BasePlugin, yaml_utils
from apispec.exceptions import APISpecError
class TornadoPlugin(BasePlugin):
"""APISpec plugin for Tornado"""
@staticmethod
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
from apispec import BasePlugin, APISpec
from apispec.exceptions import DuplicateComponentNameError
except ImportError:
raise ImportError("Missing the 'apispec' package. Try installing with 'dataclasses-jsonschema[apispec]'")
from . import T, SchemaType
def _schema_reference(name: str, schema_type: SchemaType) -> str:
if schema_type == SchemaType.SWAGGER_V2:
return f"#/definitions/{name}"
else:
return f"#/components/schemas/{name}"
class DataclassesPlugin(BasePlugin):
spec: APISpec
def init_spec(self, spec: APISpec):
super().init_spec(spec)
self.spec = spec
def resolve_schema_refs(self, data):
if "schema" in data:
data["schema"] = {"$ref": _schema_reference(data["schema"], self._schema_type)}
else:
for key in data:
if isinstance(data[key], dict):
self.resolve_schema_refs(data[key])
@property
def _schema_type(self) -> SchemaType:
from apispec import BasePlugin
from .common import resolve_schema_instance, make_schema_key, resolve_schema_cls
from .openapi import OpenAPIConverter
from .schema_resolver import SchemaResolver
def resolver(schema):
"""Default schema name resolver function that strips 'Schema' from the end of the class name."""
schema_cls = resolve_schema_cls(schema)
name = schema_cls.__name__
if name.endswith("Schema"):
return name[:-6] or name
return name
class MarshmallowPlugin(BasePlugin):
"""APISpec plugin for translating marshmallow schemas to OpenAPI/JSONSchema format.
:param callable schema_name_resolver: Callable to generate the schema definition name.
Receives the `Schema` class and returns the name to be used in refs within
the generated spec. When working with circular referencing this function
must must not return `None` for schemas in a circular reference chain.
Example: ::
from apispec.ext.marshmallow.common import resolve_schema_cls
def schema_name_resolver(schema):
schema_cls = resolve_schema_cls(schema)
return schema_cls.__name__
"""