Source code for oauth_dropins.instagram

"""Instagram OAuth drop-in.

Instagram API docs:

Almost identical to Facebook, except the access token request has `code`
and `grant_type` query parameters instead of just `auth_code`, the response
has a `user` object instead of `id`, and the call to GET_ACCESS_TOKEN_URL
is a POST instead of a GET.
TODO: unify them.
import logging
import os
import urllib.parse

from flask import abort, request
from import ndb

from . import facebook, views, models
from .webutil import flask_util, util
from .webutil.util import json_dumps, json_loads

logger = logging.getLogger(__name__)

INSTAGRAM_CLIENT_ID ='instagram_client_id')
INSTAGRAM_CLIENT_SECRET ='instagram_client_secret')
# instagram api url templates. can't (easily) use urlencode() because i want to
# keep the %(...)s placeholders as is and fill them in later in code.
GET_AUTH_CODE_URL = '&'.join((
    # redirect_uri here must be the same in the access token request!

[docs]class InstagramAuth(models.BaseAuth): """An authenticated Instagram user or page. Provides methods that return information about this user (or page) and make OAuth-signed requests to Instagram's HTTP-based APIs. Stores OAuth credentials in the datastore. See models.BaseAuth for usage details. Instagram-specific details: implements urlopen() but not api(). The key name is the Instagram username. """ auth_code = ndb.StringProperty(required=True) access_token_str = ndb.StringProperty(required=True) user_json = ndb.TextProperty(required=True)
[docs] def site_name(self): return 'Instagram'
[docs] def user_display_name(self): """Returns the Instagram username. """ return self.key_id()
[docs] def access_token(self): """Returns the OAuth access token string. """ return self.access_token_str
[docs] def urlopen(self, url, **kwargs): """Wraps urlopen() and adds OAuth credentials to the request. """ return models.BaseAuth.urlopen_access_token(url, self.access_token_str, **kwargs)
[docs]class Start(views.Start): """Starts Instagram auth. Requests an auth code and expects a redirect back. """ NAME = 'instagram' LABEL = 'Instagram' DEFAULT_SCOPE = 'basic'
[docs] def redirect_url(self, state=None): assert INSTAGRAM_CLIENT_ID and INSTAGRAM_CLIENT_SECRET, ( "Please fill in the instagram_client_id and instagram_client_secret " "files in your app's root directory.") # return GET_AUTH_CODE_URL % { 'client_id': INSTAGRAM_CLIENT_ID, # instagram uses + instead of , to separate scopes # 'scope': self.scope.replace(',', '+'), # TODO: CSRF protection identifier. 'redirect_uri': urllib.parse.quote_plus(self.to_url(state=state)), }
[docs] @classmethod def button_html(cls, *args, **kwargs): return super(cls, cls).button_html( *args, input_style='background-color: #EEEEEE; padding: 5px; padding-top: 8px; padding-bottom: 2px', **kwargs)
[docs]class Callback(views.Callback): """The auth callback. Fetches an access token, stores it, and redirects home. """
[docs] def dispatch_request(self): err = facebook.Callback.handle_error(self) if err: return err # auth_code = request.values['code'] data = { 'client_id': INSTAGRAM_CLIENT_ID, 'client_secret': INSTAGRAM_CLIENT_SECRET, 'code': auth_code, 'redirect_uri': self.request_url_with_state(), 'grant_type': 'authorization_code', } try: resp = util.requests_post(GET_ACCESS_TOKEN_URL, data=data) resp.raise_for_status() except BaseException as e: util.interpret_http_exception(e) raise try: data = json_loads(resp.text) except (ValueError, TypeError): logger.error(f'Bad response:\n{resp}', exc_info=True) flask_util.error('Bad Instagram response to access token request') if 'error_type' in resp: abort(502, f"{resp['error_type']} {data.get('code')} {data.get('error_message')}") access_token = data['access_token'] username = data['user']['username'] auth = InstagramAuth(id=username, auth_code=auth_code, access_token_str=access_token, user_json=json_dumps(data['user'])) auth.put() return self.finish(auth, state=request.values.get('state'))