我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用webob.Request()。
def __init__(self, environ, *args, **kwargs): """Constructs a Request object from a WSGI environment. :param environ: A WSGI-compliant environment dictionary. """ if kwargs.get('charset') is None and not hasattr(webob, '__version__'): # webob 0.9 didn't have a __version__ attribute and also defaulted # to None rather than UTF-8 if no charset was provided. Providing a # default charset is required for backwards compatibility. match = _charset_re.search(environ.get('CONTENT_TYPE', '')) if match: self._request_charset = ( match.group(1).lower().strip().strip('"').strip()) kwargs['charset'] = 'utf-8' super(Request, self).__init__(environ, *args, **kwargs) self.registry = {}
def __init__(self, request=None, response=None): """Initializes this request handler with the given WSGI application, Request and Response. When instantiated by ``webapp.WSGIApplication``, request and response are not set on instantiation. Instead, initialize() is called right after the handler is created to set them. Also in webapp dispatching is done by the WSGI app, while webapp2 does it here to allow more flexibility in extended classes: handlers can wrap :meth:`dispatch` to check for conditions before executing the requested method and/or post-process the response. .. note:: Parameters are optional only to support webapp's constructor which doesn't take any arguments. Consider them as required. :param request: A :class:`Request` instance. :param response: A :class:`Response` instance. """ self.initialize(request, response)
def set_globals(self, app=None, request=None): """Registers the global variables for app and request. If :mod:`webapp2_extras.local` is available, the app and request class attributes are assigned to a proxy object that returns them using thread-local, making the application thread-safe. This can also be used in environments that don't support threading. If :mod:`webapp2_extras.local` is not available, app and request will be assigned directly as class attributes. This should only be used in non-threaded environments (e.g., App Engine Python 2.5). :param app: A :class:`WSGIApplication` instance. :param request: A :class:`Request` instance. """ if _local is not None: # pragma: no cover _local.app = app _local.request = request else: # pragma: no cover WSGIApplication.app = WSGIApplication.active_instance = app WSGIApplication.request = request
def __init__(self, environ): """Constructs a Request object from a WSGI environment. If the charset isn't specified in the Content-Type header, defaults to UTF-8. Args: environ: A WSGI-compliant environment dictionary. """ match = _CHARSET_RE.search(environ.get('CONTENT_TYPE', '')) if match: charset = match.group(1).lower() else: charset = 'utf-8' webob.Request.__init__(self, environ, charset=charset, unicode_errors= 'ignore', decode_param_names=True)
def rest_controller(cls): def replacement(environ, start_response): req = Request(environ) try: instance = cls(req, **req.urlvars) action = req.urlvars.get('action') if action: action += '_' + req.method.lower() else: action = req.method.lower() try: method = getattr(instance, action) except AttributeError: raise exc.HTTPNotFound("No action %s" % action) resp = method() if isinstance(resp, type("")): resp = Response(body=resp) except exc.HTTPException as e: resp = e return resp(environ, start_response) return replacement
def __call__(self, environ, start_response): app = self.wrapped request = Request(environ, charset='utf-8') self.exception = None self.traceback = None try: to_return = b'' for i in app(environ, start_response): to_return += i except: to_return = b'' (_, self.exception, self.traceback) = sys.exc_info() traceback_html = six.text_type(traceback.format_exc()) for i in HTTPInternalServerError(content_type=ascii_as_bytes_or_str('text/plain'), charset=ascii_as_bytes_or_str('utf-8'), unicode_body=traceback_html)(environ, start_response): to_return += i yield to_return
def __call__(self, environ, start_response): """Called by WSGI when a request comes in. Args: environ: A dict holding environment variables. start_response: A WSGI callable (PEP333). Returns: Application response data as an iterable. It just returns the return value of the inner WSGI app. """ req = Request(environ) preferred_languages = list(req.accept_language) if self.default_language not in preferred_languages: preferred_languages.append(self.default_language) translation = gettext.translation( 'messages', self.locale_path, fallback=True, languages=preferred_languages, codeset='utf-8') translation.install(unicode=True, names=['gettext', 'ngettext']) environ['i18n_utils.active_translation'] = translation environ['i18n_utils.preferred_languages'] = preferred_languages return self.app(environ, start_response)
def __init__(self, environ, *args, **kwargs): """Constructs a Request object from a WSGI environment. :param environ: A WSGI-compliant environment dictionary. """ if kwargs.get('charset') is None and not hasattr(webob, '__version__'): # webob 0.9 didn't have a __version__ attribute and also defaulted # to None rather than UTF-8 if no charset was provided. Providing a # default charset is required for backwards compatibility. match = _charset_re.search(environ.get('CONTENT_TYPE', '')) if match: charset = match.group(1).lower().strip().strip('"').strip() else: charset = 'utf-8' kwargs['charset'] = charset super(Request, self).__init__(environ, *args, **kwargs) self.registry = {}
def set_globals(self, app=None, request=None): """Registers the global variables for app and request. If :mod:`webapp2_extras.local` is available the app and request class attributes are assigned to a proxy object that returns them using thread-local, making the application thread-safe. This can also be used in environments that don't support threading. If :mod:`webapp2_extras.local` is not available app and request will be assigned directly as class attributes. This should only be used in non-threaded environments (e.g., App Engine Python 2.5). :param app: A :class:`WSGIApplication` instance. :param request: A :class:`Request` instance. """ if _local is not None: # pragma: no cover _local.app = app _local.request = request else: # pragma: no cover WSGIApplication.app = WSGIApplication.active_instance = app WSGIApplication.request = request
def proxy_connection(self, req, connect_info, start_response): """Spawn bi-directional vnc proxy.""" sockets = {} t0 = utils.spawn(self.handshake, req, connect_info, sockets) t0.wait() if not sockets.get('client') or not sockets.get('server'): LOG.info(_LI("Invalid request: %s"), req) start_response('400 Invalid Request', [('content-type', 'text/html')]) return "Invalid Request" client = sockets['client'] server = sockets['server'] t1 = utils.spawn(self.one_way_proxy, client, server) t2 = utils.spawn(self.one_way_proxy, server, client) t1.wait() t2.wait() # Make sure our sockets are closed server.close() client.close()
def __call__(self, environ, start_response): try: req = webob.Request(environ) LOG.info(_LI("Request: %s"), req) token = req.params.get('token') if not token: LOG.info(_LI("Request made with missing token: %s"), req) start_response('400 Invalid Request', [('content-type', 'text/html')]) return "Invalid Request" ctxt = context.get_admin_context() api = consoleauth_rpcapi.ConsoleAuthAPI() connect_info = api.check_token(ctxt, token) if not connect_info: LOG.info(_LI("Request made with invalid token: %s"), req) start_response('401 Not Authorized', [('content-type', 'text/html')]) return "Not Authorized" return self.proxy_connection(req, connect_info, start_response) except Exception as e: LOG.info(_LI("Unexpected error: %s"), e)
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 __init__(self, *args, **kwargs): super(Request, self).__init__(*args, **kwargs) self._resource_cache = {} if not hasattr(self, 'api_version_request'): self.api_version_request = api_version.APIVersionRequest()
def blank(cls, *args, **kwargs): if args is not None: if 'v1' in args[0]: kwargs['base_url'] = 'http://localhost/v1' use_admin_context = kwargs.pop('use_admin_context', False) version = kwargs.pop('version', '1.0') out = base.Request.blank(*args, **kwargs) out.environ['cinder.context'] = FakeRequestContext( fake.USER_ID, fake.PROJECT_ID, is_admin=use_admin_context) out.api_version_request = Join(version) return out
def blank(cls, path, environ=None, base_url=None, headers=None, **kwargs): # pragma: no cover """Adds parameters compatible with WebOb > 1.2: POST and **kwargs.""" try: request = super(Request, cls).blank( path, environ=environ, base_url=base_url, headers=headers, **kwargs ) if cls._request_charset and not cls._request_charset == 'utf-8': return request.decode(cls._request_charset) return request except TypeError: if not kwargs: raise data = kwargs.pop('POST', None) if data is not None: environ = environ or {} environ['REQUEST_METHOD'] = 'POST' if hasattr(data, 'items'): data = list(data.items()) if not isinstance(data, str): data = urlencode(data) environ['wsgi.input'] = cStringIO(data) environ['webob.is_body_seekable'] = True environ['CONTENT_LENGTH'] = str(len(data)) environ['CONTENT_TYPE'] = 'application/x-www-form-urlencoded' base = super(Request, cls).blank(path, environ=environ, base_url=base_url, headers=headers) if kwargs: obj = cls(base.environ, **kwargs) obj.headers.update(base.headers) return obj else: return base
def initialize(self, request, response): """Initializes this request handler with the given WSGI application, Request and Response. :param request: A :class:`Request` instance. :param response: A :class:`Response` instance. """ self.request = request self.response = response self.app = WSGIApplication.active_instance
def match(self, request): """Matches all routes against a request object. The first one that matches is returned. :param request: A :class:`Request` instance. :returns: A tuple ``(route, args, kwargs)`` if a route matched, or None. """ raise NotImplementedError()
def default_matcher(self, request): """Matches all routes against a request object. The first one that matches is returned. :param request: A :class:`Request` instance. :returns: A tuple ``(route, args, kwargs)`` if a route matched, or None. :raises: ``exc.HTTPNotFound`` if no route matched or ``exc.HTTPMethodNotAllowed`` if a route matched but the HTTP method was not allowed. """ method_not_allowed = False for route in self.match_routes: try: match = route.match(request) if match: return match except exc.HTTPMethodNotAllowed: method_not_allowed = True if method_not_allowed: raise exc.HTTPMethodNotAllowed() raise exc.HTTPNotFound()
def default_builder(self, request, name, args, kwargs): """Returns a URI for a named :class:`Route`. :param request: The current :class:`Request` object. :param name: The route name. :param args: Tuple of positional arguments to build the URI. All positional variables defined in the route must be passed and must conform to the format set in the route. Extra arguments are ignored. :param kwargs: Dictionary of keyword arguments to build the URI. All variables not set in the route default values must be passed and must conform to the format set in the route. Extra keywords are appended as a query string. A few keywords have special meaning: - **_full**: If True, builds an absolute URI. - **_scheme**: URI scheme, e.g., `http` or `https`. If defined, an absolute URI is always returned. - **_netloc**: Network location, e.g., `www.google.com`. If defined, an absolute URI is always returned. - **_fragment**: If set, appends a fragment (or "anchor") to the generated URI. :returns: An absolute or relative URI. """ route = self.build_routes.get(name) if route is None: raise KeyError('Route named %r is not defined.' % name) return route.build(request, args, kwargs)
def default_dispatcher(self, request, response): """Dispatches a handler. :param request: A :class:`Request` instance. :param response: A :class:`Response` instance. :raises: ``exc.HTTPNotFound`` if no route matched or ``exc.HTTPMethodNotAllowed`` if a route matched but the HTTP method was not allowed. :returns: The returned value from the handler. """ route, args, kwargs = rv = self.match(request) request.route, request.route_args, request.route_kwargs = rv if route.handler_adapter is None: handler = route.handler if isinstance(handler, six.string_types): if handler not in self.handlers: self.handlers[handler] = handler = import_string(handler) else: handler = self.handlers[handler] route.handler_adapter = self.adapt(handler) return route.handler_adapter(request, response)
def handle_exception(self, request, response, e): """Handles a uncaught exception occurred in :meth:`__call__`. Uncaught exceptions can be handled by error handlers registered in :attr:`error_handlers`. This is a dictionary that maps HTTP status codes to callables that will handle the corresponding error code. If the exception is not an ``HTTPException``, the status code 500 is used. The error handlers receive (request, response, exception) and can be a callable or a string in dotted notation to be lazily imported. If no error handler is found, the exception is re-raised. Based on idea from `Flask`_. :param request: A :class:`Request` instance. :param response: A :class:`Response` instance. :param e: The uncaught exception. :returns: The returned value from the error handler. """ if isinstance(e, HTTPException): code = e.code else: code = 500 handler = self.error_handlers.get(code) if handler: if isinstance(handler, six.string_types): self.error_handlers[code] = handler = import_string(handler) return handler(request, response, e) else: # Re-raise it to be caught by the WSGI app. raise
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
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 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 __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)
