Automatically or manually refreshing access token with flask_client on Google App Engine

Question

I am successfully able to authorize my application with a 3rd party OAuth2 provider (Xero), but have been unable to refresh the token, either automatically, or manually.

The documentation suggests authlib can do this automatically. I have tried two different approaches from the Authlib documentation, on the flask client docs they give an example of "Auto Update Token via Signal", and on the web client docs they register an "update_token" function.

Using either approach, there is never an attempt made to refresh the token, the request is passed to Xero with the expired token, I receive an error, and the only way to continue is to manually re-authorize the application with Xero.

Here is the relevant code for the "update_token" method from the web client docs:

#this never ends up getting called.
def save_xero_token(name,token,refresh_token=None,access_token=None,tenant_id=None):
    logging.info('Called save xero token.')
    #removed irrelevant code that stores token in NDB here.

cache = Cache()
oauth = OAuth(app,cache=cache)
oauth.register(name='xero',
               client_id = Meta.xero_consumer_client_id,
               client_secret = Meta.xero_consumer_secret,
               access_token_url = 'https://identity.xero.com/connect/token',
               authorize_url = 'https://login.xero.com/identity/connect/authorize',
               fetch_token = fetch_xero_token,
               update_token = save_xero_token,
               client_kwargs={'scope':' '.join(Meta.xero_oauth_scopes)},
              )

xero_tenant_id = 'abcd-123-placeholder-for-stackoverflow'
url = 'https://api.xero.com/api.xro/2.0/Invoices/ABCD-123-PLACEHOLDER-FOR-STACKOVERFLOW'
headers = {'Xero-tenant-id':xero_tenant_id,'Accept':'application/json'}

response = oauth.xero.get(url,headers=headers)    #works fine until token is expired.

I am storing my token in the following NDB model:

class OAuth2Token(ndb.Model):
    name = ndb.StringProperty()
    token_type = ndb.StringProperty()
    access_token = ndb.StringProperty()
    refresh_token = ndb.StringProperty()
    expires_at = ndb.IntegerProperty()
    xero_tenant_id = ndb.StringProperty()

    def to_token(self):
        return dict(
            access_token=self.access_token,
            token_type=self.token_type,
            refresh_token=self.refresh_token,
            expires_at=self.expires_at
        )

For completeness, here's how I store the initial response from Xero (which works fine):

@app.route('/XeroOAuthRedirect')
def xeroOAuthLanding():
    token = oauth.xero.authorize_access_token()
    connections_response = oauth.xero.get('https://api.xero.com/connections')
    connections = connections_response.json()
    for tenant in connections:
        print('saving first org, this app currently supports one xero org only.')
        save_xero_token('xero',token,tenant_id=tenant['tenantId'])

    return 'Authorized application with Xero'

How can I get automatic refreshing to work, and how can I manually trigger a refresh request when using the flask client, in the event automatic refreshing fails?

Answer 1

I believe I've found the problem here, and the root of it was the passing of a Cache (for temporary credential storage) when initializing OAuth:

cache = Cache()
oauth = OAuth(app,cache=cache)

When the cache is passed, it appears to preempt the update_token (and possibly fetch_token) parameters.

It should be simply:

oauth = OAuth(app)

oauth.register(name='xero',
               client_id = Meta.xero_consumer_client_id,
               client_secret = Meta.xero_consumer_secret,
               access_token_url = 'https://identity.xero.com/connect/token',
               authorize_url = 'https://login.xero.com/identity/connect/authorize',
               fetch_token = fetch_xero_token,
               update_token = save_xero_token,
               client_kwargs={'scope':' '.join(Meta.xero_oauth_scopes)},
              )

in addition, the parameters on my "save_xero_token" function needed to be adjusted to match the documentation, however this was not relevant to the original problem the question was addressing.