python类API的实例源码

def add_handlers(self, room_handler=None, transaction_handler=None,
                     user_handler=None):
        """Adds routes to Application that use specified handlers."""
        # Add all the normal matrix API routes
        if room_handler:
            room = resources.Room(room_handler,
                                  self.Api)
            self.add_route("/rooms/{room_alias}", room)
        if transaction_handler:
            transaction = resources.Transaction(transaction_handler,
                                                self.Api)
            self.add_route("/transactions/{txn_id}", transaction)
        if user_handler:
            user = resources.User(user_handler,
                                  self.Api)
            self.add_route("/users/{user_id}", user)

def test_can_pass_additional_params_to_add_route():

    check = []

    class CustomRouter(object):
        def add_route(self, uri_template, method_map, resource, name):
            self._index = {name: uri_template}
            check.append(name)

        def find(self, uri):
            pass

    app = falcon.API(router=CustomRouter())
    app.add_route('/test', 'resource', name='my-url-name')

    assert len(check) == 1
    assert 'my-url-name' in check

    # Also as arg.
    app.add_route('/test', 'resource', 'my-url-name-arg')

    assert len(check) == 2
    assert 'my-url-name-arg' in check

def test_raise_status_in_process_request(self):
        """ Make sure we can raise status from middleware process request """
        class TestMiddleware:
            def process_request(self, req, resp):
                raise HTTPStatus(falcon.HTTP_200,
                                 headers={'X-Failed': 'False'},
                                 body='Pass')

        app = falcon.API(middleware=TestMiddleware())
        app.add_route('/status', TestHookResource())
        client = testing.TestClient(app)

        response = client.simulate_request(path='/status', method='GET')
        assert response.status == falcon.HTTP_200
        assert response.headers['x-failed'] == 'False'
        assert response.text == 'Pass'

def test_raise_status_in_process_resource(self):
        """ Make sure we can raise status from middleware process resource """
        class TestMiddleware:
            def process_resource(self, req, resp, resource, params):
                raise HTTPStatus(falcon.HTTP_200,
                                 headers={'X-Failed': 'False'},
                                 body='Pass')

        app = falcon.API(middleware=TestMiddleware())
        app.add_route('/status', TestHookResource())
        client = testing.TestClient(app)

        response = client.simulate_request(path='/status', method='GET')
        assert response.status == falcon.HTTP_200
        assert response.headers['x-failed'] == 'False'
        assert response.text == 'Pass'

def test_log_get_request(self):
        """Test that Log middleware is executed"""
        global context
        app = falcon.API(middleware=[RequestTimeMiddleware()])

        app.add_route(TEST_ROUTE, MiddlewareClassResource())
        client = testing.TestClient(app)

        response = client.simulate_request(path=TEST_ROUTE)
        assert _EXPECTED_BODY == response.json
        assert response.status == falcon.HTTP_200

        assert 'start_time' in context
        assert 'mid_time' in context
        assert 'end_time' in context
        assert context['mid_time'] >= context['start_time'], \
            'process_resource not executed after request'
        assert context['end_time'] >= context['start_time'], \
            'process_response not executed after request'

        assert context['req_succeeded']

def test_generate_trans_id_and_time_with_request(self):
        global context
        app = falcon.API(middleware=[TransactionIdMiddleware(),
                                     RequestTimeMiddleware()])

        app.add_route(TEST_ROUTE, MiddlewareClassResource())
        client = testing.TestClient(app)

        response = client.simulate_request(path=TEST_ROUTE)
        assert _EXPECTED_BODY == response.json
        assert response.status == falcon.HTTP_200
        assert 'transaction_id' in context
        assert 'unique-req-id' == context['transaction_id']
        assert 'start_time' in context
        assert 'mid_time' in context
        assert 'end_time' in context
        assert context['mid_time'] >= context['start_time'], \
            'process_resource not executed after request'
        assert context['end_time'] >= context['start_time'], \
            'process_response not executed after request'

def test_middleware_execution_order(self):
        global context
        app = falcon.API(middleware=[ExecutedFirstMiddleware(),
                                     ExecutedLastMiddleware()])

        app.add_route(TEST_ROUTE, MiddlewareClassResource())
        client = testing.TestClient(app)

        response = client.simulate_request(path=TEST_ROUTE)
        assert _EXPECTED_BODY == response.json
        assert response.status == falcon.HTTP_200
        # as the method registration is in a list, the order also is
        # tested
        expectedExecutedMethods = [
            'ExecutedFirstMiddleware.process_request',
            'ExecutedLastMiddleware.process_request',
            'ExecutedFirstMiddleware.process_resource',
            'ExecutedLastMiddleware.process_resource',
            'ExecutedLastMiddleware.process_response',
            'ExecutedFirstMiddleware.process_response'
        ]
        assert expectedExecutedMethods == context['executed_methods']

def test_independent_middleware_execution_order(self):
        global context
        app = falcon.API(independent_middleware=True,
                         middleware=[ExecutedFirstMiddleware(),
                                     ExecutedLastMiddleware()])

        app.add_route(TEST_ROUTE, MiddlewareClassResource())
        client = testing.TestClient(app)

        response = client.simulate_request(path=TEST_ROUTE)
        assert _EXPECTED_BODY == response.json
        assert response.status == falcon.HTTP_200
        # as the method registration is in a list, the order also is
        # tested
        expectedExecutedMethods = [
            'ExecutedFirstMiddleware.process_request',
            'ExecutedLastMiddleware.process_request',
            'ExecutedFirstMiddleware.process_resource',
            'ExecutedLastMiddleware.process_resource',
            'ExecutedLastMiddleware.process_response',
            'ExecutedFirstMiddleware.process_response'
        ]
        assert expectedExecutedMethods == context['executed_methods']

def test_inner_mw_throw_exception(self):
        """Test that error in inner middleware leaves"""
        global context

        class RaiseErrorMiddleware(object):

            def process_request(self, req, resp):
                raise Exception('Always fail')

        app = falcon.API(middleware=[TransactionIdMiddleware(),
                                     RequestTimeMiddleware(),
                                     RaiseErrorMiddleware()])

        app.add_route(TEST_ROUTE, MiddlewareClassResource())
        client = testing.TestClient(app)

        with pytest.raises(Exception):
            client.simulate_request(path=TEST_ROUTE)

        # RequestTimeMiddleware process_response should be executed
        assert 'transaction_id' in context
        assert 'start_time' in context
        assert 'mid_time' not in context
        assert 'end_time' in context

def test_can_access_resource_params(self):
        """Test that params can be accessed from within process_resource"""
        global context

        class Resource:
            def on_get(self, req, resp, **params):
                resp.body = json.dumps(params)

        app = falcon.API(middleware=AccessParamsMiddleware())
        app.add_route('/path/{id}', Resource())
        client = testing.TestClient(app)
        response = client.simulate_request(path='/path/22')

        assert 'params' in context
        assert context['params']
        assert context['params']['id'] == '22'
        assert response.json == {'added': True, 'id': '22'}

def test_http_status_raised_from_error_handler(self):
        mw = CaptureResponseMiddleware()
        app = falcon.API(middleware=mw)
        app.add_route('/', MiddlewareClassResource())
        client = testing.TestClient(app)

        def _http_error_handler(error, req, resp, params):
            raise falcon.HTTPStatus(falcon.HTTP_201)

        # NOTE(kgriffs): This will take precedence over the default
        # handler for facon.HTTPError.
        app.add_error_handler(falcon.HTTPError, _http_error_handler)

        response = client.simulate_request(path='/', method='POST')
        assert response.status == falcon.HTTP_201
        assert mw.resp.status == response.status

def test_pep3333(self):
        api = falcon.API()
        mock = testing.StartResponseMock()

        # Simulate a web request (normally done though a WSGI server)
        response = api(testing.create_environ(), mock)

        # Verify that the response is iterable
        assert _is_iterable(response)

        # Make sure start_response was passed a valid status string
        assert mock.call_count == 1
        assert isinstance(mock.status, str)
        assert re.match('^\d+[a-zA-Z\s]+$', mock.status)

        # Verify headers is a list of tuples, each containing a pair of strings
        assert isinstance(mock.headers, list)
        if len(mock.headers) != 0:
            header = mock.headers[0]
            assert isinstance(header, tuple)
            assert len(header) == 2
            assert isinstance(header[0], str)
            assert isinstance(header[1], str)

def falcon(body, headers):
    import falcon

    path = '/hello/{account_id}/test'
    falcon_app = falcon.API('text/plain')

    # def ask(req, resp, params):
    #     params['answer'] = 42

    # @falcon.before(ask)
    class HelloResource:
        def on_get(self, req, resp, account_id):
            user_agent = req.user_agent  # NOQA
            limit = req.get_param('limit') or '10'  # NOQA
            resp.data = body
            resp.set_headers(headers)

    falcon_app.add_route(path, HelloResource())

    return falcon_app

def create_api_app(global_conf, **local_conf):
    """Create Main Events Api application.

    :param global_conf: Global config
    :param local_conf: Local config
    :return: falcon.API
    """
    controllers = {}
    controllers.update({
        '/events': v1_events.Events()
    })

    wsgi_app = falcon.API(
        request_type=request.Request
    )

    for route, ctrl in controllers.items():
        wsgi_app.add_route(route, ctrl)

    error_handlers.register_error_handler(wsgi_app)

    return wsgi_app

def test_app(backend):
    from webtest import TestApp
    from vishnu.middleware import SessionMiddleware
    from vishnu.session import Config

    api = falcon.API()
    api.add_route("/private", PrivateHandler())
    api.add_route("/public", PublicHandler())
    api.add_route("/login/save", LoginHandler())
    api.add_route("/logout", LogoutHandler())

    config = Config(
        secret="OVc1Mbt79AK5Pmi6sWnJnXZvEPNO3BnI",
        backend=backend
    )
    session = SessionMiddleware(api, config)

    return TestApp(app=session, extra_environ={'wsgi.url_scheme': 'https'})

def on_get(self, req, resp):
        """Handles GET requests"""
        resp.body = """
<html>
  <head>
    <title>Quote API Server</title>
  </head>
  <body>
    <p>This is a toy JSON API server example.</p>
    <p>Make a GET request to <a href="%s/quote">%s/quote</a></p>
  </body>
</html>
""" % (self.prefix, self.prefix)
        resp.content_type = "text/html"
        resp.status = falcon.HTTP_200


# A Falcon middleware to implement validation of the Host header in requests

def __init__(self, falcon_api=None,
                 static_path='static', static_dir='static', log_config=None):
        if log_config is None:
            self.log = JLog().setup().bind()
        else:
            self.log = JLog().setup(config=log_config).bind()
        self.log.info(cc('falsy init', fore=77, styles=['italic', 'underlined', 'reverse']))

        self.api = self.falcon_api = falcon_api or falcon.API()
        self.static_path = static_path.strip('/')
        self.static_dir = static_dir if os.path.isdir(static_dir) else '.'

        self.api = CommonStaticMiddleware(self.falcon_api, static_dir=self.static_dir,
                                          url_prefix=self.static_path)
        self.log.info('common static middleware loaded\n\t{}'.format(
            'url_prefix(static_path):' + reverse() + self.static_path + rreverse() +
            ', static_dir:' + reverse() + self.static_dir + rreverse()))

def traverse(roots, parent='', verbose=False):
    """
    Recursive call which also handles printing output.

    :param api: The falcon.API or callable that returns an instance to look at.
    :type api: falcon.API or callable
    :param parent: The parent uri path to the current iteration.
    :type parent: str
    :param verbose: If the output should be verbose.
    :type verbose: bool
    """
    for root in roots:
        if root.method_map:
            print('->', parent + '/' + root.raw_segment)
            if verbose:
                for method, func in root.method_map.items():
                    if func.__name__ != 'method_not_allowed':
                        print('-->{0} {1}:{2}'.format(
                            method,
                            inspect.getsourcefile(func),
                            inspect.getsourcelines(func)[1]))
        if root.children:
            traverse(root.children, parent + '/' + root.raw_segment, verbose)

def setUp(self):
        super(TestCase, self).setUp()

        if self.api_class is None:
            app = falcon.API()
        else:
            app = self.api_class()

        # NOTE(kgriffs): Don't use super() to avoid triggering
        # unittest.TestCase.__init__()
        TestClient.__init__(self, app)

        # Reset to simulate "restarting" the WSGI container
        falcon.request._maybe_wrap_wsgi_stream = True

    # NOTE(warsaw): Pythons earlier than 2.7 do not have a
    # self.assertIn() method, so use this compatibility function
    # instead.

def on_get(self, req, resp):
        token = req.get_param('token', True)
        data = {}
        for key in self.data_keys:
            data[key] = req.get_param(key, True)

        if not self.validate_token(token, data):
            raise falcon.HTTPForbidden('Invalid token for these given values', '')

        endpoint = self.config['iris']['hook']['gmail_one_click']

        try:
            result = self.iclient.post(endpoint, data)
        except MaxRetryError:
            logger.exception('Hitting iris-api failed for gmail oneclick')
        else:
            if result.status == 204:
                resp.status = falcon.HTTP_204
                return
            else:
                logger.error('Unexpected status code from api %s for gmail oneclick', result.status)

        raise falcon.HTTPInternalServerError('Internal Server Error', 'Invalid response from API')

def on_post(self, req, resp):
        """
        Accept twilio SMS webhook and forward to iris API
        """
        try:
            path = self.config['iris']['hook']['twilio_messages']
            re = self.iclient.post(path, req.context['body'], raw=True)
        except MaxRetryError as e:
            logger.error(e.reason)
            self.return_twixml_message('Connection error to web hook.', resp)
            return

        if re.status is not 200:
            self.return_twixml_message(
                'Got status code: %d, content: %s' % (re.status,
                                                      re.data[0:100]), resp)
            return
        else:
            body = process_api_response(re.data)
            self.return_twixml_message(body, resp)
            return

def on_post(self, req, resp):
        """
        Accept twilio POST that has message delivery status, and pass it
        to iris-api
        """

        try:
            re = self.iclient.post(self.endpoint, req.context['body'], raw=True)
        except MaxRetryError:
            logger.exception('Failed posting data to iris-api')
            raise falcon.HTTPInternalServerError('Internal Server Error', 'API call failed')

        if re.status is not 204:
            logger.error('Invalid response from API for delivery status update: %s', re.status)
            raise falcon.HTTPBadRequest('Likely bad params passed', 'Invalid response from API')

        resp.status = falcon.HTTP_204

def on_get(self, req, resp):
        """Handles GET requests"""
        resp.body = """
<html>
  <head>
    <title>Quote API Server</title>
  </head>
  <body>
    <p>This is a toy JSON API server example.</p>
    <p>Make a GET request to <a href="%s/quote">%s/quote</a></p>
  </body>
</html>
""" % (self.prefix, self.prefix)
        resp.content_type = "text/html"
        resp.status = falcon.HTTP_200


# A Falcon middleware to implement validation of the Host header in requests

def __init__(self, port, address, prefix, hosts):
        assert prefix is not None
        assert port is not None
        assert address is not None
        self.application = falcon.API(middleware=HostFilter(hosts))
        # add_route is pedantic about this
        if prefix != '' and not prefix.startswith("/"):
            prefix = "/" + prefix
        self.application.add_route(prefix + '/quote', QuoteResource())
        self.application.add_route(prefix + "/", IndexResource(prefix))
        self.port = port
        self.address = address
        self.prefix = prefix
        super(QuoteApplication, self).__init__()

        print("Only connections via these hosts are allowed: " + repr(hosts))

def __init__(self, djenkins, pipeline_results):
        """
        Initialize the API with instantiated DJenkins, PipelineResults and AnalysisService objects
        :param djenkins: DJenkins instance
        :param pipeline_results: PipelineResults instance
        """
        super(self.__class__, self).__init__()
        self.djenkins = djenkins
        self.db = pipeline_results
        heatmap = HeatmapResource(database=self.db)
        self.add_route('/heatmap/', heatmap)
        self.add_route('/heatmap/{project}', heatmap)
        results = ResultsResource(database=self.db)
        self.add_route('/results/', results)
        self.add_route('/results/{project}', results)
        self.add_route('/results/{project}/{repo}', results)
        projects = ProjectResource(database=self.db)
        self.add_route('/projects/', projects)
        self.add_route('/projects/{project}', projects)

def __init__(self, hs_address, hs_token, *args, **kwargs):
        # This runs the __init__ method of falcon.API
        super().__init__(*args, **kwargs)
        self.hs_address = hs_address
        self.hs_token = hs_token
        # Returns api object prepopulated with hs_address and token
        self.Api = functools.partial(MatrixASHttpAPI, hs_address,
                                     token=hs_token)

def _init_routes_and_middlewares(self):
        """Initialize hooks and URI routes to resources."""
        self._init_middlewares()
        self._init_endpoints()

        self.app = falcon.API(middleware=self.middleware)
        self.app.add_error_handler(Exception, self._error_handler)

        for version_path, endpoints in self.catalog:
            for route, resource in endpoints:
                self.app.add_route(version_path + route, resource)

def _get_server_cls(self, host):
        """Return an appropriate WSGI server class base on provided host
        :param host: The listen host for the zaqar API server.
        """
        server_cls = simple_server.WSGIServer
        if netutils.is_valid_ipv6(host):
            if getattr(server_cls, 'address_family') == socket.AF_INET:
                class server_cls(server_cls):
                    address_family = socket.AF_INET6
        return server_cls

def simulate_cors_api(self, cors, route='/'):
        self.api = falcon.API(middleware=[cors.middleware])
        self.api.add_route(route, self.resource)

def deckhand_app_factory(global_config, **local_config):
    # The order of the middleware is important because the `process_response`
    # method for `YAMLTranslator` should execute after that of any other
    # middleware to convert the response to YAML format.
    middleware_list = [middleware.YAMLTranslator(),
                       middleware.ContextMiddleware()]

    app = falcon.API(request_type=base.DeckhandRequest,
                     middleware=middleware_list)

    return configure_app(app, version='v1.0')