Skip to content

Instantly share code, notes, and snippets.

@knoguchi
Forked from aparrish/fitbit_auth_example.py
Created June 17, 2019 02:07
Show Gist options
  • Save knoguchi/2be22ba6c8b77a31cf1aab9f4954de3f to your computer and use it in GitHub Desktop.
Save knoguchi/2be22ba6c8b77a31cf1aab9f4954de3f to your computer and use it in GitHub Desktop.
A simple Tornado application that implements OAuth login and authenticated requests for Fitbit.
# simple tornado app that implements fitbit oauth login.
# requires tornado 2.4-ish. run on the command line like so:
# $ python fitbit_auth_example.py --port=<your port> \
# --fitbit_consumer_key=<your consumer key> \
# --fitbit_consumer_secret=<your consumer secret>
#
# make sure to set your fitbit app's callback URL to
# http://your-app-url.example.com/login
import tornado.web
import tornado.auth
import tornado.httpserver
import os
import urllib
import json
class FitbitApp(tornado.web.Application):
def __init__(self, options):
handlers = [
(r"/login", LoginHandler),
(r"/logout", LogoutHandler),
(r"/", HomeHandler),
]
settings = dict(
template_path=os.path.join(os.path.dirname(__file__), "templates"),
static_path=os.path.join(os.path.dirname(__file__), "static"),
cookie_secret=options.fitbit_consumer_secret + "salt",
fitbit_consumer_key=options.fitbit_consumer_key,
fitbit_consumer_secret=options.fitbit_consumer_secret
)
tornado.web.Application.__init__(self, handlers, **settings)
# I think I can see the RFC from here. *waves*
def build_oauth_header(params):
return "OAuth " + ", ".join(
['%s="%s"' % (k, urllib.quote(v)) for k, v in params.iteritems()])
class FitbitMixin(tornado.auth.OAuthMixin):
_OAUTH_REQUEST_TOKEN_URL = "http://api.fitbit.com/oauth/request_token"
_OAUTH_ACCESS_TOKEN_URL = "http://api.fitbit.com/oauth/access_token"
_OAUTH_AUTHORIZE_URL = "http://www.fitbit.com/oauth/authorize"
_OAUTH_AUTHENTICATE_URL = "http://www.fitbit.com/oauth/authenticate"
_OAUTH_NO_CALLBACKS = False
_OAUTH_VERSION="1.0"
def _oauth_consumer_token(self):
return {
'key': self.settings['fitbit_consumer_key'],
'secret': self.settings['fitbit_consumer_secret']
}
def _oauth_get_user(self, access_token, callback):
callback = self.async_callback(self._parse_user_response, callback)
self.fitbit_request("http://api.fitbit.com/1/user/-/profile.json",
access_token=access_token, callback=callback)
def _parse_user_response(self, callback, user):
user = json.loads(user)
callback(user)
def fitbit_request(self, url, callback, access_token=None, post_args=None,
**args):
if access_token:
all_args = dict()
all_args.update(args)
if post_args is not None:
all_args.update(post_args)
method = "POST" if post_args is not None else "GET"
oauth = self._oauth_request_parameters(
url, access_token, all_args, method=method)
if args:
url += "?" + urllib.urlencode(args)
callback = self.async_callback(self._on_fitbit_request, callback)
httpc = self.get_auth_http_client()
log(build_oauth_header(oauth))
if method == 'POST':
httpc.fetch(url, method="POST", body=urllib.urlencode(post_args),
callback=callback,
headers={'Authorization': build_oauth_header(oauth)})
else:
httpc.fetch(url, callback=callback,
headers={'Authorization': build_oauth_header(oauth)})
def _on_fitbit_request(self, callback, response):
log("got response: %s %s" % (str(response), response.body))
callback(response.body)
class LoginHandler(tornado.web.RequestHandler, FitbitMixin):
@tornado.web.asynchronous
def get(self):
if self.get_argument("oauth_token", None):
self.get_authenticated_user(self.async_callback(self._on_auth))
return
self.authorize_redirect()
def _on_auth(self, user):
log(user)
self.set_secure_cookie("u", json.dumps(user))
self.redirect("/")
class HomeHandler(tornado.web.RequestHandler, FitbitMixin):
@tornado.web.asynchronous
def get(self):
self.write("hi!")
user = self.get_secure_cookie("u")
if user:
user = json.loads(user)
self.write("<br>You are %s. <a href='/logout'>Logout</a>" % \
user['user']['displayName'])
self.fitbit_request("http://api.fitbit.com/1/user/-/activities/recent.json",
access_token=user['access_token'], callback=self._on_data)
else:
self.write("<br><a href='/login'>Login</a>")
self.finish()
def _on_data(self, response):
self.write("<br>" + response)
self.finish()
class LogoutHandler(tornado.web.RequestHandler):
def get(self):
self.set_secure_cookie("u", "")
self.redirect("/")
if __name__ == '__main__':
from tornado.options import define, options
define("port", help="listen for http requests on this port", type=int)
define("fitbit_consumer_key", help="fitbit consumer key")
define("fitbit_consumer_secret", help="fitbit consumer secret")
import logging
log = logging.info
tornado.options.parse_command_line()
http_server = tornado.httpserver.HTTPServer(FitbitApp(options))
http_server.listen(options.port)
log('starting ioloop')
tornado.ioloop.IOLoop.instance().start()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment