PK!`OJ55 bbrest.pyimport maya import requests import re import types import asyncio import aiohttp import urllib class BbRest: session = '' expiration_epoch = '' version = '' functions = {} def __init__(self, key, secret, url, headers=None): #these variables are accessible in the class, but not externally. self.__key = key self.__secret = secret self.__url = url self.__headers = headers #Authenticate the session session = requests.Session() payload = {'grant_type':'client_credentials'} if self.__headers: session.headers.update(self.__headers) #Sends a post to get the authentication token r = session.post(f"{self.__url}/learn/api/public/v1/oauth2/token", data=payload, auth=(self.__key, self.__secret)) #Adds the token to the headers for future requests. if r.status_code == 200: token = r.json()["access_token"] session.headers.update({"Authorization":f"Bearer {token}"}) self.expiration_epoch = maya.now() + r.json()["expires_in"] else: print('Authorization failed, check your key, secret and url') return #set the session within the class self.session = session #get the current version via a REST call r = self.session.get(f'{url}/learn/api/public/v1/system/version') if r.status_code == 200: major = r.json()['learn']['major'] minor = r.json()['learn']['minor'] version = f"{major}.{minor}.0" #Ignore incremental patches else: print(f"Could not retrieve version, error code: {r.status_code}") version = '3000.0.0' #Version wasn't supported until 3000.3.0 #This helps only pull down APIs your version has access to. self.version = version #use the functions that exist in functions.p, #or retrieve from the swagger_json definitions swagger_json = requests.get(f'https://developer.blackboard.com/portal/docs/apis/learn-swagger-{version}.json').json() p = r'\d+.\d+.\d+' functions = [] for path in swagger_json['paths']: for call in swagger_json['paths'][path].keys(): meta = swagger_json['paths'][path][call] functions.append( {'summary':meta['summary'].replace(' ',''), 'description':meta['description'], 'parameters':meta['parameters'], 'method':call, 'path':path, #'version':re.findall(p,meta['description']) }) #store all functions in a class visible list self.__all_functions = functions self.supported_functions() self.method_generator() def is_supported(self, function): if not function['version']: return False start = function['version'][0] if len(function['version']) == 1: return start <= self.version else: end = function['version'][1] return start <= self.version < end def supported_functions(self): """ This method generates all API methods for the BB class, using some python magic. There are three primary benefits against generating a file: 1. Obfuscation of paths - there's no risk of publishing url information. 2. Auto Updates - choses correct version of the path based on internal version. 3. Self contained class - no need to pass the BbSession elsewhere. """ #filter out unsupported rest calls, based on current version #functions = [f for f in self.__all_functions if self.is_supported(f)] functions = [f for f in self.__all_functions] #generate a dictionary of supported methods d_functions = {} for function in functions: summary = function['summary'] description = function['description'] parameters = function['parameters'] method = function['method'] path = function['path'] #Work around for 4 methods with similar names. if summary in ['GetChildren','GetMemberships']: if summary == 'GetChildren' and 'contentId' in path: summary = 'GetContentChildren' elif summary == 'GetChildren' and 'courseId' in path: summary = 'GetCourseChildren' elif summary == 'GetMemberships' and 'userId' in path: summary = 'GetUserMemberships' elif summary == 'GetMemberships' and 'courseId' in path: summary = 'GetCourseMemberships' d_functions[summary] = {'method':method, 'path':path, 'description':description, 'parameters':parameters} self.functions = d_functions def method_generator(self): #Go through each supported method, and figure out parameters, #Then create a function on the fly, and save this function as a class method. #This is complex, and probably not pythonic, but the results are hard to argue with. functions = self.functions for function in functions: path = functions[function]['path'] description = functions[function]['description'] parameters = functions[function]['parameters'] p = r'{\w+}' def_params = ['self']+[param[1:-1]+'= None' for param in re.findall(p,path)] params = [param[1:-1]+'= '+param[1:-1] for param in re.findall(p,path)] #put, post, patch methods have payload as an argument #get has params as an argument if functions[function]['method'][0] == 'p': def_params.append('payload= {}') params.append('payload= payload') if functions[function]['method'] == 'get': def_params.append('params= {}') params.append('params= params') if function[-1] == 's': def_params.append('limit= 100') params.append('limit= limit') def_params.append('sync= True') params.append('sync= sync') def_param_string = ', '.join(def_params) param_string = ', '.join(params) exec(f"""def {function}({def_param_string}): return self.call('{function}', **clean_kwargs({param_string}))""") exec(f"""{function}.__doc__ = '''{description}\nParameters:\n{parameters}\n '''""") exec(f"""self.{function} = types.MethodType({function},self)""") #One way to get async methods is to generate them all #I opted to use a keyword argument instead, asynch, #to reduce the number of methods. #exec(f"""async def {function}Async({def_param_string}): return await self.acall('{function}', **clean_kwargs({param_string}))""") #exec(f"""{function}Async.__doc__ = '''{description}\nParameters:\n{parameters}\n '''""") #exec(f"""self.{function}Async = types.MethodType({function}Async,self)""") async def acall(self, summary, **kwargs): if self.is_expired(): self.refresh_token() method = self.functions[summary]['method'] path = self.__url + self.functions[summary]['path'] url = path.format(**kwargs) params = kwargs.get('params', {}) payload = kwargs.get('payload', {}) limit = kwargs.get('limit', 100) if limit == 100: async with aiohttp.ClientSession(headers=self.session.headers) as session: #print(url, params) async with session.request(method, url=url, json=payload, params=params) as resp: return await resp.json() tasks = [] print(f'Limit is {limit}') for i in range(0,limit,100): new_params = params.copy() new_params['limit'] = 100 new_params['offset'] = i tasks.append(self.acall(summary, params=new_params)) resps = await asyncio.gather(*tasks) results = [] #print(f'There are {len(resps)} responses') for resp in resps: if 'results' in resp: #print(f"There are {len(resp['results'])} results in this response") results.extend(resp['results']) #print(len(results)) if len(results) > limit: resp = {'results':results[:limit]} params['offset'] = limit resp['paging'] = {'nextPage': f'{url}?{urllib.parse.urlencode(params)}'} else: resp = {'results':results} return resp def call(self, summary, **kwargs): r''' Constructs and sends a :class:`Request `. :param summary: method for the new `Request` . :param params: (optional) Dictionary, list of tuples or bytes to send in the body of the :class:`Request`. :param payload: (optional) A JSON serializable Python object to send in the body of the :class:`Request`. :return: :class:`Response ` object :rtype: requests.Response ''' if kwargs.get('sync','') != True: return self.acall(summary, **kwargs) method = self.functions[summary]['method'] path = self.__url + self.functions[summary]['path'] url = path.format(**kwargs) params = kwargs.get('params', {}) payload = kwargs.get('payload', {}) limit = kwargs.get('limit', 100) if self.is_expired(): self.refresh_token() req = requests.Request(method=method, url=url, params = params, json = payload) prepped = self.session.prepare_request(req) #delete doesn't return json... for some reason if method == 'delete': return self.session.send(prepped) resp = self.session.send(prepped).json() cur_resp = resp if 'results' in cur_resp: while 'paging' in cur_resp and len(resp['results']) < limit: next_page = self.__url + cur_resp['paging']['nextPage'] req = requests.Request(method=method, url=next_page) prepped = self.session.prepare_request(req) cur_resp = self.session.send(prepped).json() if 'results' in cur_resp: resp['results'].extend(cur_resp['results']) if len(resp['results']) > limit: resp['results'] = resp['results'][:limit] vals = resp['paging']['nextPage'].split('=') vals[-1] = str(limit) resp['paging']['nextPage'] = '='.join(vals) return resp def is_expired(self): return maya.now() > self.expiration_epoch def refresh_token(self): payload = {'grant_type':'client_credentials'} r = self.session.post(f"{self.__url}/learn/api/public/v1/oauth2/token", data=payload, auth=(self.__key, self.__secret)) if r.status_code == 200: token = r.json()["access_token"] self.session.headers.update({"Authorization":f"Bearer {token}"}) self.expiration_epoch = maya.now() + r.json()["expires_in"] def expiration(self): return(self.expiration_epoch.slang_time()) def calls_remaining(self): r = self.GetUser(userId='dne') if 'X-Rate-Limit-Remaining' not in r.headers: print('Rate limits not in the headers for your version') return calls_limit = int(r.headers['X-Rate-Limit-Limit']) calls_remaining = int(r.headers['X-Rate-Limit-Remaining']) reset_seconds = int(r.headers['X-Rate-Limit-Reset']) calls_perc = (100 * calls_remaining / calls_limit) reset_time = maya.now() + reset_seconds used_calls = calls_limit - calls_remaining #weird fomatting issue with f-strings, didn't want to display tabs. call_str = f"""You've used {used_calls} REST calls so far.\nYou have {calls_perc:.2f}% left until {reset_time.slang_time()}\nAfter that, they should reset""" print(call_str) def clean_kwargs(courseId=None, userId=None, columnId=None, groupId=None, **kwargs): if userId: if userId[0] != '_' and ':' not in userId: kwargs['userId'] = f'userName:{userId}' else: kwargs['userId'] = userId if courseId: if courseId[0] != '_' and ':' not in courseId: kwargs['courseId'] = f'courseId:{courseId}' else: kwargs['courseId'] = courseId if columnId: if columnId[0] != '_' and columnId != 'finalGrade': kwargs['columnId'] = f'externalId:{columnId}' else: kwargs['columnId'] = columnId if groupId: if groupId[0] != '_': kwargs['groupId'] = f'externalId:{groupId}' else: kwargs['groupId'] = groupId return kwargs def clean_params(parameters): return parametersPK!00bbrest-0.3.2.dist-info/LICENSEMIT License Copyright (c) 2018 Matthew Deakyne Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. PK!HnHTUbbrest-0.3.2.dist-info/WHEEL A н#Z;/"d&F[xzw@Zpy3Fv]\fi4WZ^EgM_-]#0(q7PK!H3Bbbrest-0.3.2.dist-info/METADATAXob 8)8Ѳ!J+iĒ]Kn $k·߽of;Jv8#;+BKBݗQ~]mv4-zյ:B-WVې=w-wzCo/|eҔdykBǨZ^SMPm^ϩ7-VЏgO, kV^+RUj=o=g&ą1v_zx!{FlPPn)tש뛊m0fؘX$m B>̂^3:\&<&,W4x'ٹ#x cVߟ?gueǾk\<΋ S}=S)W[׳Ũ k@ pH V@g GyEv=sQRr O)6i% I@rR^riahO]ӷVL~9' a)SOCg2S84x؟4kQS0<`q܈3dl1[릧٧l6NWKm}gs~>OgX~z0)pptpuEo8e!~ZY&tjvGONg@cV5<|2H& (31#K JLİk& 䆩q-pݓDMj01Bu3d\*{gn{s=ǻS8}2.OzD2R1 e|8Cx9 < wT+fVdA @e|߈ƕR2X `LKr&tܠX>h\O3itAr;k㜤C馫 PK!HNebbrest-0.3.2.dist-info/RECORDurC@{e¡-lArɔ?E,o3nc{@6D/a!v-G1 s+M#ܐ`-l*/X.X4Yv0Ud{h~F5)^I%ͨx⥷G]c.tW`9(kn/aN@Ubv?Mg IOUY\*-C p^Kʥ,ʉxt P={Tʌ#uSEj~PK!`OJ55 bbrest.pyPK!005bbrest-0.3.2.dist-info/LICENSEPK!HnHTUW:bbrest-0.3.2.dist-info/WHEELPK!H3B:bbrest-0.3.2.dist-info/METADATAPK!HNeCbbrest-0.3.2.dist-info/RECORDPKe%E