ws: update web-service API to latest specifications
This commit is contained in:
parent
b5f33cf68d
commit
5484011502
|
|
@ -1,121 +1 @@
|
|||
import requests
|
||||
import os.path
|
||||
import xml.etree.ElementTree as etree
|
||||
import collections
|
||||
import logging
|
||||
|
||||
CourseByOwnerResponse = collections.namedtuple('CourseByOwnerResponse',
|
||||
('message', 'courses'))
|
||||
Course = collections.namedtuple('Course',
|
||||
('id', 'name', 'available'))
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
def bool2str(b):
|
||||
return 'YES' if b else 'NO'
|
||||
|
||||
class BlackboardConnector(object):
|
||||
def __init__(self, url, shared_key):
|
||||
self.url = url
|
||||
self.shared_key = shared_key
|
||||
|
||||
def _get_course_helper(self, r):
|
||||
r.encoding = 'utf-8'
|
||||
if r.status_code == 200:
|
||||
x = etree.XML(r.text.encode('utf-8'))
|
||||
info_message_elt = x.find('infoMessage')
|
||||
course_elts = x.findall('*/course')
|
||||
courses = []
|
||||
for course_elt in course_elts:
|
||||
id_elt = course_elt.find('ID')
|
||||
name_elt = course_elt.find('name')
|
||||
available_elt = course_elt.find('available')
|
||||
if None in (id_elt, name_elt, available_elt) or available_elt.text not in ('YES', 'NO'):
|
||||
logger.error('invalid answer to GetCourseByOwner: %r', r.text)
|
||||
return False, 'Erreur BlackBoard'
|
||||
course = Course(id_elt.text.strip(), name_elt.text.strip(),
|
||||
available_elt.text == 'YES')
|
||||
courses.append(course)
|
||||
return True, CourseByOwnerResponse(info_message_elt.text.strip(),
|
||||
courses)
|
||||
elif r.status_code == 400:
|
||||
return False, r.text
|
||||
else:
|
||||
logging.error('BlackBoard code HTTP %s', r.status_code)
|
||||
return False, 'Erreur BlackBoard'
|
||||
|
||||
def get_course_by_owner(self, user_login):
|
||||
assert user_login
|
||||
url = os.path.join(self.url, 'GetCourseByOwner')
|
||||
logging.debug('sending request %s', url)
|
||||
r = requests.get(url, params={'user_login':
|
||||
user_login.encode('ascii')})
|
||||
return self._get_course_helper(r)
|
||||
|
||||
def get_course_by_ue(self, entity_code):
|
||||
assert entity_code
|
||||
url = os.path.join(self.url, 'GetCourseByUE')
|
||||
logging.debug('sending request %s', url)
|
||||
r = requests.get(url,
|
||||
params={'UE_ID': entity_code.encode('ascii')})
|
||||
return self._get_course_helper(r)
|
||||
|
||||
def send_file(self, uploadfile, courses_id):
|
||||
assert hasattr(uploadfile, 'read')
|
||||
assert hasattr(courses_id, '__iter__')
|
||||
assert len(courses_id) > 0
|
||||
assert all(map(bool, courses_id))
|
||||
url = os.path.join(self.url, 'SendFile')
|
||||
r = requests.post(url, params={'course_ID': courses_id},
|
||||
files={'uploadfile': uploadfile})
|
||||
if r.status_code == 200:
|
||||
return True, r.text
|
||||
return False, r.text
|
||||
|
||||
def create_course(self, course_id, course_name, category,
|
||||
open_to_visitors=True, auto_inscription=True, password=None):
|
||||
assert isinstance(course_id, unicode) and course_id
|
||||
assert isinstance(course_name, unicode) and course_name
|
||||
assert isinstance(category, unicode) and unicode
|
||||
|
||||
open_to_visitors = bool2str(open_to_visitors)
|
||||
auto_inscription = bool2str(auto_inscription)
|
||||
|
||||
params = {
|
||||
'course_ID': course_id,
|
||||
'course_name': course_name,
|
||||
'open_to_visitors': open_to_visitors,
|
||||
'auto_inscription': auto_inscription,
|
||||
'category': category,
|
||||
}
|
||||
if password:
|
||||
params['password'] = password
|
||||
url = os.path.join(self.url, 'CreateCourse')
|
||||
r = requests.post(url, params=params)
|
||||
if r.status_code == 200:
|
||||
return True, r.text
|
||||
return False, r.text
|
||||
|
||||
def get_categories(self):
|
||||
return True, []
|
||||
|
||||
if __name__ == '__main__':
|
||||
import sys
|
||||
import logging
|
||||
logging.basicConfig(level=logging.DEBUG)
|
||||
try:
|
||||
connector = BlackboardConnector(sys.argv[1])
|
||||
if len(sys.argv) == 4 and sys.argv[2] == 'GetCourseByOwner':
|
||||
print connector.get_course_by_owner(sys.argv[3])
|
||||
elif len(sys.argv) == 4 and sys.argv[2] == 'GetCourseByUE':
|
||||
print connector.get_course_by_ue(sys.argv[3])
|
||||
elif len(sys.argv) == 5 and sys.argv[2] == 'SendFile':
|
||||
print connector.send_file(file(sys.argv[3]), sys.argv[4].split(','))
|
||||
else:
|
||||
raise ValueError
|
||||
except:
|
||||
print 'Syntax: python', sys.argv[0], '<url> [GetCourseByOwner <user_login>|GetCourseByUE <UE_ID>|SendFile <filename> <courseid1,courseid2,courseid3,etc.>]'
|
||||
|
||||
|
||||
|
||||
|
||||
__VERSION__ = (0, 1, 0)
|
||||
|
|
|
|||
|
|
@ -0,0 +1,175 @@
|
|||
import requests
|
||||
import os.path
|
||||
import xml.etree.ElementTree as etree
|
||||
import collections
|
||||
import logging
|
||||
|
||||
CourseByOwnerResponse = collections.namedtuple('CourseByOwnerResponse',
|
||||
('message', 'courses'))
|
||||
Course = collections.namedtuple('Course',
|
||||
('id', 'name', 'available'))
|
||||
|
||||
GetCategoryResponse = collections.namedtuple('GetCategoryResponse',
|
||||
('message', 'categories'))
|
||||
|
||||
Category = collections.namedtuple('Category',
|
||||
('id', 'name'))
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
def bool2str(b):
|
||||
return 'YES' if b else 'NO'
|
||||
|
||||
class BlackboardConnector(object):
|
||||
def __init__(self, url, login=None, password=None, shared_key=None):
|
||||
self.url = url
|
||||
self.login = login
|
||||
self.password = password
|
||||
self.shared_key = shared_key
|
||||
self.session = requests.session()
|
||||
url = os.path.join(self.url, 'login/')
|
||||
print url
|
||||
self.session.post(url,
|
||||
params=dict(user_id=self.login,
|
||||
password=self.password, action='login'))
|
||||
# print login
|
||||
# print r.status_code
|
||||
# print self.session.__dict__
|
||||
|
||||
def _get_course_helper(self, r):
|
||||
r.encoding = 'utf-8'
|
||||
if r.status_code == 200:
|
||||
try:
|
||||
x = etree.XML(r.content)
|
||||
except etree.ParseError:
|
||||
print 'Parse Error', r.content, r.status_code
|
||||
raise
|
||||
info_message_elt = x.find('infoMessage')
|
||||
course_elts = x.findall('*/course')
|
||||
courses = []
|
||||
for course_elt in course_elts:
|
||||
id_elt = course_elt.find('ID')
|
||||
name_elt = course_elt.find('name')
|
||||
available_elt = course_elt.find('available')
|
||||
if None in (id_elt, name_elt, available_elt) or available_elt.text not in ('YES', 'NO'):
|
||||
logger.error('invalid answer to GetCourseByOwner: %r', r.text)
|
||||
return False, 'Erreur BlackBoard'
|
||||
course = Course(id_elt.text.strip(), name_elt.text.strip(),
|
||||
available_elt.text == 'YES')
|
||||
courses.append(course)
|
||||
return True, CourseByOwnerResponse(info_message_elt.text.strip(),
|
||||
courses)
|
||||
elif r.status_code == 400:
|
||||
return False, r.text
|
||||
else:
|
||||
logging.error('BlackBoard code HTTP %s', r.status_code)
|
||||
return False, 'Erreur BlackBoard'
|
||||
|
||||
def get_course_by_owner(self, user_login):
|
||||
r = self.send_request('GetCoursesByOwner',
|
||||
user_login=user_login.encode('ascii'))
|
||||
return self._get_course_helper(r)
|
||||
|
||||
def get_course_by_ue(self, entity_code):
|
||||
r = self.send_request('GetCoursesByOwner',
|
||||
UE_ID=entity_code.encode('ascii'))
|
||||
return self._get_course_helper(r)
|
||||
|
||||
def send_file(self, uploadfile, courses_id):
|
||||
assert hasattr(uploadfile, 'read')
|
||||
assert hasattr(courses_id, '__iter__')
|
||||
assert len(courses_id) > 0
|
||||
assert all(map(bool, courses_id))
|
||||
r = self.send_request('SendFile', course_ID=courses_id,
|
||||
files=dict(uploadfile=uploadfile), method='post')
|
||||
if r.status_code == 200:
|
||||
return True, r.text
|
||||
return False, r.text
|
||||
|
||||
def create_course(self, course_id, course_name, category,
|
||||
open_to_visitors=True, auto_inscription=True, password=None):
|
||||
assert isinstance(course_id, unicode) and course_id
|
||||
assert isinstance(course_name, unicode) and course_name
|
||||
assert isinstance(category, unicode) and unicode
|
||||
|
||||
open_to_visitors = bool2str(open_to_visitors)
|
||||
auto_inscription = bool2str(auto_inscription)
|
||||
|
||||
params = {
|
||||
'course_ID': course_id,
|
||||
'course_name': course_name,
|
||||
'open_to_visitors': open_to_visitors,
|
||||
'auto_inscription': auto_inscription,
|
||||
'category': category,
|
||||
}
|
||||
if password:
|
||||
params['password'] = password
|
||||
r = self.send_request('CreateCourse', **params)
|
||||
if r.status_code == 200:
|
||||
return True, r.text
|
||||
return False, r.text
|
||||
|
||||
def send_request(self, request_name, **params):
|
||||
params['secret_key'] = self.shared_key
|
||||
kwargs = { 'params': params }
|
||||
files = params.pop('files', None)
|
||||
method = params.pop('method', 'get')
|
||||
if files:
|
||||
kwargs['files'] = files
|
||||
url = os.path.join(self.url, 'ui-ui-Polynum-BBLEARN/app', request_name)
|
||||
logging.debug('sending request %s', url)
|
||||
r = getattr(self.session, method)(url, **kwargs)
|
||||
print 'content', r.content
|
||||
r.encoding = 'utf-8'
|
||||
if r.status_code == 500:
|
||||
print r.text
|
||||
return r
|
||||
|
||||
def get_categories(self):
|
||||
r = self.send_request('GetCategories')
|
||||
if r.status_code == 200:
|
||||
x = etree.XML(r.content)
|
||||
info_message_elt = x.find('infoMessage')
|
||||
category_elts = x.findall('categories/category')
|
||||
categories = []
|
||||
for category_elt in category_elts:
|
||||
id_elt = category_elt.find('ID')
|
||||
name_elt = category_elt.find('name')
|
||||
category = Category(id_elt.text.strip(),
|
||||
name_elt.text.strip())
|
||||
categories.append(category)
|
||||
return True, GetCategoryResponse(info_message_elt.text.strip(),
|
||||
categories)
|
||||
elif r.status_code == 400:
|
||||
return False, r.text
|
||||
else:
|
||||
print r.text
|
||||
logging.error('BlackBoard code HTTP %s', r.status_code)
|
||||
return False, 'Erreur BlackBoard {0}'.format(r.status_code)
|
||||
|
||||
if __name__ == '__main__':
|
||||
import sys
|
||||
import logging
|
||||
import pprint
|
||||
|
||||
logging.basicConfig(level=logging.DEBUG)
|
||||
try:
|
||||
connector = BlackboardConnector(sys.argv[1], *sys.argv[2:5])
|
||||
if len(sys.argv) == 7 and sys.argv[5] == 'GetCoursesByOwner':
|
||||
ok, response = pprint.pprint(connector.get_course_by_owner(sys.argv[6])[1])
|
||||
elif len(sys.argv) == 7 and sys.argv[5] == 'GetCoursesByUE':
|
||||
ok, response = connector.get_course_by_ue(sys.argv[6])
|
||||
elif len(sys.argv) == 8 and sys.argv[5] == 'SendFile':
|
||||
ok, text = connector.send_file(file(sys.argv[6]), sys.argv[7].split(','))
|
||||
elif len(sys.argv) == 6 and sys.argv[5] == 'GetCategories':
|
||||
ok, response = connector.get_categories()
|
||||
else:
|
||||
raise ValueError
|
||||
if ok:
|
||||
print 'OK'
|
||||
else:
|
||||
print 'NOK'
|
||||
print response
|
||||
except Exception, e:
|
||||
raise
|
||||
print 'Syntax: python', sys.argv[0], '<url> [GetCoursesByOwner <user_login>|GetCoursesByUE <UE_ID>|SendFile <filename> <courseid1,courseid2,courseid3,etc.>]'
|
||||
Reference in New Issue