def get_request():
"""Returns the active request instance.
:returns:
A :class:`Request` instance.
"""
if _local:
assert getattr(_local, 'request', None) is not None, _get_request_error
else:
assert WSGIApplication.request is not None, _get_request_error
return WSGIApplication.request
python类Request()的实例源码
def get_wsgi_application(self):
def create_response(environ, start_response, status=200):
from webob import Response
res = Response(
content_type='application/json',
body='{}',
status=status
)
return res(environ, start_response)
def application(environ, start_response):
import json
from webob import Request
request = Request(environ)
# Check HMAC-SHA256
if not self.is_valid_sign(request.body, request.headers.get('X-Line-Signature')):
return create_response(environ, start_response, 403)
# Check Handler
if not self.handler:
return create_response(environ, start_response, 501)
# Parse Event
try:
obj = json.loads(request.body.decode('utf-8'))
from .parse import event_parser
events = [event_parser(event_json, self.channel) for event_json in obj['events']]
except:
return create_response(environ, start_response, 400)
# Call Event Handler
for event in events:
self.handler(event)
return create_response(environ, start_response)
return application
def __call__(self, environ, start_response):
r"""Subclasses will probably want to implement __call__ like this:
@webob.dec.wsgify(RequestClass=Request)
def __call__(self, req):
# Any of the following objects work as responses:
# Option 1: simple string
res = 'message\n'
# Option 2: a nicely formatted HTTP exception page
res = exc.HTTPForbidden(explanation='Nice try')
# Option 3: a webob Response object (in case you need to play with
# headers, or you want to be treated like an iterable)
res = Response();
res.app_iter = open('somefile')
# Option 4: any wsgi app to be run next
res = self.application
# Option 5: you can get a Response object for a wsgi app, too, to
# play with headers etc
res = req.get_response(self.application)
# You can then just return your response...
return res
# ... or set req.response and return None.
req.response = res
See the end of http://pythonpaste.org/webob/modules/dec.html
for more info.
"""
raise NotImplementedError(_('You must implement __call__'))
def __init__(self, *args, **kwargs):
super(Request, self).__init__(*args, **kwargs)
self._resource_cache = {}
def get(self, argument_name, default_value='', allow_multiple=False):
"""Returns the query or POST argument with the given name.
We parse the query string and POST payload lazily, so this will be a
slower operation on the first call.
Args:
argument_name: the name of the query or POST argument
default_value: the value to return if the given argument is not present
allow_multiple: return a list of values with the given name (deprecated)
Returns:
If allow_multiple is False (which it is by default), we return the first
value with the given name given in the request. If it is True, we always
return a list.
"""
param_value = self.get_all(argument_name)
if allow_multiple:
logging.warning('allow_multiple is a deprecated param, please use the '
'Request.get_all() method instead.')
if len(param_value) > 0:
if allow_multiple:
return param_value
return param_value[0]
else:
if allow_multiple and not default_value:
return []
return default_value
def initialize(self, request, response):
"""Initializes this request handler with the given Request and Response."""
self.request = request
self.response = response
def get(self, argument_name, default_value='', allow_multiple=False):
"""Returns the query or POST argument with the given name.
We parse the query string and POST payload lazily, so this will be a
slower operation on the first call.
Args:
argument_name: the name of the query or POST argument
default_value: the value to return if the given argument is not present
allow_multiple: return a list of values with the given name (deprecated)
Returns:
If allow_multiple is False (which it is by default), we return the first
value with the given name given in the request. If it is True, we always
return a list.
"""
param_value = self.get_all(argument_name)
if allow_multiple:
logging.warning('allow_multiple is a deprecated param, please use the '
'Request.get_all() method instead.')
if len(param_value) > 0:
if allow_multiple:
return param_value
return param_value[0]
else:
if allow_multiple and not default_value:
return []
return default_value
def initialize(self, request, response):
"""Initializes this request handler with the given Request and Response."""
self.request = request
self.response = response
def __call__(self, environ, start_response):
req = Request(environ)
for regex, controller, vars in self.routes:
match = regex.match(req.path_info)
if match:
req.urlvars = match.groupdict()
req.urlvars.update(vars)
return controller(environ, start_response)
return exc.HTTPNotFound()(environ, start_response)
#wrapper make a function to a controller
def controller(func):
def replacement(environ, start_response):
req = Request(environ)
try:
resp = func(req, **req.urlvars)
except exc.HTTPException as e:
resp = e
if isinstance(resp, type("")):
resp = Response(body=resp)
return resp(environ, start_response)
return replacement
#restful controller
#cls is shorten fron class
def __call__(self, environ, start_response):
req = Request(environ)
get_request.register(req)
try:
return self.app(environ, statr_response)
finally:
get_request.unregister()
#URL Generation
def get(self, argument_name, default_value='', allow_multiple=False):
"""Returns the query or POST argument with the given name.
We parse the query string and POST payload lazily, so this will be a
slower operation on the first call.
Args:
argument_name: the name of the query or POST argument
default_value: the value to return if the given argument is not present
allow_multiple: return a list of values with the given name (deprecated)
Returns:
If allow_multiple is False (which it is by default), we return the first
value with the given name given in the request. If it is True, we always
return a list.
"""
param_value = self.get_all(argument_name)
if allow_multiple:
logging.warning('allow_multiple is a deprecated param, please use the '
'Request.get_all() method instead.')
if len(param_value) > 0:
if allow_multiple:
return param_value
return param_value[0]
else:
if allow_multiple and not default_value:
return []
return default_value
def initialize(self, request, response):
"""Initializes this request handler with the given Request and Response."""
self.request = request
self.response = response
def __call__(self, environ, start_response):
request = Request(environ)
response = self.process_request(request)
return response(environ, start_response)
def get_current_url(cls, request=None):
"""Returns the Url requested by the current Request."""
request = request or ExecutionContext.get_context().request
return cls(six.text_type(request.url))
def make_network_absolute(self):
"""Ensures that this URL has a scheme, hostname and port matching that of the current Request URL."""
# Note: It is tempting to want to use request.scheme, request.server_name and request.server_port
# in this method.
#
# But, request.server_name can be different from request_url.hostname; and if this
# happens, and you use it here, the changed url will result in invalidating the existing
# session cookie on the client which is bound to the request_url.hostname by the browser.
#
request_url = Url.get_current_url()
self.scheme = self.scheme or request_url.scheme
self.hostname = self.hostname or request_url.hostname
self.port = self.port or request_url.port
def current_view(self):
"""The :class:`View` which is targetted by the URL of the current :class:webob.Request."""
return self.view_for(self.relative_path)
def create_file(self, relative_path):
path = relative_path[1:]
context = ExecutionContext.get_context()
static_root = context.config.web.static_root
relative_path = self.root_path.split('/')+path.split('/')
full_path = os.path.join(static_root, *relative_path)
logging.getLogger(__name__).debug('Request is for static file "%s"' % full_path)
if os.path.isfile(full_path):
return FileOnDisk(full_path, relative_path)
raise NoMatchingFactoryFound(relative_path)
def get(self, argument_name, default_value='', allow_multiple=False):
"""Returns the query or POST argument with the given name.
We parse the query string and POST payload lazily, so this will be a
slower operation on the first call.
Args:
argument_name: the name of the query or POST argument
default_value: the value to return if the given argument is not present
allow_multiple: return a list of values with the given name (deprecated)
Returns:
If allow_multiple is False (which it is by default), we return the first
value with the given name given in the request. If it is True, we always
return a list.
"""
param_value = self.get_all(argument_name)
if allow_multiple:
logging.warning('allow_multiple is a deprecated param, please use the '
'Request.get_all() method instead.')
if len(param_value) > 0:
if allow_multiple:
return param_value
return param_value[0]
else:
if allow_multiple and not default_value:
return []
return default_value
def initialize(self, request, response):
"""Initializes this request handler with the given Request and Response."""
self.request = request
self.response = response
