python类HTTP_200的实例源码

resources.py 文件源码 项目:gyr 作者: non-Jedi 项目源码 文件源码 阅读 22 收藏 0 点赞 0 评论 0
def on_put(self, request, response, txn_id=None):
        """Responds to PUT request containing events."""
        response.body = "{}"

        # Check whether repeat txn_id
        if not self._is_new(txn_id):
            response.status = falcon.HTTP_200
            return

        request.context["body"] = request.stream.read()
        try:
            events = json.loads(request.context["body"].decode("utf-8"))["events"]
        except(KeyError, ValueError, UnicodeDecodeError):
            response.status = falcon.HTTP_400
            response.body = "Malformed request body"
            return

        if self.handler(EventStream(events, self.Api)):
            response.status = falcon.HTTP_200
        else:
            response.status = falcon.HTTP_400
health_resource.py 文件源码 项目:antenna 作者: mozilla-services 项目源码 文件源码 阅读 16 收藏 0 点赞 0 评论 0
def on_get(self, req, resp):
        mymetrics.incr('heartbeat.count')
        state = HealthState()

        # So we're going to think of Antenna like a big object graph and
        # traverse passing along the HealthState instance. Then, after
        # traversing the object graph, we'll tally everything up and deliver
        # the news.
        for resource in self.antenna_app.get_resources():
            if hasattr(resource, 'check_health'):
                resource.check_health(state)

        # Go through and call gauge for each statsd item.
        for k, v in state.statsd.items():
            mymetrics.gauge(k, value=v)

        if state.is_healthy():
            resp.status = falcon.HTTP_200
        else:
            resp.status = falcon.HTTP_503
        resp.body = json.dumps(state.to_dict())
server.py 文件源码 项目:spacy-services 作者: explosion 项目源码 文件源码 阅读 18 收藏 0 点赞 0 评论 0
def on_get(self, req, resp, model_name):
        try:
            model = get_model(model_name)
            output = {
                'dep_types': get_dep_types(model),
                'ent_types': get_ent_types(model),
                'pos_types': get_pos_types(model)
            }

            resp.body = json.dumps(output, sort_keys=True, indent=2)
            resp.content_type = 'text/string'
            resp.append_header('Access-Control-Allow-Origin', "*")
            resp.status = falcon.HTTP_200
        except Exception as e:
            raise falcon.HTTPBadRequest(
                'Schema construction failed',
                '{}'.format(e))
            resp.status = falcon.HTTP_500
server.py 文件源码 项目:spacy-services 作者: explosion 项目源码 文件源码 阅读 20 收藏 0 点赞 0 评论 0
def on_post(self, req, resp):
        req_body = req.stream.read()
        json_data = json.loads(req_body.decode('utf8'))
        text = json_data.get('text')
        model_name = json_data.get('model', 'en')
        collapse_punctuation = json_data.get('collapse_punctuation', True)
        collapse_phrases = json_data.get('collapse_phrases', True)

        try:
            model = get_model(model_name)
            parse = Parse(model, text, collapse_punctuation, collapse_phrases)
            resp.body = json.dumps(parse.to_json(), sort_keys=True, indent=2)
            resp.content_type = 'text/string'
            resp.append_header('Access-Control-Allow-Origin', "*")
            resp.status = falcon.HTTP_200
        except Exception as e:
            raise falcon.HTTPBadRequest(
                'Dependency parsing failed',
                '{}'.format(e))
            resp.status = falcon.HTTP_500
validation.py 文件源码 项目:armada 作者: att-comdev 项目源码 文件源码 阅读 28 收藏 0 点赞 0 评论 0
def on_post(self, req, resp):
        try:
            manifest = self.req_yaml(req)
            documents = list(manifest)

            message = {
                'valid': validate_armada_documents(documents)
            }

            resp.status = falcon.HTTP_200
            resp.body = json.dumps(message)
            resp.content_type = 'application/json'

        except Exception:
            err_message = 'Failed to validate Armada Manifest'
            self.error(req.context, err_message)
            self.return_error(
                resp, falcon.HTTP_400, message=err_message)
test_httpstatus.py 文件源码 项目:deb-python-falcon 作者: openstack 项目源码 文件源码 阅读 17 收藏 0 点赞 0 评论 0
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'
test_httpstatus.py 文件源码 项目:deb-python-falcon 作者: openstack 项目源码 文件源码 阅读 16 收藏 0 点赞 0 评论 0
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'
test_httpstatus.py 文件源码 项目:deb-python-falcon 作者: openstack 项目源码 文件源码 阅读 18 收藏 0 点赞 0 评论 0
def test_raise_status_runs_process_response(self):
        """ Make sure process_response still runs """
        class TestMiddleware:
            def process_response(self, req, resp, response):
                resp.status = falcon.HTTP_200
                resp.set_header('X-Failed', 'False')
                resp.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'
test_middleware.py 文件源码 项目:deb-python-falcon 作者: openstack 项目源码 文件源码 阅读 19 收藏 0 点赞 0 评论 0
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']
test_middleware.py 文件源码 项目:deb-python-falcon 作者: openstack 项目源码 文件源码 阅读 23 收藏 0 点赞 0 评论 0
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'
test_middleware.py 文件源码 项目:deb-python-falcon 作者: openstack 项目源码 文件源码 阅读 26 收藏 0 点赞 0 评论 0
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']
test_uri_templates.py 文件源码 项目:deb-python-falcon 作者: openstack 项目源码 文件源码 阅读 17 收藏 0 点赞 0 评论 0
def test_single_trailing_slash(client):
    resource1 = IDResource()
    client.app.add_route('/1/{id}/', resource1)
    result = client.simulate_get('/1/123')
    assert result.status == falcon.HTTP_200
    assert resource1.called
    assert resource1.id == '123'
    assert resource1.req.path == '/1/123'

    resource2 = IDResource()
    client.app.add_route('/2/{id}/', resource2)
    result = client.simulate_get('/2/123/')
    assert result.status == falcon.HTTP_200
    assert resource2.called
    assert resource2.id == '123'
    assert resource2.req.path == '/2/123'

    resource3 = IDResource()
    client.app.add_route('/3/{id}', resource3)
    result = client.simulate_get('/3/123/')
    assert result.status == falcon.HTTP_200
    assert resource3.called
    assert resource3.id == '123'
    assert resource3.req.path == '/3/123'
resource.py 文件源码 项目:deb-python-falcon 作者: openstack 项目源码 文件源码 阅读 16 收藏 0 点赞 0 评论 0
def on_get(self, req, resp, **kwargs):
        """GET responder.

        Captures `req`, `resp`, and `kwargs`. Also sets up a sample response.

        Args:
            req: Falcon ``Request`` instance.
            resp: Falcon ``Response`` instance.
            kwargs: URI template *name=value* pairs, if any, along with
                any extra args injected by middleware.

        """

        # Don't try this at home - classes aren't recreated
        # for every request
        self.req, self.resp, self.kwargs = req, resp, kwargs

        self.called = True
        resp.status = falcon.HTTP_200
        resp.body = self.sample_body
        resp.set_headers(self.resp_headers)
designs.py 文件源码 项目:drydock 作者: att-comdev 项目源码 文件源码 阅读 19 收藏 0 点赞 0 评论 0
def on_get(self, req, resp):
        """Method handler for GET requests.

        :param req: Falcon request object
        :param resp: Falcon response object
        """
        state = self.state_manager

        try:
            designs = list(state.designs.keys())

            resp.body = json.dumps(designs)
            resp.status = falcon.HTTP_200
        except Exception as ex:
            self.error(req.context, "Exception raised: %s" % str(ex))
            self.return_error(
                resp,
                falcon.HTTP_500,
                message="Error accessing design list",
                retry=True)
test_events_v1.py 文件源码 项目:monasca-events-api 作者: openstack 项目源码 文件源码 阅读 15 收藏 0 点赞 0 评论 0
def test_should_pass_simple_event(self, bulk_processor):
        events_resource = _init_resource(self)
        events_resource._processor = bulk_processor
        unit_test_patch = os.path.dirname(__file__)
        json_file_path = 'event_template_json/req_simple_event.json'
        patch_to_req_simple_event_file = os.path.join(unit_test_patch,
                                                      json_file_path)
        with open(patch_to_req_simple_event_file, 'r') as fi:
            body = fi.read()
        self.simulate_request(
            path=ENDPOINT,
            method='POST',
            headers={
                'Content-Type': 'application/json',
                'X_ROLES': 'monasca'
            },
            body=body
        )
        self.assertEqual(falcon.HTTP_200, self.srmock.status)
test_events_v1.py 文件源码 项目:monasca-events-api 作者: openstack 项目源码 文件源码 阅读 17 收藏 0 点赞 0 评论 0
def test_should_multiple_events(self, bulk_processor):
        events_resource = _init_resource(self)
        events_resource._processor = bulk_processor
        unit_test_patch = os.path.dirname(__file__)
        json_file_path = 'event_template_json/req_multiple_events.json'
        req_multiple_events_json = os.path.join(unit_test_patch,
                                                json_file_path)
        with open(req_multiple_events_json, 'r') as fi:
            body = fi.read()
        self.simulate_request(
            path=ENDPOINT,
            method='POST',
            headers={
                'Content-Type': 'application/json',
                'X_ROLES': 'monasca'
            },
            body=body
        )
        self.assertEqual(falcon.HTTP_200, self.srmock.status)
server.py 文件源码 项目:spacy-dev-resources 作者: explosion 项目源码 文件源码 阅读 18 收藏 0 点赞 0 评论 0
def on_post(self, req, resp):
        req_body = req.stream.read()
        json_data = json.loads(req_body.decode('utf8'))
        paragraphs = json_data.get('paragraphs')
        model_name = json_data.get('model', 'en')
        try:
            model = get_model(model_name)
            entities = []
            for p in paragraphs:
                e = Entities(model, p.get('text'))
                entities.append(e.to_json())
            resp.body = json.dumps(entities, sort_keys=True, indent=2)
            resp.content_type = 'application/json'
            resp.status = falcon.HTTP_200
        except Exception:
            resp.status = falcon.HTTP_500
server.py 文件源码 项目:spacy-dev-resources 作者: explosion 项目源码 文件源码 阅读 20 收藏 0 点赞 0 评论 0
def on_post(self, req, resp):
        req_body = req.stream.read()
        json_data = json.loads(req_body.decode('utf8'))
        paragraphs = json_data.get('paragraphs')
        model_name = json_data.get('model', 'en')
        try:
            model = get_model(model_name)
            texts = [paragraph.get('text') for paragraph in paragraphs]
            update_vocabulary(model, texts)
            entities = []
            for p in paragraphs:
                e = TrainEntities(model, p.get('text'), p.get('tags'))
                entities.append(e.to_json())
            resp.body = json.dumps(entities, sort_keys=True, indent=2)
            resp.content_type = 'application/json'
            resp.status = falcon.HTTP_200
        except Exception:
            print("Unexpected error:", sys.exc_info()[0])
            resp.status = falcon.HTTP_500
wechatapiresource.py 文件源码 项目:wechat-bot-skeleton-python 作者: edonosotti 项目源码 文件源码 阅读 19 收藏 0 点赞 0 评论 0
def on_get(self, request, response):
        # Get the parameters from the query string
        signature = request.get_param('signature')
        timestamp = request.get_param('timestamp')
        nonce = request.get_param('nonce')
        echostr = request.get_param('echostr')

        # Compute the signature (note that the shared token is used too)
        verification_elements = [self.token, timestamp, nonce]
        verification_elements.sort()
        verification_string = "".join(verification_elements)
        verification_string = hashlib.sha1(verification_string.encode('utf-8')).hexdigest()

        # If the signature is correct, output the same "echostr" provided by the WeChat server as a parameter
        if signature == verification_string:
            response.status = falcon.HTTP_200
            response.body = echostr
        else:
            response.status = falcon.HTTP_500
            response.body = ""

    # Messages will be POSTed from the WeChat server to the chatbot backend server,
    # see: http://admin.wechat.com/wiki/index.php?title=Common_Messages
wechatapiresource.py 文件源码 项目:wechat-bot-skeleton-python 作者: edonosotti 项目源码 文件源码 阅读 18 收藏 0 点赞 0 评论 0
def on_post(self, request, response):
        formatter = WeChatMessageFormatter()
        message = formatter.parse_incoming_message(request.bounded_stream.read())

        if message['valid']:
            # Queue the message for delayed processing
            self.db_manager.queue_message(message)
            # WeChat always requires incoming user messages to be acknowledged at
            # least with an empty string (empty strings are not shown to users),
            # see: https://chatbotsmagazine.com/building-chatbots-for-wechat-part-1-dba8f160349
            # In this sample app, we simulate a "Customer Service"-like scenario
            # providing an instant reply to the user, announcing that a complete
            # reply will follow.
            reply = "Thank you for your message. We will get back to you as soon as possible!"
            response.status = falcon.HTTP_200
            response.body = formatter.format_instant_reply(message, reply)
        else:
            response.status = falcon.HTTP_200
            response.body = "Message was sent in a wrong format."
falcon_ex.py 文件源码 项目:keycloak-python 作者: ibotty 项目源码 文件源码 阅读 17 收藏 0 点赞 0 评论 0
def on_get(self, req, resp):
        resp.status = falcon.HTTP_200

        if self.keycloak:
            grant = self.keycloak.get_grant(req, resp)

            if self.keycloak.grant_has_role(grant, "resources-writer"):
                resp.body = "has role! token: %s" % self.keycloak.manager.decode_token(grant.access_token)
            else:
                try:
                    resp.body = "No role! token: %s" % self.keycloak.manager.decode_token(grant.access_token)
                except:
                    resp.body = "No valid bearer token."
        else:
            resp.body = """
            Everything a-ok.
            """
webserver.py 文件源码 项目:orbital-cotwo-web 作者: chronos-pramantha 项目源码 文件源码 阅读 29 收藏 0 点赞 0 评论 0
def on_post(self, req, resp):
        """Handles POST requests"""
        resp.status = falcon.HTTP_200
        # grab 'geojson' from req.context
        data = req.context['geojson']
        # get coordinates from geojson
        coords = spatial.from_list_to_ewkt(
            spatial.coordinates_from_geojson(data)
        )
        print(coords)
        # create the controller view
        controller = Controller(coords)
        # find the areas contained by controller's view and connected points' data
        controller.which_areas_contains_this_polygon()
        # dump the retrieved data
        json = controller.serialize_features_from_database()

        print(str(controller))
        print(json)

        req.context['result'] = json
quote.py 文件源码 项目:kapsel 作者: conda 项目源码 文件源码 阅读 29 收藏 0 点赞 0 评论 0
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
validatedesign.py 文件源码 项目:promenade 作者: att-comdev 项目源码 文件源码 阅读 16 收藏 0 点赞 0 评论 0
def _return_msg(self, resp, status_code, status="Valid", message=""):
        if status_code is falcon.HTTP_200:
            count = 0
            msg_list = []
        else:
            count = 1
            msg_list = [message]
        resp.body = json.dumps({
            "kind": "Status",
            "apiVersion": "v1",
            "metadata": {},
            "status": status,
            "message": message,
            "reason": "Validation",
            "details": {
                "errorCount": count,
                "messageList": msg_list,
            },
            "code": status_code,
        })
elb_resource_test.py 文件源码 项目:Cuppa 作者: flipkart-incubator 项目源码 文件源码 阅读 20 收藏 0 点赞 0 评论 0
def test_on_get(self):
        time.sleep(1)
        req = None
        resp = falcon.Response()
        resp.status = None
        resp.body = None
        self.elb_resource.on_get(req, resp)
        response_body = json.loads(resp.body)
        self.assertEquals(response_body['capacity'], 100)
        self.assertEquals(response_body['requests'], 1)
        self.assertGreaterEqual(response_body['uptime'], 1)
        self.assertEquals(resp.status, falcon.HTTP_200)

        self.load_balancer.check_if_model_to_workers_map_is_empty = MagicMock(return_value = True)

        resp_1 = falcon.Response()
        resp_1.status = None
        resp_1.body = None

        self.assertRaises(falcon.HTTPInternalServerError, lambda : self.elb_resource.on_get(req, resp_1))
elb_resource.py 文件源码 项目:Cuppa 作者: flipkart-incubator 项目源码 文件源码 阅读 17 收藏 0 点赞 0 评论 0
def on_get(self, req, resp):
        """
        Handles GET requests for ELB HealthCheck Resource
        :param req:
        :param resp:
        :return:
        """
        uptime = int(time.time()) - self.start_time

        if self.load_balancer.check_if_model_to_workers_map_is_empty():
            resp.status = falcon.HTTP_500
            resp.body = "Model To Workers Map is Empty"
            raise falcon.HTTPInternalServerError('Internal Server Error', 'Model To Workers Map is Empty! ')

        resp.status = falcon.HTTP_200

        # TODO requests and capacity have to be calculated. They are hardcoded for now

        resp.body = json.dumps({'uptime':uptime, 'requests': 1, 'capacity': 100})
caffe_model_resource.py 文件源码 项目:Cuppa 作者: flipkart-incubator 项目源码 文件源码 阅读 17 收藏 0 点赞 0 评论 0
def on_post(self, req, resp):
        payload = json.loads(req.stream.read())
        pi = PredictionInput()
        if ('url' in payload.keys()) and ('modelId' in payload.keys()):
            pi.url = payload['url']
            pi.model_id = payload['modelId']
        else:
            resp.status = falcon.HTTP_400
            raise falcon.HTTPBadRequest("Bad Request", "Url and(or) modelId missing in the payload")

        po = self.caffe_worker_client.predict(pi)

        if po.bo.status == 'Success':
            resp.status = falcon.HTTP_200
            resp.body = (str(po.values))
        elif po.bo.status == 'Failure':
            resp.body = json.dumps({'status': 'Failure', 'message' : 'Error occurred'})
            resp.status = falcon.HTTP_500
            raise falcon.HTTPInternalServerError('Internal Server Error', 'Predict failed! ')
api.py 文件源码 项目:peach 作者: sebastiandev 项目源码 文件源码 阅读 20 收藏 0 点赞 0 评论 0
def __init__(self,
                 app,
                 prefix,
                 name=None,
                 version=None,
                 media_type=None,
                 **kwargs):
        self._media_type = media_type or self.MEDIA_TYPE

        class EntryPoint(object):

            def on_get(self, req, resp):
                resp.body = json.dumps({
                    'name': name or 'Peach Rest Api',
                    'version': version or '0.0.0',
                })
                resp.status = falcon.HTTP_200

        app._media_type = self._media_type
        app.add_route('/{}'.format(prefix), EntryPoint())
        app.add_error_handler(ApiException, self.handle_error)
falcon_app.py 文件源码 项目:py2swagger 作者: Arello-Mobile 项目源码 文件源码 阅读 18 收藏 0 点赞 0 评论 0
def on_get(self, req, resp, id):
        """
        Handles GET requests for another test
        ---
        tags:
        - test
        responses:
          200:
            schema:
              type: string
        security:
          api_key: []
        securityDefinitions:
          api_key:
            type: apiKey
            in: Header
            name: Authorization
        """
        resp.status = falcon.HTTP_200  # This is the default status
        resp.body = '\nHello world with {}\n\n'.format(id)
resource.py 文件源码 项目:zenchmarks 作者: squeaky-pl 项目源码 文件源码 阅读 17 收藏 0 点赞 0 评论 0
def on_get(self, req, resp, **kwargs):
        """GET responder.

        Captures `req`, `resp`, and `kwargs`. Also sets up a sample response.

        Args:
            req: Falcon ``Request`` instance.
            resp: Falcon ``Response`` instance.
            kwargs: URI template *name=value* pairs, if any, along with
                any extra args injected by middleware.

        """

        # Don't try this at home - classes aren't recreated
        # for every request
        self.req, self.resp, self.kwargs = req, resp, kwargs

        self.called = True
        resp.status = falcon.HTTP_200
        resp.body = self.sample_body
        resp.set_headers(self.resp_headers)


问题


面经


文章

微信
公众号

扫码关注公众号