Add files via upload

__init__.py

@@ -0,0 +1,135 @@
+print(" ██╗   ██╗██████╗  ██████╗")
+print(" ██║   ██║██╔══██╗██╔════╝")
+print(" ██║   ██║██████╔╝██║     ")
+print(" ██║   ██║██╔══██╗██║     ")
+print(" ╚██████╔╝██████╔╝╚██████╗")
+print("  ╚═════╝ ╚═════╝  ╚═════╝")
+
+from CanvasAPI import callapi
+from CanvasAPI.util import file, zipfile, getpass, printer
+import sys, os, ntpath
+import configparser
+from os.path import isabs
+import logging
+
+
+logging.basicConfig(filename='canvasapi.log',level=logging.INFO, format='%(asctime)s:%(levelname)s:%(message)s', datefmt='%Y-%m-%d %H:%M:%S')
+logging.getLogger().addHandler(logging.StreamHandler())
+
+parser = configparser.ConfigParser()
+parser.read("{}/config-canvas.def".format(os.path.dirname(__file__)))
+
+_CONNECT_AUTOMATICALLY = parser['Base']['ConnectAutomatically']
+_DEFAULT_INSTANCE = int(parser['Base']['DefaultInstance'])
+_DEFAULT_TOKEN_LOCATION = parser['Base']['TokenLocation']
+if not isabs(_DEFAULT_TOKEN_LOCATION) or not _DEFAULT_TOKEN_LOCATION:
+    _DEFAULT_TOKEN_LOCATION = os.getcwd() + "/" + _DEFAULT_TOKEN_LOCATION
+
+_DEFAULT_TOKEN_FILENAMES = []
+_DEFAULT_SERVERS = []
+_ENVIRONMENTS = []
+_CONNECTED = False
+
+for section in parser.sections():
+    if section != "Base":
+        _ENVIRONMENTS.append(str(section))
+        _DEFAULT_TOKEN_FILENAMES.append(parser[section]['TokenFile'])
+        _DEFAULT_SERVERS.append(parser[section]['Server'])
+
+if len(_ENVIRONMENTS) <= 0:
+    print("No environments found")
+    input("Press enter to close...")
+    sys.exit(1)
+
+
+if _CONNECT_AUTOMATICALLY:
+    if _DEFAULT_INSTANCE <= len(_ENVIRONMENTS) and _DEFAULT_INSTANCE >= 1:
+        _INSTANCE = _DEFAULT_INSTANCE
+    else:
+        printer.print_list(_ENVIRONMENTS, numbered=True)
+        try:
+            i = int(input("Select instance: "))
+            if i <= len(_ENVIRONMENTS) and i >= 1:
+                _INSTANCE = i
+            else:
+                print("Bad input")
+                input("Press enter to close...")
+                sys.exit(1)
+        except ValueError as e:
+            print("Bad input")
+            input("Press enter to close...")
+            sys.exit(1)
+    _SERVER = _DEFAULT_SERVERS[_INSTANCE-1]   
+    if zipfile.is_zipfile(_DEFAULT_TOKEN_LOCATION):
+        zf = zipfile.ZipFile(_DEFAULT_TOKEN_LOCATION)
+        for zinfo in zf.infolist():
+            is_encrypted = zinfo.flag_bits & 0x1
+        if is_encrypted:
+            print("{} is password protected".format(_DEFAULT_TOKEN_LOCATION))
+            tries = 0
+            while True:
+                try:
+                    _TOKEN = zf.read(_DEFAULT_TOKEN_FILENAMES[_INSTANCE-1], pwd=bytes(getpass.getpass(),'utf-8')).decode()
+                    break
+                except RuntimeError as e:
+                    tries += 1
+                    print(e)
+                    if tries >= 3:
+                        _TOKEN = ""
+                        break
+                    continue
+
+        else:
+            _TOKEN = zf.read(_DEFAULT_TOKEN_FILENAMES[_INSTANCE-1]).decode().strip()
+    else:
+        try:
+            f = open("{}/{}".format(_DEFAULT_TOKEN_LOCATION, _DEFAULT_TOKEN_FILENAMES[_INSTANCE-1]))
+            _TOKEN = f.readline().strip()
+            f.close()
+        except FileNotFoundError as e:            
+            print(e)
+            _TOKEN = input("Input token: ")
+    instance = callapi.Instance(_SERVER, _TOKEN.rstrip())
+    _CONNECTED = True
+else:
+    instance = callapi.Instance()
+
+from CanvasAPI import accounts
+from CanvasAPI import assignments
+from CanvasAPI import courses
+from CanvasAPI import custom_gradebook_columns
+from CanvasAPI import enrollments
+from CanvasAPI import external_tools
+from CanvasAPI import favorites
+from CanvasAPI import feature_flags
+from CanvasAPI import files
+from CanvasAPI import groups
+from CanvasAPI import grade_change_log
+from CanvasAPI import navigation
+from CanvasAPI import roles
+from CanvasAPI import sections
+from CanvasAPI import users
+from CanvasAPI import quizzes
+
+__all__ = ['instance',
+           'accounts',
+           'assignments',
+           'courses',
+           'custom_gradebook_columns',
+           'enrollments',
+           'external_tools',
+           'favorites',
+           'feature_flags',
+           'files',
+           'grade_change_log',
+           'groups',
+           'navigation',
+           'quizzes',
+           'roles',
+           'sections',
+           'users']
+
+if _CONNECTED:
+    _SELF = users.self()
+    if _SELF is None:
+        logging.error("Invalid token. Failed to connect to '{}'".format(_SERVER))

accounts/__init__.py

@@ -0,0 +1,10 @@
+from CanvasAPI.util import callhelper
+from CanvasAPI.accounts import sub_accounts
+from CanvasAPI import instance
+
+__all__ = ["get", "sub_accounts"]
+
+def get(account_id):
+    '''Gets a course'''
+    url_str = "accounts/{}".format(account_id)
+    return instance.call_api(url_str)

accounts/sub_accounts.py

@@ -0,0 +1,9 @@
+from CanvasAPI.util import callhelper
+from CanvasAPI import instance
+
+__all__ = ["get"]
+
+def get(account_id, recursive=False):
+    '''Gets a course'''
+    url_str = "accounts/{}/sub_accounts?recursive={}".format(account_id, recursive)
+    return instance.all_pages(url_str)

assignments/__init__.py

@@ -0,0 +1,9 @@
+from CanvasAPI.util import callhelper
+from CanvasAPI import instance
+
+__all__ = ["get_all"]
+
+def get_all(course_id):
+    '''list_assignments'''
+    url_str = "/courses/{}/assignments".format(course_id)
+    return instance.call_api(url_str)

callapi.py

@@ -0,0 +1,147 @@
+from CanvasAPI.util import callhelper
+from urllib.request import Request, urlopen
+from urllib.parse import urlencode
+import urllib
+import collections
+import logging
+
+
+class Instance:
+    @property
+    def server_url(self):
+        return self._server_url
+    @server_url.setter
+    def server_url(self, value):
+        self._server_url = value
+    @property
+    def token(self):
+        return self._token
+    @token.setter
+    def token(self, value):
+        self._token = value
+    @property
+    def version(self):
+        return self._version
+    @version.setter
+    def version(self, value):
+        self._version = value
+
+    @property
+    def initialized(self):
+        ##Check whether _server_url and _token are None or empty
+        return not (not self._server_url or not self._token)
+
+    def __init__(self, server_url="", token="", version="v1"):
+        self.connect(server_url, token, version)
+
+    def connect(self, server_url, token, version="v1"):
+        self._server_url = server_url
+        self._token = token
+        self._version = version
+
+    def call_api(self, url, method="GET", is_URL_absolute=False, post_fields=None):
+        response = self._call_api_raw(url, method=method, is_URL_absolute=is_URL_absolute, post_fields=post_fields)
+        if response is not None:
+            return callhelper.get_response_body(response)
+        else:
+            return None
+
+
+    def _call_api_raw(self, url, method="GET", is_URL_absolute=False, post_fields=None):
+        '''Makes an API call, returning up to 10 results.
+
+        Keyword arguments:
+        url - url to call (example "courses")
+        method - the API method (default "GET")
+               - optionally, "GET, POST, PUT, DELETE"
+        is_URL_absolute - whether the given URL is absolute, or
+                          uses the given server (default False)
+        post_fields - set of terms to include in a post call (default None)
+        '''
+
+        if not self.initialized:
+            raise ValueError("Instance must be initialized before making a call")
+
+        server_url = self.server_url
+        version = self.version
+        token = self._token
+
+        if is_URL_absolute:
+            urlstr = url
+        else:
+            urlstr = 'https://{}/api/{}/{}'.format(server_url, version, url)
+
+        request = Request(urlstr)
+        request.add_header('Authorization', 'Bearer {}'.format(token))
+        request.method = method
+        if post_fields is not None:
+            request.data = urlencode(post_fields).encode()
+
+
+        try:
+            r = urlopen(request) 
+        except urllib.error.HTTPError as e:
+            if post_fields is not None:
+                logging.warning("HTTP Error {} '{}': urlstr='{}' method='{}' post_fields='{}'".format(e.code, e.reason, urlstr, method, ", ".join(["{}:{}".format(key, value) for key, value in post_fields.items()])))
+            else:
+                logging.warning("HTTP Error {} '{}': urlstr='{}' method='{}'".format(e.code, e.reason, urlstr, method))       
+            r = None
+        return r
+
+
+    def _pages(self, url, method="GET", is_URL_absolute=False, post_fields=None):
+        '''Gets an iteratable return of all results of an API call
+
+        Keyword arguments:
+        url - url to call (example "courses")
+        method - the API method (default "GET")
+               - optionally, "GET, POST, PUT, DELETE"
+        is_URL_absolute - whether the given URL is absolute, or
+                          uses the given server (default False)
+        post_fields - set of terms to include in a post call (default None)
+        '''
+        response = self._call_api_raw(url, method, is_URL_absolute, post_fields)
+        yield response
+        more_pages = True
+        while more_pages:
+            link_header = None
+            for header in response.headers.items():
+                if header[0] == 'Link':
+                    link_header = header
+                    break
+            else:
+                more_pages = False
+                raise StopIteration()
+            links = link_header[1].split(',')
+            for link in links:
+                parts = link.split(';')
+                if parts[1].find('next') >= 0:
+                    next_page = parts[0]
+                    next_page = next_page.replace('<', '')
+                    next_page = next_page.replace('>', '')
+                    next_page = next_page.strip()
+                    response = self._call_api_raw(url=next_page, method=method, is_URL_absolute=True, post_fields=None)
+                    yield response
+                    break
+            else:
+                more_pages = False
+                raise StopIteration()
+
+    def all_pages(self, url, method="GET", is_URL_absolute=False, post_fields=None):
+        '''Makes an API call, returning all results.
+
+        Keyword arguments:
+        url - url to call (example "courses")
+        method - the API method (default "GET")
+               - optionally, "GET, POST, PUT, DELETE"
+        is_URL_absolute - whether the given URL is absolute, or
+                          uses the given server (default False)
+        post_fields - set of terms to include in a post call (default None)
+        '''
+        collector = []
+        for response in self._pages(url, method, is_URL_absolute, post_fields):
+            response_body = callhelper.get_response_body(response)
+            if isinstance(response_body, collections.OrderedDict):
+                response_body = [response_body]
+            collector = collector + response_body
+        return collector

config-canvas.def

@@ -0,0 +1,16 @@
+[Base]
+DefaultInstance = 1
+TokenLocation = 
+ConnectAutomatically = True
+
+[Prod]
+TokenFile = prod.txt
+Server = canvas.ubc.ca
+
+[Test]
+TokenFile = test.txt
+Server = ubc.test.instructure.com
+
+[Beta]
+TokenFile = beta.txt
+Server = ubc.beta.instructure.com