How to use the connexion.resolver.Resolver function in connexion

To help you get started, we’ve selected a few connexion examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github zalando / connexion / tests / test_resolver.py View on Github external
def test_standard_resolve_x_router_controller():
    operation = Swagger2Operation(api=None,
                                  method='GET',
                                  path='endpoint',
                                  path_parameters=[],
                                  operation={
                                      'x-swagger-router-controller': 'fakeapi.hello',
                                      'operationId': 'post_greeting',
                                  },
                                  app_produces=['application/json'],
                                  app_consumes=['application/json'],
                                  app_security=[],
                                  security_definitions={},
                                  definitions={},
                                  parameter_definitions=PARAMETER_DEFINITIONS,
                                  resolver=Resolver())
    assert operation.operation_id == 'fakeapi.hello.post_greeting'
github zalando / connexion / tests / test_operation2.py View on Github external
def test_default(api):
    op_spec = make_operation(OPERATION4)
    op_spec['parameters'][1]['default'] = 1
    Swagger2Operation(
        api=api, method='GET', path='endpoint', path_parameters=[],
        operation=op_spec, app_produces=['application/json'],
        app_consumes=['application/json'], app_security=[],
        security_definitions={}, definitions=DEFINITIONS,
        parameter_definitions=PARAMETER_DEFINITIONS, resolver=Resolver()
    )
    op_spec = make_operation(OPERATION6, parameters=False)
    op_spec['parameters'][0]['default'] = {
        'keep_stacks': 1,
        'image_version': 'one',
        'senza_yaml': 'senza.yaml',
        'new_traffic': 100
    }
    Swagger2Operation(
        api=api, method='POST', path='endpoint', path_parameters=[],
        operation=op_spec, app_produces=['application/json'],
        app_consumes=['application/json'], app_security=[],
        security_definitions={}, definitions=DEFINITIONS,
        parameter_definitions={}, resolver=Resolver()
    )
github zalando / connexion / tests / test_operation2.py View on Github external
verify_oauth = mock.MagicMock(return_value=dummy)
    monkeypatch.setattr('connexion.operations.secure.verify_oauth', verify_oauth)

    op_spec = make_operation(OPERATION8)
    operation = Swagger2Operation(api=api,
                                  method='GET',
                                  path='endpoint',
                                  path_parameters=[],
                                  operation=op_spec,
                                  app_produces=['application/json'],
                                  app_consumes=['application/json'],
                                  app_security=[],
                                  security_definitions=SECURITY_DEFINITIONS_BOTH,
                                  definitions=DEFINITIONS,
                                  parameter_definitions=PARAMETER_DEFINITIONS,
                                  resolver=Resolver())
    assert isinstance(operation.function, types.FunctionType)

    security_decorator = operation.security_decorator
    assert security_decorator.func is verify_security
    assert len(security_decorator.args[0]) == 1
    assert security_decorator.args[0][0] is dummy
    assert security_decorator.args[1] == ['uid']
    call_args = verify_oauth.call_args[0]
    assert call_args[0] is math.ceil
    assert call_args[1] is validate_scope

    assert operation.method == 'GET'
    assert operation.produces == ['application/json']
    assert operation.consumes == ['application/json']
    assert operation.security == [{'oauth': ['uid']}]
    expected_body_schema = op_spec["parameters"][0]["schema"]
github zalando / connexion / tests / test_operation2.py View on Github external
def test_no_token_info(api):
    op_spec = make_operation(OPERATION1)
    operation = Swagger2Operation(api=api,
                                  method='GET',
                                  path='endpoint',
                                  path_parameters=[],
                                  operation=op_spec,
                                  app_produces=['application/json'],
                                  app_consumes=['application/json'],
                                  app_security=SECURITY_DEFINITIONS_WO_INFO,
                                  security_definitions=SECURITY_DEFINITIONS_WO_INFO,
                                  definitions=DEFINITIONS,
                                  parameter_definitions=PARAMETER_DEFINITIONS,
                                  resolver=Resolver())
    assert isinstance(operation.function, types.FunctionType)

    security_decorator = operation.security_decorator
    assert security_decorator.func is verify_security
    assert len(security_decorator.args[0]) == 0

    assert operation.method == 'GET'
    assert operation.produces == ['application/json']
    assert operation.consumes == ['application/json']
    assert operation.security == [{'oauth': ['uid']}]

    expected_body_schema = {'definitions': DEFINITIONS}
    expected_body_schema.update(DEFINITIONS["new_stack"])
    assert operation.body_schema == expected_body_schema
github zalando / connexion / tests / test_operation2.py View on Github external
def test_multi_body(api):
    with pytest.raises(InvalidSpecification) as exc_info:  # type: py.code.ExceptionInfo
        op_spec = make_operation(OPERATION2)
        operation = Swagger2Operation(api=api,
                                      method='GET',
                                      path='endpoint',
                                      path_parameters=[],
                                      operation=op_spec,
                                      app_produces=['application/json'],
                                      app_consumes=['application/json'],
                                      app_security=[],
                                      security_definitions={},
                                      definitions=DEFINITIONS,
                                      parameter_definitions=PARAMETER_DEFINITIONS,
                                      resolver=Resolver())
        operation.body_schema

    exception = exc_info.value
    assert str(exception) == "GET endpoint There can be one 'body' parameter at most"
    assert repr(exception) == """"""
github common-workflow-language / workflow-service / server.py View on Github external
import connexion
from connexion.resolver import Resolver
import connexion.utils as utils
import myapp

app = connexion.App(__name__, specification_dir='swagger/')
def rs(x):
    return utils.get_function_from_name("cwl_runner_wes." + x)

app.add_api('proto/workflow_execution.swagger.json', resolver=Resolver(rs))

app.run(port=8080)
github zalando / connexion / connexion / apps / abstract.py View on Github external
:param options: New style options dictionary.
        :type options: dict | None
        :param pass_context_arg_name: Name of argument in handler functions to pass request context to.
        :type pass_context_arg_name: str | None
        :param validator_map: map of validators
        :type validator_map: dict
        :rtype: AbstractAPI
        """
        # Turn the resolver_error code into a handler object
        self.resolver_error = resolver_error
        resolver_error_handler = None
        if self.resolver_error is not None:
            resolver_error_handler = self._resolver_error_handler

        resolver = resolver or self.resolver
        resolver = Resolver(resolver) if hasattr(resolver, '__call__') else resolver

        auth_all_paths = auth_all_paths if auth_all_paths is not None else self.auth_all_paths
        # TODO test if base_path starts with an / (if not none)
        arguments = arguments or dict()
        arguments = dict(self.arguments, **arguments)  # copy global arguments and update with api specfic

        if isinstance(specification, dict):
            specification = specification
        else:
            specification = self.specification_dir / specification

        api_options = self.options.extend(options)

        api = self.api_cls(specification,
                           base_path=base_path,
                           arguments=arguments,
github zalando / connexion / connexion / resolver.py View on Github external
def resolve_operation_id(self, operation):
        """
        Resolves the operationId using REST semantics unless explicitly configured in the spec

        :type operation: connexion.operations.AbstractOperation
        """
        if operation.operation_id:
            return Resolver.resolve_operation_id(self, operation)

        return self.resolve_operation_id_using_rest_semantics(operation)
github common-workflow-language / workflow-service / wes_service / __init__.py View on Github external
logging.info("Using config:")
    for n in args.__dict__:
        logging.info("  %s: %s", n, getattr(args, n))

    app = connexion.App(__name__)
    backend = utils.get_function_from_name(
        args.backend + ".create_backend")(args.opt)

    def rs(x):
        return getattr(backend, x)

    app.add_api(
        'openapi/workflow_execution_service.swagger.yaml',
        resolver=Resolver(rs))

    return app
github zalando / connexion / connexion / apis / abstract.py View on Github external
self.specification = Specification.load(specification, arguments=arguments)

        logger.debug('Read specification', extra={'spec': self.specification})

        self.options = ConnexionOptions(options, oas_version=self.specification.version)

        logger.debug('Options Loaded',
                     extra={'swagger_ui': self.options.openapi_console_ui_available,
                            'swagger_path': self.options.openapi_console_ui_from_dir,
                            'swagger_url': self.options.openapi_console_ui_path})

        self._set_base_path(base_path)

        logger.debug('Security Definitions: %s', self.specification.security_definitions)

        self.resolver = resolver or Resolver()

        logger.debug('Validate Responses: %s', str(validate_responses))
        self.validate_responses = validate_responses

        logger.debug('Strict Request Validation: %s', str(strict_validation))
        self.strict_validation = strict_validation

        logger.debug('Pythonic params: %s', str(pythonic_params))
        self.pythonic_params = pythonic_params

        logger.debug('pass_context_arg_name: %s', pass_context_arg_name)
        self.pass_context_arg_name = pass_context_arg_name

        if self.options.openapi_spec_available:
            self.add_openapi_json()
            self.add_openapi_yaml()