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
python类RequestHandler()的实例源码
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 _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 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 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 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 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 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)
