Skip to content

Instantly share code, notes, and snippets.

@dmahugh
Created June 12, 2017 16:20
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save dmahugh/bdd2abb9f00c34bdb1d8c1707aed4ba7 to your computer and use it in GitHub Desktop.
Save dmahugh/bdd2abb9f00c34bdb1d8c1707aed4ba7 to your computer and use it in GitHub Desktop.
This sample is part of our exploration of various approaches to starter samples for working with Microsoft Graph. It requires Requests, Bottle, and Python 3.x. It's a work in progress - we'll be publishing a more comprehensive version with complete documentation soon, but posting this here for anyone who'd like to try it out.
"""Copyright (c) Microsoft Corporation. All rights reserved.
Licensed under the MIT License.
"""
import base64
import json
import os
import time
import urllib.parse
import uuid
import bottle
from bottle import redirect, request, route, template, view
import requests
class Connect(object):
"""Auth handler for AAD/MSA authentication."""
def __init__(self):
self.app_id = 'your_app_id_here'
self.app_secret = 'your_app_secret_here'
self.state = ''
self.access_token = None
self.loggedin = False
self.loggedin_name = ''
self.loggedin_email = ''
self.loggedin_photo = None
def authorized(self):
"""Use received authorization code to request a token from Azure AD."""
# Verify that this authorization attempt came from this app, by checking
# the received state, which should match the state (uuid) sent with our
# authorization request.
if self.state != request.query.state:
raise Exception(' -> SHUTTING DOWN: state mismatch' + \
'\n\nState SENT: {0}\n\nState RECEIVED: {1}'. \
format(str(self.state), str(request.query.state)))
self.state = '' # reset session state to prevent re-use
# try to fetch an access token
token_response = self.fetch_token(request.query.code)
if not token_response:
# no response
print(' -> auth.Connect: request for access token failed')
return redirect('/')
if not token_response.ok:
# error response
return token_response.text
# set properties for current authenticated user
me_response = self.get('me')
me_data = me_response.json()
if 'error' in me_data:
print(' -> auth.Connect: /me endpoint returned an error ... ' + \
str(me_data))
self.loggedin_name = me_data['displayName']
self.loggedin_email = me_data['userPrincipalName']
# save profile photo
profile_pic = self.get('me/photo/$value', stream=True)
if profile_pic.ok:
self.loggedin_photo = base64.b64encode(profile_pic.raw.read())
else:
self.loggedin_photo = None # no profile photo available
return redirect('/')
def fetch_token(self, authcode):
"""attempt to fetch an access token, using specified authorization code.
"""
response = requests.post( \
'https://login.microsoftonline.com/common/oauth2/v2.0/token', \
data=dict(client_id=self.app_id,
client_secret=self.app_secret,
grant_type='authorization_code',
code=authcode,
redirect_uri='http://localhost:5000/login/authorized'))
if self.save_token(response):
return response # valid token returned and saved
else:
return None # token request failed
def get(self, endpoint, stream=False, jsononly=False):
"""Wrapper for authenticated HTTP GET from API endpoint.
endpoint = relative URL (e.g., "me/contacts")
stream = Requests stream argument; e.g., use True for image data
jsononly = if True, the JSON 'value' is returned instead of the response
object
"""
response = requests.get( \
urllib.parse.urljoin('https://graph.microsoft.com/beta/',endpoint),
headers={'User-Agent' : 'graph-python-quickstart/1.0',
'Authorization' : 'Bearer {0}'.format(self.access_token),
'Accept' : 'application/json',
'Content-Type' : 'application/json',
'client-request-id' : str(uuid.uuid4()), # unique identifier
'return-client-request-id' : 'true'},
stream=stream)
if jsononly:
return response.json().get('value', None)
else:
return response
def login(self, redirect_to):
"""Ask user to authenticate via web interface."""
self.state = str(uuid.uuid4()) # used to verify source of auth request
print(' -> auth.Connect: asking user to authenticate')
redirect( \
'https://login.microsoftonline.com/common/oauth2/v2.0/authorize/' + \
'?response_type=code&client_id=' + self.app_id + \
'&redirect_uri=http://localhost:5000/login/authorized' + \
'&scope=Mail.Read%20User.Read' + \
'&state=' + self.state, 302)
def logout(self):
"""Close current connection and redirect to specified route."""
self.access_token = None
self.loggedin = False
redirect('/')
def save_token(self, response):
"""Save a new access token and related metadata.
response = response object from the token request
"""
jsondata = response.json()
if not 'access_token' in jsondata:
# no access token found in the response
print(' -> auth.Connect: request for access token failed')
self.logout()
return False
self.access_token = jsondata['access_token']
self.loggedin = True
print(' -> auth.Connect: access token acquired ({0} bytes)'. \
format(len(self.access_token)))
return True
def render_table(dataset, columns):
"""Convert list of dictionaries to HTML string for display."""
coltuples = [column.split(':') for column in columns.split(' ')]
html_strings = ['<table class = "sample">']
html_strings.append('<tr>')
for fldname, _ in coltuples:
html_strings.append('<th>' + fldname.capitalize() + '</th>')
html_strings.append('</tr>')
for entity in dataset:
html_strings.append('<tr>')
for fldname, nchars in coltuples:
fldsize = int(nchars)
if fldname in entity and entity[fldname]:
html_strings.append( \
'<td>' + str(entity[fldname])[:fldsize].ljust(fldsize) + '</td>')
else:
html_strings.append('<td></td>') # missing value
html_strings.append('</tr>')
html_strings.append('</table>')
return ''.join(html_strings)
msgraph = Connect()
@route('/')
@view('homepage')
def home():
"""Render the home page."""
if msgraph.loggedin:
response = msgraph.get('me/mailFolders/Inbox/messages')
jsondata = response.json().get('value', None)
emails = []
for email in jsondata:
email_date = email['receivedDateTime'][:10]
sender = '{0} ({1})'.format(email['from']['emailAddress']['name'],
email['from']['emailAddress']['address'])
subject = email['subject']
emails.append(dict(date=email_date, sender=sender, subject=subject))
return dict(sample='INBOX',
sampledata=render_table(emails, 'date:10 sender:35 subject:35'))
else:
return dict(sample=None, sampledata=None)
@route('/login')
def login():
"""Prompt user to authenticate."""
msgraph.login()
@route('/login/authorized')
def authorized():
"""Fetch access token for authenticated user."""
return msgraph.authorized()
@route('/logout')
def logout():
"""Log out from MS Graph connection."""
msgraph.logout()
if __name__ == '__main__':
@bottle.route('/static/<filepath:path>')
def server_static(filepath):
"""Handler for static files, used with the development server."""
PROJECT_ROOT = os.path.abspath(os.path.dirname(__file__))
STATIC_ROOT = os.path.join(PROJECT_ROOT, 'static').replace('\\', '/')
return bottle.static_file(filepath, root=STATIC_ROOT)
bottle.run(app=bottle.app(), server='wsgiref', host='localhost', port=5000)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment