TornadoΒΆ

Use Tornado to serve token requests:

import tornado.web
import tornado.ioloop
from oauth2.store import LocalClientStore, LocalTokenStore
from oauth2 import AuthorizationController
from oauth2.tokengenerator import Uuid4
from oauth2.web import SiteAdapter
from oauth2.grant import ImplicitGrant, AuthorizationCodeGrant

class Request(object):
    """
    Wraps ``tornado.web.RequestHandler``.
    """
    def __init__(self, request_handler):
        self.request_handler = request_handler
        self.path = request_handler.request.path

    def get_param(self, name, default=None):
        return self._read_argument(name, default, source="GET")

    def post_param(self, name, default=None):
        return self._read_argument(name, default, source="POST")

    def _read_argument(self, name, default, source):
        if self.request_handler.request.method != source:
            return None
        try:
            return self.request_handler.get_argument(name)
        except tornado.web.MissingArgumentError:
            return default

class OAuth2Handler(tornado.web.RequestHandler):
    """
    Dispatches requests to an authorization controller
    """
    def initialize(self, controller):
        self.controller = controller

    def get(self):
        response = self._dispatch_request()

        self._map_response(response)

    def post(self):
        response = self._dispatch_request()

        self._map_response(response)

    def _dispatch_request(self):
        request = Request(request_handler=self)

        return self.controller.dispatch(request, environ={})

    def _map_response(self, response):
        for name, value in list(response.headers.items()):
            self.set_header(name, value)

        self.set_status(response.status_code)
        self.write(response.body)

class MySiteAdapter(SiteAdapter):
    def authenticate(self, request, environ, scopes):
        # Authenticate every request
        return {}

def main():
    # Initialize AuthorizationController as usual
    client_store = LocalClientStore()
    client_store.add_client(client_id="abc", client_secret="xyz",
                            redirect_uris=["http://localhost:8081/callback"])

    token_store = LocalTokenStore()

    auth_controller = AuthorizationController(
        access_token_store=token_store,
        auth_token_store=token_store,
        client_store=client_store,
        site_adapter=MySiteAdapter(),
        token_generator=Uuid4()
    )

    auth_controller.add_grant(AuthorizationCodeGrant())
    auth_controller.add_grant(ImplicitGrant())

    # Create your Tornado application and add the handler
    app = tornado.web.Application([
        (r'/authorize', OAuth2Handler, dict(controller=auth_controller))
    ])

    # Start the server
    app.listen(8888)
    tornado.ioloop.IOLoop.instance().start()

if __name__ == "__main__":
    main()