我们从Python开源项目中,提取了以下50个代码示例,用于说明如何使用webapp2.RequestHandler()。
def _build_state_value(request_handler, user): """Composes the value for the 'state' parameter. Packs the current request URI and an XSRF token into an opaque string that can be passed to the authentication server via the 'state' parameter. Args: request_handler: webapp.RequestHandler, The request. user: google.appengine.api.users.User, The current user. Returns: The state value as a string. """ uri = request_handler.request.url token = xsrfutil.generate_token(xsrf_secret_key(), user.user_id(), action_id=str(uri)) return uri + ':' + token
def _create_flow(self, request_handler): """Create the Flow object. The Flow is calculated lazily since we don't know where this app is running until it receives a request, at which point redirect_uri can be calculated and then the Flow object can be constructed. Args: request_handler: webapp.RequestHandler, the request handler. """ if self.flow is None: redirect_uri = request_handler.request.relative_url( self._callback_path) # Usually /oauth2callback self.flow = OAuth2WebServerFlow( self._client_id, self._client_secret, self._scope, redirect_uri=redirect_uri, user_agent=self._user_agent, auth_uri=self._auth_uri, token_uri=self._token_uri, revoke_uri=self._revoke_uri, **self._kwargs)
def _create_flow(self, request_handler): """Create the Flow object. The Flow is calculated lazily since we don't know where this app is running until it receives a request, at which point redirect_uri can be calculated and then the Flow object can be constructed. Args: request_handler: webapp.RequestHandler, the request handler. """ if self.flow is None: redirect_uri = request_handler.request.relative_url( self._callback_path) # Usually /oauth2callback self.flow = client.OAuth2WebServerFlow( self._client_id, self._client_secret, self._scope, redirect_uri=redirect_uri, user_agent=self._user_agent, auth_uri=self._auth_uri, token_uri=self._token_uri, revoke_uri=self._revoke_uri, **self._kwargs)
def test_handle_exception_with_error(self): class HomeHandler(webapp2.RequestHandler): def get(self, **kwargs): raise TypeError() def handle_exception(request, response, exception): raise ValueError() app = webapp2.WSGIApplication([ webapp2.Route('/', HomeHandler, name='home'), ], debug=False) app.error_handlers[500] = handle_exception req = webapp2.Request.blank('/') rsp = req.get_response(app) self.assertEqual(rsp.status_int, 500)
def test_handle_exception_with_error_debug(self): class HomeHandler(webapp2.RequestHandler): def get(self, **kwargs): raise TypeError() def handle_exception(request, response, exception): raise ValueError() app = webapp2.WSGIApplication([ webapp2.Route('/', HomeHandler, name='home'), ], debug=True) app.error_handlers[500] = handle_exception req = webapp2.Request.blank('/') rsp = req.get_response(app) self.assertEqual(rsp.status_int, 500)
def test_custom_method(self): class MyHandler(webapp2.RequestHandler): def my_method(self): self.response.out.write('Hello, custom method world!') def my_other_method(self): self.response.out.write('Hello again, custom method world!') app = webapp2.WSGIApplication([ webapp2.Route('/', MyHandler, handler_method='my_method'), webapp2.Route('/other', MyHandler, handler_method='my_other_method'), ]) req = webapp2.Request.blank('/') rsp = req.get_response(app) self.assertEqual(rsp.status_int, 200) self.assertEqual(rsp.body, b'Hello, custom method world!') req = webapp2.Request.blank('/other') rsp = req.get_response(app) self.assertEqual(rsp.status_int, 200) self.assertEqual(rsp.body, b'Hello again, custom method world!')
def dispatch(self): self.session_store = sessions.get_store(request=self.request) if 'credentials' in self.session: credentials = oauth2client.client.OAuth2Credentials.from_json(self.session['credentials']) if credentials.access_token_expired: try: credentials.refresh(httplib2.Http()) self.session['credentials'] = credentials.to_json() except: self.session.pop('email', None) self.session.pop('user_id', None) self.session.pop('credentials', None) try: webapp2.RequestHandler.dispatch(self) finally: self.session_store.save_sessions(self.response)
def oauth_aware(self, method): """Decorator that sets up for OAuth 2.0 dance, but doesn't do it. Does all the setup for the OAuth dance, but doesn't initiate it. This decorator is useful if you want to create a page that knows whether or not the user has granted access to this application. From within a method decorated with @oauth_aware the has_credentials() and authorize_url() methods can be called. Args: method: callable, to be decorated method of a webapp.RequestHandler instance. """ def setup_oauth(request_handler, *args, **kwargs): if self._in_error: self._display_error_message(request_handler) return user = users.get_current_user() # Don't use @login_decorator as this could be used in a # POST request. if not user: request_handler.redirect(users.create_login_url( request_handler.request.uri)) return self._create_flow(request_handler) self.flow.params['state'] = _build_state_value(request_handler, user) self.credentials = self._storage_class( self._credentials_class, None, self._credentials_property_name, user=user).get() try: resp = method(request_handler, *args, **kwargs) finally: self.credentials = None return resp return setup_oauth
def has_credentials(self): """True if for the logged in user there are valid access Credentials. Must only be called from with a webapp.RequestHandler subclassed method that had been decorated with either @oauth_required or @oauth_aware. """ return self.credentials is not None and not self.credentials.invalid
def callback_application(self): """WSGI application for handling the OAuth 2.0 redirect callback. If you need finer grained control use `callback_handler` which returns just the webapp.RequestHandler. Returns: A webapp.WSGIApplication that handles the redirect back from the server during the OAuth 2.0 dance. """ return webapp.WSGIApplication([ (self.callback_path, self.callback_handler()) ])
def initialize(self, *a, **kw): """ This methods gets executed for each page and verfies user login status, using oookie information. """ webapp2.RequestHandler.initialize(self, *a, **kw) uid = self.read_secure_cookie('user_id') self.user = uid and User.by_id(int(uid))
def get(self): setTemplate(self, {"indexPage":True}, 'index.html') # class Error(webapp2.RequestHandler): # def get(self): # setTemplate(self, {}, 'error.html', _templateFolder='_templates/') # # self.redirect('/error.html', permanent=True)
def __init__(self, request, response): self.response = response response.write('I am not a RequestHandler but I work.')
def test_escaping(self): def get_req(uri): req = webapp2.Request.blank(uri) app.set_globals(app=app, request=req) handler = webapp2.RequestHandler(req, None) handler.app = req.app = app return req, handler req, handler = get_req('http://localhost:80/') uri = webapp2.uri_for('escape', name='with space') req, handler = get_req(uri) rsp = req.get_response(app) self.assertEqual(rsp.status_int, 200) self.assertEqual(rsp.body, b'with space') req, handler = get_req('http://localhost:80/') uri = webapp2.uri_for('escape', name='with+plus') req, handler = get_req(uri) rsp = req.get_response(app) self.assertEqual(rsp.status_int, 200) self.assertEqual(rsp.body, b'with plus') req, handler = get_req('http://localhost:80/') uri = webapp2.uri_for('escape', name='with/slash') req, handler = get_req(uri) rsp = req.get_response(app) self.assertEqual(rsp.status_int, 200) self.assertEqual(rsp.body, b'with/slash')
def test_encoding(self): class PostHandler(webapp2.RequestHandler): def post(self): foo = self.request.POST['foo'] if not foo: foo = 'empty' self.response.write(foo) app = webapp2.WSGIApplication([ webapp2.Route('/', PostHandler), ], debug=True) # foo with umlauts in the vowels. value = b'f\xc3\xb6\xc3\xb6' rsp = app.get_response( '/', POST={'foo': value}, headers=[('Content-Type', 'application/x-www-form-urlencoded; charset=utf-8')] ) self.assertEqual(rsp.unicode_body, u'föö') self.assertEqual(rsp.body, value) rsp = app.get_response( '/', POST={'foo': value}, headers=[('Content-Type', 'application/x-www-form-urlencoded')] ) self.assertEqual(rsp.unicode_body, u'föö') self.assertEqual(rsp.body, value)
def test_handle_exception_that_returns_response(self): class HomeHandler(webapp2.RequestHandler): def get(self, **kwargs): raise TypeError() app = webapp2.WSGIApplication([ webapp2.Route('/', HomeHandler, name='home'), ]) app.error_handlers[500] = 'tests.resources.handlers.handle_exception' req = webapp2.Request.blank('/') rsp = req.get_response(app) self.assertEqual(rsp.status_int, 200) self.assertEqual(rsp.body, b'Hello, custom response world!')
def test_return_is_not_wsgi_app(self): class HomeHandler(webapp2.RequestHandler): def get(self, **kwargs): return '' app = webapp2.WSGIApplication([ webapp2.Route('/', HomeHandler, name='home'), ], debug=False) req = webapp2.Request.blank('/') rsp = req.get_response(app) self.assertEqual(rsp.status_int, 500)
def test_old_app_new_handler(self): req = webapp2.Request.blank('/test/foo') rsp = req.get_response(app) self.assertEqual(rsp.status_int, 200) self.assertEqual(rsp.body, 'foo') req = webapp2.Request.blank('/test/bar') rsp = req.get_response(app) self.assertEqual(rsp.status_int, 200) self.assertEqual(rsp.body, 'bar') self.assertTrue(issubclass(OldStyleHandler, webapp.RequestHandler))
def authorize_url(self): """Returns the URL to start the OAuth dance. Must only be called from with a webapp.RequestHandler subclassed method that had been decorated with either @oauth_required or @oauth_aware. """ url = self.flow.step1_get_authorize_url() return str(url)