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
python类before()的实例源码
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 test_raise_status_in_before_hook(self):
""" Make sure we get the 200 raised by before hook """
app = falcon.API()
app.add_route('/status', TestStatusResource())
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 validate_request(req, resp, resource, params):
'''
Decorator method to validate token before processing request.
'''
# read request body and parse it into a json object.
data = req.stream
# token to grant access to API
# this is set in the environment of the system where API is deployed.
valid_token = os.environ.get('PY_GOOGLE_AUTH_TOKEN')
if 'token' not in data:
msg = 'Please send access token along with your request'
raise falcon.HTTPBadRequest('Token Required', msg)
else:
# token received from the request data.
req_token = data['token']
if req_token != valid_token:
msg = 'Please supply a valid token.'
raise falcon.HTTPBadRequest('Invalid Token', msg)
# since stream is a file, it has been read once so won't be able to read it again in the end
# point functions that are called afterwards, so setting it to the data that was already parsed
# so that it is available in the functions that follows.
req.stream = data
def json_app(app):
class JSONResponseTestResource(SimpleTestResource):
@jsonresponse
def on_get(self, req, resp, **kwargs):
return {"foo": "bar"}
@jsonrequest
@falcon.before(capture_responder_args)
def on_post(self, req, resp):
return {}
resource = JSONResponseTestResource()
app.app.add_route("/_testing", resource)
return (resource, app)