Query Badgr API for Expired Assertions

README.md

Badgr API Example: Querying for expired assertions

Notifications and related UX around expired or soon-to-expire badges will be available as a Badgr Pro feature soon. In the meantime, intrepid API clients may implement their own custom behavior around notifications by processing awarded assertions to determine which are expired.

Instructions for use

• Download the files to a directory on your local system.
• Activate a virtualenvironment for dependencies
• Install the python dependencies in requirements.txt with pip install -r requirements.txt
• Configure a settings_local.py file to expose BADGR_HOST (like https://api.badgr.io) and BADGR_API_KEY strings.
• You may obtain an API key with the quickstart instructions at badgr.org
• Enter a python interpreter: python
• Import the function: from assertions_expiration_report import *
• Run the function for a particular issuer you control: export_expiring_assertions('abc123abc123abc123a')
• Optionally, pass in a export filename and whether to skip past expirations like export_expiring_assertions('abc123abc123abc123a', filename='myexport.csv', include_past=False)

assertions_expiration_report.py

import csv
import datetime
from dateutil.parser import parse
from dateutil.tz import tzutc
import json
import requests

from settings_local import BADGR_HOST, BADGR_API_KEY

REQUEST_HEADERS = request_headers = {'Authorization': 'Bearer ' + BADGR_API_KEY}
DEFAULT_COLUMNS = ['url', 'email', 'issuedOn', 'expires', 'badge_entityId', 'badge_name']


def export_expiring_assertions(issuer_id, filename='expiration_report.csv', include_past=True):
    badge_response = requests.get(
        "{}/v2/issuers/{}/badgeclasses".format(BADGR_HOST, issuer_id),
        headers=REQUEST_HEADERS
    )
    badge_data = badge_response.json()['result']

    with open(filename, 'w') as f:
        now = datetime.datetime.now().replace(tzinfo=tzutc())
        per_page = 100
        next_url = "{}/v2/issuers/{}/assertions?num={}".format(BADGR_HOST, issuer_id, per_page)
        f.truncate()
        writer = csv.DictWriter(f, DEFAULT_COLUMNS)
        writer.writeheader()

        while next_url is not None:
            assertion_response = requests.get(
                next_url, headers=REQUEST_HEADERS)

            assertions = assertion_response.json()['result']

            for assertion in assertions:
                if assertion['expires'] is not None:
                    if include_past or parse(assertion['expires']) > now:
                        badgeclass_name = [b for b in badge_data if b.get('entityId') == assertion['badgeclass']][0]['name']
                        data = {
                            'url': assertion['openBadgeId'],
                            'email': assertion['recipient']['plaintextIdentity'].encode('utf-8'),
                            'issuedOn': assertion['issuedOn'],
                            'expires': assertion['expires'],
                            'badge_entityId': assertion['badgeclass'],
                            'badge_name': badgeclass_name.encode('utf-8')
                        }
                        writer.writerow(data)

            links = assertion_response.headers.get('link')
            pagination_links = None
            if links:
                pagination_links = requests.utils.parse_header_links(links)
            if links and pagination_links is not None:
                try:
                    next_url = [l.get('url') for l in pagination_links if l.get('rel') == 'next'][0]
                except IndexError:
                    next_url = None
            else:
                next_url = None

requirements.txt

python-dateutil==2.8.0
requests==2.22.0