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 http://[hostname]:[port]/login/fb for Facebook.

# 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.

# main.py

import urllib

import webapp2
from authomatic import Authomatic
from authomatic.adapters import Webapp2Adapter

from config import CONFIG

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.

authomatic = Authomatic(config=CONFIG, secret='a-long-secret-string')

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.

class Login(webapp2.RequestHandler):
    def any(self, provider_name):
        result = authomatic.login(Webapp2Adapter(self), provider_name)

If there is LoginResult.user, the login procedure was successful and we can welcome the user.

        if result:
            if result.user:
                result.user.update()
                self.response.write('<h1>Hi {0}</h1>'.format(result.user.name))

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.

                self.response.set_cookie('user_id', result.user.id)
                self.response.set_cookie('user_name', urllib.quote(result.user.name))

If the user has logged in with Facebook or Twitter, he/she gave us Credentials.

                if result.user.credentials:

You can serialize the Credentials into a lightweight URL-safe string. Store also the serialized credentials to a cookie.

                    serialized_credentials = result.user.credentials.serialize()
                    self.response.set_cookie('credentials', serialized_credentials)

Store also the possible error message to a cookie.

            elif result.error:
                self.response.set_cookie('error', urllib.quote(result.error.message))

Redirect the user to the Home handler which we are going to create next.

            self.redirect('/')

The Home handler only needs a GET method.

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 the stored 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', ''))

Handle possible errors.

        if error:
            self.response.write('<p>Damn that error: {0}</p>'.format(error))

If there is no error, there must be a user ID.

        elif user_id:
            self.response.write('<h1>Hi {0}</h1>'.format(user_name))

Let’s look at what we can do with the Credentials.

            if serialized_credentials:

We can deserialize them.

                credentials = authomatic.credentials(serialized_credentials)

They know the provider name which we defined in the Config.

                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]))

Credentials issued by OAuth 2.0 providers have limited lifetime. We can test whether they are still valid.

                valid = 'still' if credentials.valid else 'not anymore'

Whether they expire soon.

                expire_soon = 'less' if credentials.expire_soon(60 * 60 * 24) else 'more'

The remaining number of seconds till they expire.

Note

If the number is negative, the Credentials will never expire.

                remaining = credentials.expire_in

Their expiration date.

Note

If Credentials will never expire it returns None.

                expire_on = credentials.expiration_date

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.

                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>')

Create the Refresh handler, retrieve the serialized Credentials from the cookie, deserialize them and get their expiration date.

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

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.

        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>')

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.

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))

In the POST method, retrieve the message from POST parameters and the values from cookies.

    def post(self, provider_name):
        self.response.write('<a href="..">Home</a>')
        
        message = self.request.POST.get('message')        
        serialized_credentials = self.request.cookies.get('credentials')
        user_id = self.request.cookies.get('user_id')

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.

        if provider_name == 'fb':
            url = 'https://graph.facebook.com/{0}/feed'.format(user_id)

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.

            response = authomatic.access(serialized_credentials, url,
                                         params=dict(message=message),
                                         method='POST')

Parse the Response. The Response.data is a data structure (list or dictionary) parsed from the Response.content which usually is JSON.

            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))

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.

        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>
        """)

The Logout handler is pretty simple. We just need to delete all those cookies we set and redirect the user to the Home handler.

class Logout(webapp2.RequestHandler):
    def get(self):
        self.response.delete_cookie('user_id')
        self.response.delete_cookie('user_name')
        self.response.delete_cookie('credentials')
        self.redirect('./')

Finally create routes to all those handlers,

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)]

and instantiate the Webapp2 WSGI application.

app = webapp2.WSGIApplication(ROUTES, debug=True)

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

And here is the complete app. Remember that you can download all the files we just created from GitHub.

# 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()
                    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)
Fork me on GitHub