Skip to content

Instantly share code, notes, and snippets.

@kieransnapes
Last active January 22, 2024 13:18
Show Gist options
  • Save kieransnapes/b5bc80ac16c8c154646a8595168bc76b to your computer and use it in GitHub Desktop.
Save kieransnapes/b5bc80ac16c8c154646a8595168bc76b to your computer and use it in GitHub Desktop.
xero oauth2 with django
import os
from functools import wraps
from django.urls import reverse
from django.shortcuts import redirect
from xero_python.api_client import ApiClient, serialize
from xero_python.api_client.configuration import Configuration
from xero_python.api_client.oauth2 import OAuth2Token
from requests_oauthlib import OAuth2Session
from authlib.integrations.django_client import OAuth, token_update
from django.core.cache import cache
from django.http import HttpResponse
from django.dispatch import receiver
#for dev
os.environ["OAUTHLIB_INSECURE_TRANSPORT"] = "1"
oauth = OAuth()
xero = oauth.register(
name="xero",
version="2",
client_id="client_id",
client_secret="client_secret",
endpoint_url="https://api.xero.com/",
authorization_url="https://login.xero.com/identity/connect/authorize",
access_token_url="https://identity.xero.com/connect/token",
refresh_token_url="https://identity.xero.com/connect/token",
scope="offline_access openid profile email accounting.transactions "
"accounting.transactions.read accounting.reports.read "
"accounting.journals.read accounting.settings accounting.settings.read "
"accounting.contacts accounting.contacts.read accounting.attachments "
"accounting.attachments.read assets projects"
)
api_client = ApiClient(
Configuration(
debug=True, # app.config["DEBUG"],
oauth2_token=OAuth2Token(
client_id="client_id", client_secret="client_secret"
),
),
pool_threads=1,
)
def login(request):
redirect_uri = request.build_absolute_uri('authorise/')
return oauth.xero.authorize_redirect(request, redirect_uri)
def oauth_callback(request):
try:
token = oauth.xero.authorize_access_token(request)
except Exception as e:
print(e)
raise
if token is None:
return HttpResponse("Access Denied")
store_xero_oauth2_token(token)
return redirect('/backend/tenants/')
def logout():
store_xero_oauth2_token(None)
return redirect('/')
def xero_token_required(function):
@wraps(function)
def decorator(*args, **kwargs):
xero_token = obtain_xero_oauth2_token()
time = datetime.datetime.now().timestamp()
if time > xero_token['expires_at']:
new_token = oauth.xero.fetch_access_token(
refresh_token=xero_token['refresh_token'],
grant_type='refresh_token')
store_xero_oauth2_token(new_token)
xero_token = new_token
if not xero_token:
return redirect(reverse("oauth"))
return function(*args, **kwargs)
return decorator
@api_client.oauth2_token_getter
def obtain_xero_oauth2_token():
return cache.get('token')
@api_client.oauth2_token_saver
def store_xero_oauth2_token(token):
cache.set('token', token, None)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment