Advanced Webapp2 Google App Engine Example¶
In this tutorial we will create a Google App Engine Webapp2 application
that will be able to log users in with Facebook, Twitter and OpenID
and we will use the Credentials of an authenticated user to
post tweets and Facebook statuses on the user’s behalf.
You can download all the source files we are about to create here.
First create the Config dictionary where you set up all the providers you want to use.
Yo will need the consumer_key and consumer_secret which you can get
here for Facebook and
here for Twitter.
Note
Facebook and other OAuth 2.0 providers require a redirect URI
which should be the URL of the login request handler
which we will create in this tutorial and whose walue in our case will be
https://[hostname]:[port]/login/fb for Facebook.
# -*- coding: utf-8 -*-
# config.py
from authomatic.providers import oauth2, oauth1, openid, gaeopenid
import authomatic
CONFIG = {
'tw': { # Your internal provider name
# Provider class
'class_': oauth1.Twitter,
# Twitter is an AuthorizationProvider so we need to set several other
# properties too:
'consumer_key': '####################',
'consumer_secret': '####################',
'id': authomatic.provider_id()
},
'fb': {
'class_': oauth2.Facebook,
# Facebook is AuthorizationProvider too.
'consumer_key': '####################',
'consumer_secret': '####################',
'id': authomatic.provider_id(),
# We need the "publish_stream" scope to post to users timeline,
# the "offline_access" scope to be able to refresh credentials,
# and the other scopes to get user info.
'scope': [
'publish_stream',
'offline_access',
'user_about_me',
'email'
],
},
'gae_oi': {
# OpenID provider based on Google App Engine Users API.
# Works only on GAE and returns only the id and email of a user.
# Moreover, the id is not available in the development environment!
'class_': gaeopenid.GAEOpenID,
},
'oi': {
# OpenID provider based on the python-openid library.
# Works everywhere, is flexible, but requires more resources.
'class_': openid.OpenID,
'store': openid.SessionOpenIDStore,
}
}
Create the main.py module and import what’s needed.
# -*- coding: utf-8 -*-
# main.py
import urllib
import webapp2
from authomatic import Authomatic
from authomatic.adapters import Webapp2Adapter
Make an instance of the Authomatic class and pass
the Config together with a random secret string
used for session and CSRF token generation to it’s constructor.
# Setup Authomatic.
Create a simple request handler which accepts GET and POST HTTP methods and
receives the provider_name URL variable.
Log the user in by calling the Authomatic.login() method.
If there is LoginResult.user, the login procedure was successful
and we can welcome the user.
result = authomatic.login(Webapp2Adapter(self), provider_name)
if result:
if result.user:
Save the user’s name and ID to cookies so we can use them in other handlers. We use cookies only for simplicity of the example, in real app you will probably use some User datamodel.
# Save the user name and ID to cookies that we can use it in
If the user has logged in with Facebook or Twitter, he/she gave us Credentials.
self.response.set_cookie('user_id', result.user.id)
You can serialize the Credentials into a lightweight URL-safe string.
Store also the serialized credentials to a cookie.
'user_name', urllib.quote(
result.user.name))
Store also the possible error message to a cookie.
if result.user.credentials:
# Serialize credentials and store it as well.
Redirect the user to the Home handler which we are going to create next.
self.response.set_cookie(
The Home handler only needs a GET method.
elif result.error:
self.response.set_cookie(
Create links to the Login handler.
result.error.message))
Retrieve the stored values from cookies.
class Home(webapp2.RequestHandler):
def get(self):
# Create links to the Login handler.
Handle possible errors.
self.response.write('<a href="login/tw">Twitter</a>')
If there is no error, there must be a user ID.
# Retrieve values from cookies.
serialized_credentials = self.request.cookies.get('credentials')
Let’s look at what we can do with the Credentials.
user_name = urllib.unquote(self.request.cookies.get('user_name', ''))
We can deserialize them.
They know the provider name which we defined in the Config.
self.response.write('<p>Damn that error: {0}</p>'.format(error))
elif user_id:
self.response.write('<h1>Hi {0}</h1>'.format(user_name))
if serialized_credentials:
Credentials issued by OAuth 2.0 providers have limited lifetime.
We can test whether they are still valid.
credentials = authomatic.credentials(serialized_credentials)
Whether they expire soon.
The remaining number of seconds till they expire.
Note
If the number is negative, the Credentials will never expire.
self.response.write("""
Their expiration date.
Note
If Credentials will never expire it returns None.
<p>
We can refresh the Credentials without the user while they are valid.
If they are expired we only can get new Credentials by repeating the login procedure
with Authomatic.login().
Inform the user about his/her Credentials and create links to
Refresh, Action and Logout handlers, which we are going to create next.
credentials.
</p>
""".format(
dict(
fb='Facebook',
tw='Twitter'
)[credentials.provider_name])
)
valid = 'still' if credentials.valid else 'not anymore'
expire_soon = 'less' if credentials.expire_soon(
60 * 60 * 24) else 'more'
remaining = credentials.expire_in
expire_on = credentials.expiration_date
self.response.write("""
<p>
They are <b>{0}</b> valid and
will expire in <b>{0}</b> than one day
(in <b>{0}</b> seconds to be precise).
It will be on <b>{0}</b>.
</p>
""".format(valid, expire_soon, remaining, expire_on))
if credentials.valid:
self.response.write("""
Create the Refresh handler, retrieve the serialized Credentials from
the cookie, deserialize them and get their expiration date.
<p>Moreover, we can do powerful stuff with them.</p>
<a href="action/{0}">Show me what you can do!</a>
""".format(credentials.provider_name))
else:
self.response.write("""
<p>
Repeat the <b>login procedure</b>to get new
Refresh the Credentials with the Credentials.refresh() method.
It returns a Response object, but only if the Credentials support refreshment.
Otherwise the method returns None.
Note
Only OAuth 2.0 Credentials support refreshment but it also depends on the provider’s
implementation, e.g. Facebook allows you to refresh Credentials only if you requested the
offline_access scope in the Config.
</p>
<a href="login/{0}">Refresh</a>
""".format(credentials.provider_name))
self.response.write('<p>We can also log you out.</p>')
self.response.write('<a href="logout">OK, log me out!</a>')
class Refresh(webapp2.RequestHandler):
def get(self):
self.response.write('<a href="..">Home</a>')
serialized_credentials = self.request.cookies.get('credentials')
credentials = authomatic.credentials(serialized_credentials)
old_expiration = credentials.expiration_date
response = credentials.refresh(force=True)
if response:
new_expiration = credentials.expiration_date
if response.status == 200:
self.response.write("""
<p>
The most interesting things will happen in the Action handler.
Let’s first create a method for GET requests which will accept the provider_name URL variable.
Inside create a simple HTML form which submits to this handler’s POST method.
<b>{0}</b> to <b>{0}</b>.
</p>
""".format(old_expiration, new_expiration))
else:
self.response.write("""
<p>Refreshment failed!</p>
<p>Status code: {0}</p>
<p>Error message:</p>
<pre>{0}</pre>
""".format(response.status, response.content))
else:
self.response.write(
'<p>Your credentials don\'t support refreshment!</p>')
self.response.write('<a href="">Try again!</a>')
In the POST method, retrieve the message from POST parameters and the values from cookies.
class Action(webapp2.RequestHandler):
def get(self, provider_name):
text = 'post a status on your Facebook timeline'
elif provider_name == 'tw':
text = 'tweet'
Let’s first post a status to the user’s Facebook timeline by accessing the Facebook Graph API endpoint.
Note
You need to include the "publish_stream" scope in the "fb" section of the config
to be able to post to the user’s timeline.
Prepare the URL.
self.response.write("""
<p>We can {0} on your behalf.</p>
Access the protected resource of the user by calling the Authomatic.access() method.
You must pass it the Credentials (normal or serialized) and the URL.
The URL can contain query string parameters, but
you can also pass them to the method as a dictionary.
The method returns a Response object.
value="Have you got a bandage?" />
<input type="submit" value="Do it!">
</form>
Parse the Response. The Response.data is a data structure (list or dictionary)
parsed from the Response.content which usually is JSON.
def post(self, provider_name):
self.response.write('<a href="..">Home</a>')
# Retrieve the message from POST parameters and the values from
# cookies.
message = self.request.POST.get('message')
serialized_credentials = self.request.cookies.get('credentials')
user_id = self.request.cookies.get('user_id')
if provider_name == 'fb':
# Prepare the URL for Facebook Graph API.
Do the same with Twitter.
Note
You need to set the Application Type of your Twitter app to Read and Write to be able to post tweets on the user’s behalf.
# Access user's protected resource.
response = authomatic.access(serialized_credentials, url,
params=dict(message=message),
method='POST')
# Parse response.
post_id = response.data.get('id')
error = response.data.get('error')
if error:
self.response.write(
'<p>Damn that error: {0}!</p>'.format(error))
elif post_id:
self.response.write(
'<p>You just posted a status with id ' +
'{0} to your Facebook timeline.<p/>'.format(post_id))
else:
self.response.write(
'<p>Damn that unknown error! Status code: {0}</p>'.format(
response.status
))
elif provider_name == 'tw':
Let the user repeat the action.
serialized_credentials,
url='https://api.twitter.com/1.1/statuses/update.json',
params=dict(status=message),
method='POST')
error = response.data.get('errors')
The Logout handler is pretty simple. We just need to delete all those cookies we set and redirect the user to the Home handler.
if error:
self.response.write(
elif tweet_id:
self.response.write("""
<p>
""".format(tweet_id))
Finally create routes to all those handlers,
Damn that unknown error! Status code: {0}
</p>
""".format(response.status))
# Let the user repeat the action.
and instantiate the Webapp2 WSGI application.
<input type="text" name="message" />
Don’t forget to create the app.yaml file.
# app.yaml
application: authomatic-credentials
version: 0
runtime: python27
api_version: 1
threadsafe: true
handlers:
- url: /.*
script: main.app
That’s it. Now just run the application.
$ python dev_appserver.py [path to the root folder of this app]
Tip
Some of the providers don’t support authorization from apps running on localhost.
Probably the best solution is to use an arbitrary domain as an alias of the 127.0.0.1 IP address.
You can do this on UNIX systems by adding an alias to the /etc/hosts file.
# /etc/hosts
127.0.0.1 localhost
127.0.0.1 yourlocalhostalias.com
You can do this on Windows systems by adding an alias in the C:\Windows\system32\drivers\etc\hosts file.
127.0.0.1 yourlocalhostalias.com
And here is the complete app. Remember that you can download all the files we just created from GitHub.
# -*- coding: utf-8 -*-
# main.py
import urllib
import webapp2
from authomatic import Authomatic
from authomatic.adapters import Webapp2Adapter
from config import CONFIG
# Setup Authomatic.
authomatic = Authomatic(config=CONFIG, secret='a-long-secret-string')
class Login(webapp2.RequestHandler):
def any(self, provider_name):
# Log the user in.
result = authomatic.login(Webapp2Adapter(self), provider_name)
if result:
if result.user:
result.user.update()
self.response.write('<h1>Hi {0}</h1>'.format(result.user.name))
# Save the user name and ID to cookies that we can use it in
# other handlers.
self.response.set_cookie('user_id', result.user.id)
self.response.set_cookie(
'user_name', urllib.quote(
result.user.name))
if result.user.credentials:
# Serialize credentials and store it as well.
serialized_credentials = result.user.credentials.serialize() # noqa
self.response.set_cookie(
'credentials', serialized_credentials)
elif result.error:
self.response.set_cookie(
'error', urllib.quote(
result.error.message))
self.redirect('/')
class Home(webapp2.RequestHandler):
def get(self):
# Create links to the Login handler.
self.response.write('Login with <a href="login/fb">Facebook</a> or ')
self.response.write('<a href="login/tw">Twitter</a>')
# Retrieve values from cookies.
serialized_credentials = self.request.cookies.get('credentials')
user_id = self.request.cookies.get('user_id')
user_name = urllib.unquote(self.request.cookies.get('user_name', ''))
error = urllib.unquote(self.request.cookies.get('error', ''))
if error:
self.response.write('<p>Damn that error: {0}</p>'.format(error))
elif user_id:
self.response.write('<h1>Hi {0}</h1>'.format(user_name))
if serialized_credentials:
# Deserialize credentials.
credentials = authomatic.credentials(serialized_credentials)
self.response.write("""
<p>
You are logged in with <b>{0}</b> and we have your
credentials.
</p>
""".format(
dict(
fb='Facebook',
tw='Twitter'
)[credentials.provider_name])
)
valid = 'still' if credentials.valid else 'not anymore'
expire_soon = 'less' if credentials.expire_soon(
60 * 60 * 24) else 'more'
remaining = credentials.expire_in
expire_on = credentials.expiration_date
self.response.write("""
<p>
They are <b>{0}</b> valid and
will expire in <b>{0}</b> than one day
(in <b>{0}</b> seconds to be precise).
It will be on <b>{0}</b>.
</p>
""".format(valid, expire_soon, remaining, expire_on))
if credentials.valid:
self.response.write("""
<p>We can refresh them while they are valid.</p>
<a href="refresh">OK, refresh them!</a>
<p>Moreover, we can do powerful stuff with them.</p>
<a href="action/{0}">Show me what you can do!</a>
""".format(credentials.provider_name))
else:
self.response.write("""
<p>
Repeat the <b>login procedure</b>to get new
credentials.
</p>
<a href="login/{0}">Refresh</a>
""".format(credentials.provider_name))
self.response.write('<p>We can also log you out.</p>')
self.response.write('<a href="logout">OK, log me out!</a>')
class Refresh(webapp2.RequestHandler):
def get(self):
self.response.write('<a href="..">Home</a>')
serialized_credentials = self.request.cookies.get('credentials')
credentials = authomatic.credentials(serialized_credentials)
old_expiration = credentials.expiration_date
response = credentials.refresh(force=True)
if response:
new_expiration = credentials.expiration_date
if response.status == 200:
self.response.write("""
<p>
Credentials were refresshed successfully.
Their expiration date was extended from
<b>{0}</b> to <b>{0}</b>.
</p>
""".format(old_expiration, new_expiration))
else:
self.response.write("""
<p>Refreshment failed!</p>
<p>Status code: {0}</p>
<p>Error message:</p>
<pre>{0}</pre>
""".format(response.status, response.content))
else:
self.response.write(
'<p>Your credentials don\'t support refreshment!</p>')
self.response.write('<a href="">Try again!</a>')
class Action(webapp2.RequestHandler):
def get(self, provider_name):
if provider_name == 'fb':
text = 'post a status on your Facebook timeline'
elif provider_name == 'tw':
text = 'tweet'
self.response.write("""
<a href="..">Home</a>
<p>We can {0} on your behalf.</p>
<form method="post">
<input type="text" name="message"
value="Have you got a bandage?" />
<input type="submit" value="Do it!">
</form>
""".format(text))
def post(self, provider_name):
self.response.write('<a href="..">Home</a>')
# Retrieve the message from POST parameters and the values from
# cookies.
message = self.request.POST.get('message')
serialized_credentials = self.request.cookies.get('credentials')
user_id = self.request.cookies.get('user_id')
if provider_name == 'fb':
# Prepare the URL for Facebook Graph API.
url = 'https://graph.facebook.com/{0}/feed'.format(user_id)
# Access user's protected resource.
response = authomatic.access(serialized_credentials, url,
params=dict(message=message),
method='POST')
# Parse response.
post_id = response.data.get('id')
error = response.data.get('error')
if error:
self.response.write(
'<p>Damn that error: {0}!</p>'.format(error))
elif post_id:
self.response.write(
'<p>You just posted a status with id ' +
'{0} to your Facebook timeline.<p/>'.format(post_id))
else:
self.response.write(
'<p>Damn that unknown error! Status code: {0}</p>'.format(
response.status
))
elif provider_name == 'tw':
response = authomatic.access(
serialized_credentials,
url='https://api.twitter.com/1.1/statuses/update.json',
params=dict(status=message),
method='POST')
error = response.data.get('errors')
tweet_id = response.data.get('id')
if error:
self.response.write(
'<p>Damn that error: {0}!</p>'.format(error))
elif tweet_id:
self.response.write("""
<p>
You just tweeted a tweet with id {0}.
</p>
""".format(tweet_id))
else:
self.response.write("""
<p>
Damn that unknown error! Status code: {0}
</p>
""".format(response.status))
# Let the user repeat the action.
self.response.write("""
<form method="post">
<input type="text" name="message" />
<input type="submit" value="Do it again!">
</form>
""")
class Logout(webapp2.RequestHandler):
def get(self):
# Delete cookies.
self.response.delete_cookie('user_id')
self.response.delete_cookie('user_name')
self.response.delete_cookie('credentials')
# Redirect home.
self.redirect('./')
# Create the routes.
ROUTES = [webapp2.Route(r'/login/<:.*>', Login, handler_method='any'),
webapp2.Route(r'/refresh', Refresh),
webapp2.Route(r'/action/<:.*>', Action),
webapp2.Route(r'/logout', Logout),
webapp2.Route(r'/', Home)]
# Instantiate the WSGI application.
app = webapp2.WSGIApplication(ROUTES, debug=True)
