Python Wrapper for the QuickSchools REST API. Work in progress.

qs.py

#!/Library/Frameworks/Python.framework/Versions/2.7/bin/python

'''
various tools for working with the QuickSchools API
'''

import sys
import os
import json
import random
import requests
import time
import api_logging

# api key should be stored in separate file and pasted in
api_key = 'API KEY HERE'
use_live_server = True

WAIT_THRESHOLD = 5
WAIT_TIME = 1

base_uri = None
sections = None
_error_count = 0
wait_count = 0


# =================
# = REST API calls
# ==================
"""All API functions return a dictionary or a list of dictionaries"""


def get_sections(critical=False):
    """a little lazy instantiation up in here to avoid lots of GETs"""
    global sections

    if not sections:
        r = Request()
        r.uri = "/sections"
        r.description = "GET all sections"
        r.critical = critical
        r.make_request()
        sections = r.data

    return sections


def get_all_section_ids(critical=False):
    sections = get_all_sections()

    section_ids = []
    for section in sections:
        section_ids.append(section['id'])

    return section_ids


def get_section(section_id, critical=False):
    r = Request()
    r.description = "GET section"
    r.uri = "/sections/{}".format(section_id)
    r.critical = critical
    r.make_request()

    return r.data


def get_section_enrollments(section_id, critical=False):
    r = Request()
    r.description = "GET section enrollments"
    r.uri = "/sectionenrollments/{}".format(section_id)
    r.critical = critical
    r.make_request()

    return r.data


def get_assignments(section_id, critical=False):
    r = Request()
    r.description = "GET assignments"
    r.uri = "/sections/{}/assignments".format(section_id)
    r.critical = critical
    r.make_request()

    return r.data
    
def delete_assignment(section_id, assignment_id, critical=False):
    r = Request()
    r.description = "DELETE assignment"
    r.uri = "/sections/{}/assignments/{}".format(section_id, assignment)
    r.critical = critical
    r.function = requests.delete
    r.make_request()
    
    return r.data()


def get_grades(section_id, assignment_id, critical=False):
    r = Request()
    r.description = "GET grades"
    r.uri = "/grades"
    r.params = {'sectionId': section_id, 'assignmentId': assignment_id}
    r.critical = critical
    r.make_request()

    return r.data


def post_grades(section_id, assignment_id, grades, critical=False):
    """grades is list formatted like at apidocs.quickschools.com/#grades"""
    r = Request()
    r.function = requests.post
    r.description = "POST grades"
    r.uri = "/grades"
    r.critical = critical
    r.params = {
        'sectionId': section_id, 'assignmentId': assignment_id,
        'grades': json.dumps(grades)}
    r.make_request()

    return r.data


def post_assignment(section_id, assignment, critical=False):
    """assignment is dict, supports all the fields at apidocs.quickschools.com/#assignment"""
    r = Request()
    r.function = requests.post
    r.description = "POST assignment"
    r.uri = "/sections/{}/assignments".format(section_id)
    r.critical = critical
    r.params = assignment
    r.make_request()

    return r.data


def delete_assignment(section_id, assignment_id, critical=False):
    r = Request()
    r.function = requests.delete
    r.description = "DELETE assignment"
    r.uri = "/sections/{}/assignments/{}".format(section_id, assignment_id)
    r.critical = critical
    r.make_request()
 
    return r.data


def get_error_count():
    return _error_count


def get_schoolcode():
    return api_key[:api_key.find('.')]


def get_base_uri():
    global base_uri

    if not base_uri:
        if use_live_server:
            base_uri = 'https://api.quickschools.com/sms/v1'
        else:
            base_uri = 'https://api.smartschoolcentral.com/sms/v1'
    return base_uri


def get_matching_section(match_section_id, match_teacher=False):
    """
    get the matching section from the current semester based on
    (out of semester) section passed in, so pass in last sem's gym id
    to get the ID of this semester's gym class
    """

    match_section = get_section(match_section_id)

    match_section_name = match_section['sectionName']
    match_class_name = match_section['className']
    match_section_code = match_section['sectionCode']
    match_teacher_id = match_section['teachers'][0]['id']

    new_section_id = ''
    all_sections = get_sections()
    for section in all_sections:
        if (section['sectionName'] == match_section_name
            and section['className'] == match_class_name
            and section['sectionCode'] == match_section_code
            and (not match_teacher
                or (match_teacher
                    and section['teachers'][0]['id'] == match_teacher_id))):
            new_section_id = section['id']

    if new_section_id == '':
        api_logging.info("error: couldn't find match in current semester", [match_section_id, match_teacher_id])
        return None
    else:
        return new_section_id.strip('u').strip("'")


# either cd into school's path or create it; schema: /script/schoolcode
def cd():
    os.chdir(os.path.dirname(os.path.abspath(__file__)))
    school_path = os.path.dirname(os.path.abspath(__file__)) + "/" + get_schoolcode()
    if not os.path.exists(school_path):
        os.makedirs(school_path)
    os.chdir(school_path)


def wait():
    global wait_count

    if use_live_server:
        wait_count += 1
        if not wait_count % 5:
            time.sleep(1)


# ===========
# = Classes =
# ===========


class Request(object):
    """Request class is for making requests via API"""

    def __init__(self):
        self.response = None
        self.uri = ''
        self.params = {}
        self.description = ''
        self.function = requests.get
        self.critical = False
        self.json = None
        self.data = None

    def full_uri(self):
        return get_base_uri() + self.uri

    def full_params(self):
        basic_param = {'apiKey': api_key, 'itemsPerPage': 1000, 'URI': self.uri}
        return dict(basic_param.items() + self.params.items())

    def make_request(self):
        verify_certificate = True if use_live_server else False
        
        self.check_for_ready()

        api_logging.info(self.description, self.full_params(), False, is_request=True)
        self.response = self.function(
            self.full_uri(),
            params=self.full_params(),
            verify=verify_certificate)
        self.json = self.response.json()
        self.process_response()
        self.set_data()
        wait()

    def process_response(self):
        global _error_count

        output = self.description
        if self.success():
            api_logging.info(self.description, self.json, True)
        else:
            if self.critical:
                api_logging.critical(self.description, self.json, True)
            else:
                api_logging.error(self.description, self.json, True)

    def success(self):
        return not 'false' in self.response.text

    def check_for_ready(self):
        if not self.uri or not self.function or not self.description:
            api_logging.critical(
                "\n\n***can't make request - missing variables***\n"
                "must set uri, and function, description to make request")

    def set_data(self):
        if not self.success(): return

        try:
            if type(self.json) is list:
                self.data = self.json
            elif type(self.json) is dict:
                if 'list' in self.json.keys():
                    self.data = self.json['list']
                elif 'students' in self.json.keys():
                    self.data = self.json['students']
                elif 'response' in self.json.keys():
                    self.data = self.json['response']
                elif 'id' in self.json.keys():
                    self.data = self.json
                elif 'success' in self.json.keys():
                    self.data = self.json
                else:
                    raise BadResponseError
            else:

                raise BadResponseError
        except BadResponseError:
            api_logging.critical("Unrecognized response data type", self.json)


class BadResponseError(Exception):
    pass


# init
cd()
get_base_uri()