Page MenuHomePhorge
Differential D3779 Diff 10769 pykolab/wap_client/__init__.py

Changeset View

Standalone View

View Options

pykolab/wap_client/__init__.py

Context not available.
kolab_wap_url = conf.get('kolab_wap', 'api_url') kolab_wap_url = conf.get('kolab_wap', 'api_url')
if not kolab_wap_url == None: if kolab_wap_url is not None:
result = urlparse(kolab_wap_url) result = urlparse(kolab_wap_url)
else: else:
result = None result = None
Context not available.
conn = None conn = None
def authenticate(username=None, password=None, domain=None): def authenticate(username=None, password=None, domain=None):
global session_id global session_id
if username == None: if username is None:
username = conf.get('ldap', 'bind_dn') username = conf.get('ldap', 'bind_dn')
if password == None: if password is None:
password = conf.get('ldap', 'bind_pw') password = conf.get('ldap', 'bind_pw')
if domain == None: if domain is None:
domain = conf.get('kolab', 'primary_domain') domain = conf.get('kolab', 'primary_domain')
post = json.dumps( post = json.dumps(
Context not available.
session_id = response['session_token'] session_id = response['session_token']
return True return True
def connect(uri=None): def connect(uri=None):
global conn, API_SSL, API_PORT, API_HOSTNAME, API_BASE global conn, API_SSL, API_PORT, API_HOSTNAME, API_BASE
if not uri == None: if uri is not None:
result = urlparse(uri) result = urlparse(uri)
if hasattr(result, 'scheme') and result.scheme == 'https': if hasattr(result, 'scheme') and result.scheme == 'https':
Context not available.
if hasattr(result, 'path'): if hasattr(result, 'path'):
API_BASE = result.path API_BASE = result.path
if conn == None: if conn is None:
if API_SSL: if API_SSL:
conn = httplib.HTTPSConnection(API_HOSTNAME, API_PORT) conn = httplib.HTTPSConnection(API_HOSTNAME, API_PORT)
else: else:
Context not available.
return conn return conn
def disconnect(quit=False): def disconnect(quit=False):
global conn, session_id global conn, session_id
Context not available.
conn.close() conn.close()
conn = None conn = None
def domain_add(domain, aliases=[]): def domain_add(domain, aliases=[]):
dna = conf.get('ldap', 'domain_name_attribute') dna = conf.get('ldap', 'domain_name_attribute')
post = json.dumps({ post = json.dumps({
dna: [ domain ] + aliases dna: [domain] + aliases
}) })
return request('POST', 'domain.add', post=post) return request('POST', 'domain.add', post=post)
def domain_delete(domain, force=False): def domain_delete(domain, force=False):
domain_id, domain_attrs = domain_find(domain).popitem() domain_id, domain_attrs = domain_find(domain).popitem()
Context not available.
return request('POST', 'domain.delete', post=post) return request('POST', 'domain.delete', post=post)
def domain_find(domain): def domain_find(domain):
dna = conf.get('ldap', 'domain_name_attribute') dna = conf.get('ldap', 'domain_name_attribute')
get = { dna: domain } get = {dna: domain}
return request('GET', 'domain.find', get=get) return request('GET', 'domain.find', get=get)
def domain_info(domain): def domain_info(domain):
domain_id, domain_attrs = domain_find(domain) domain_id, domain_attrs = domain_find(domain)
get = { 'id': domain_id } get = {'id': domain_id}
return request('GET', 'domain.info', get=get) return request('GET', 'domain.info', get=get)
def domains_capabilities(): def domains_capabilities():
return request('GET', 'domains.capabilities') return request('GET', 'domains.capabilities')
def domains_list(): def domains_list():
return request('GET', 'domains.list') return request('GET', 'domains.list')
def form_value_generate(params): def form_value_generate(params):
post = json.dumps(params) post = json.dumps(params)
return request('POST', 'form_value.generate', post=post) return request('POST', 'form_value.generate', post=post)
def form_value_generate_password(*args, **kw): def form_value_generate_password(*args, **kw):
return request('GET', 'form_value.generate_password') return request('GET', 'form_value.generate_password')
def form_value_list_options(object_type, object_type_id, attribute): def form_value_list_options(object_type, object_type_id, attribute):
post = json.dumps( post = json.dumps(
{ {
Context not available.
return request('POST', 'form_value.list_options', post=post) return request('POST', 'form_value.list_options', post=post)
def form_value_select_options(object_type, object_type_id, attribute): def form_value_select_options(object_type, object_type_id, attribute):
post = json.dumps( post = json.dumps(
{ {
'object_type': object_type, 'object_type': object_type,
'type_id': object_type_id, 'type_id': object_type_id,
'attributes': [ attribute ] 'attributes': [attribute]
} }
) )
return request('POST', 'form_value.select_options', post=post) return request('POST', 'form_value.select_options', post=post)
def get_group_input(): def get_group_input():
group_types = group_types_list() group_types = group_types_list()
if len(group_types) > 1: if len(group_types) > 1:
for key in group_types: for key in group_types:
if not key == "status": if not key == "status":
print("%s) %s" % (key,group_types[key]['name'])) print("%s) %s" % (key, group_types[key]['name']))
group_type_id = utils.ask_question("Please select the group type") group_type_id = utils.ask_question("Please select the group type")
Context not available.
params[attribute] = utils.ask_question(attribute) params[attribute] = utils.ask_question(attribute)
for attribute in group_type_info['auto_form_fields']: for attribute in group_type_info['auto_form_fields']:
retval = eval("group_form_value_generate_%s(params)" % (attribute)) retval = eval("group_form_value_generate_%s(params)" % attribute)
params[attribute] = retval[attribute] params[attribute] = retval[attribute]
return params return params
def get_user_input(): def get_user_input():
user_types = user_types_list() user_types = user_types_list()
Context not available.
print("") print("")
for key in user_types['list']: for key in user_types['list']:
if not key == "status": if not key == "status":
print("%s) %s" % (key,user_types['list'][key]['name'])) print("%s) %s" % (key, user_types['list'][key]['name']))
print("") print("")
user_type_id = utils.ask_question("Please select the user type") user_type_id = utils.ask_question("Please select the user type")
Context not available.
for attribute in user_type_info['form_fields']: for attribute in user_type_info['form_fields']:
if isinstance(user_type_info['form_fields'][attribute], dict): if isinstance(user_type_info['form_fields'][attribute], dict):
if 'optional' in user_type_info['form_fields'][attribute] and user_type_info['form_fields'][attribute]['optional']: if not (not ('optional' in user_type_info['form_fields'][attribute]) or not
user_type_info['form_fields'][attribute]['optional']):
may_attrs.append(attribute) may_attrs.append(attribute)
else: else:
must_attrs.append(attribute) must_attrs.append(attribute)
Context not available.
default = attribute_values[attribute]['default'] default = attribute_values[attribute]['default']
params[attribute] = utils.ask_menu( params[attribute] = utils.ask_menu(
"Choose the %s value" % (attribute), "Choose the %s value" % attribute,
attribute_values[attribute]['list'], attribute_values[attribute]['list'],
default=default default=default
) )
Context not available.
default = user_type_info['form_fields'][attribute]['default'] default = user_type_info['form_fields'][attribute]['default']
params[attribute] = utils.ask_menu( params[attribute] = utils.ask_menu(
"Choose the %s value" % (attribute), "Choose the %s value" % attribute,
user_type_info['form_fields'][attribute]['values'], user_type_info['form_fields'][attribute]['values'],
default=default default=default
) )
Context not available.
return params return params
def group_add(params=None): def group_add(params=None):
if params == None: if params is None:
params = get_group_input() params = get_group_input()
post = json.dumps(params) post = json.dumps(params)
return request('POST', 'group.add', post=post) return request('POST', 'group.add', post=post)
def group_delete(params=None): def group_delete(params=None):
if params == None: if params is None:
params = { params = {
'id': utils.ask_question("Name of group to delete", "group") 'id': utils.ask_question("Name of group to delete", "group")
} }
Context not available.
return request('POST', 'group.delete', post=post) return request('POST', 'group.delete', post=post)
def group_form_value_generate_mail(params=None): def group_form_value_generate_mail(params=None):
if params == None: if params is None:
params = get_user_input() params = get_user_input()
params = json.dumps(params) params = json.dumps(params)
return request('POST', 'group_form_value.generate_mail', params) return request('POST', 'group_form_value.generate_mail', params)
def group_find(params=None): def group_find(params=None):
post = { 'search': { 'params': {} } } post = {'search': {'params': {}}}
for (k,v) in params.items(): for (k, v) in params.items():
post['search']['params'][k] = { 'value': v, 'type': 'exact' } post['search']['params'][k] = {'value': v, 'type': 'exact'}
return request('POST', 'group.find', post=json.dumps(post)) return request('POST', 'group.find', post=json.dumps(post))
def group_info(group=None): def group_info(group=None):
if group == None: if group is None:
group = utils.ask_question("group DN") group = utils.ask_question("group DN")
return request('GET', 'group.info', get={ 'id': group }) return request('GET', 'group.info', get={'id': group})
def group_members_list(group=None): def group_members_list(group=None):
if group == None: if group is None:
group = utils.ask_question("Group email address") group = utils.ask_question("Group email address")
group = request('GET', 'group.members_list?group=%s' % (group)) group = request('GET', 'group.members_list?group=%s' % group)
return group return group
def group_types_list(): def group_types_list():
return request('GET', 'group_types.list') return request('GET', 'group_types.list')
def groups_list(params={}): def groups_list(params={}):
return request('POST', 'groups.list', post=json.dumps(params)) return request('POST', 'groups.list', post=json.dumps(params))
def ou_add(params={}): def ou_add(params={}):
return request('POST', 'ou.add', post=json.dumps(params)) return request('POST', 'ou.add', post=json.dumps(params))
def ou_delete(params={}): def ou_delete(params={}):
return request('POST', 'ou.delete', post=json.dumps(params)) return request('POST', 'ou.delete', post=json.dumps(params))
def ou_edit(params={}): def ou_edit(params={}):
return request('POST', 'ou.edit', post=json.dumps(params)) return request('POST', 'ou.edit', post=json.dumps(params))
def ou_find(params=None): def ou_find(params=None):
post = { 'search': { 'params': {} } } post = {'search': {'params': {}}}
for (k,v) in params.items(): for (k, v) in params.items():
post['search']['params'][k] = { 'value': v, 'type': 'exact' } post['search']['params'][k] = {'value': v, 'type': 'exact'}
return request('POST', 'ou.find', post=json.dumps(post)) return request('POST', 'ou.find', post=json.dumps(post))
def ou_info(ou): def ou_info(ou):
_params = { 'id': ou } _params = {'id': ou}
ou = request('GET', 'ou.info', get=_params) ou = request('GET', 'ou.info', get=_params)
return ou return ou
def ous_list(params={}): def ous_list(params={}):
return request('POST', 'ous.list', post=json.dumps(params)) return request('POST', 'ous.list', post=json.dumps(params))
def request(method, api_uri, get=None, post=None, headers={}): def request(method, api_uri, get=None, post=None, headers={}):
response_data = request_raw(method, api_uri, get, post, headers) response_data = request_raw(method, api_uri, get, post, headers)
Context not available.
print("%s: %s (code %s)" % (response_data['status'], response_data['reason'], response_data['code'])) print("%s: %s (code %s)" % (response_data['status'], response_data['reason'], response_data['code']))
return False return False
def request_raw(method, api_uri, get=None, post=None, headers={}, isretry=False): def request_raw(method, api_uri, get=None, post=None, headers={}, isretry=False):
global session_id global session_id
if not session_id == None: if session_id is not None:
headers["X-Session-Token"] = session_id headers["X-Session-Token"] = session_id
reconnect = False reconnect = False
Context not available.
if conf.debuglevel > 8: if conf.debuglevel > 8:
conn.set_debuglevel(9) conn.set_debuglevel(9)
if not get == None: if get is not None:
_get = "?%s" % (urlencode(get)) _get = "?%s" % (urlencode(get))
else: else:
_get = "" _get = ""
log.debug(_("Requesting %r with params %r") % ("%s/%s" % (API_BASE,api_uri), (get, post)), level=8) log.debug(_("Requesting %r with params %r") % ("%s/%s" % (API_BASE, api_uri), (get, post)), level=8)
try: try:
conn.request(method.upper(), "%s/%s%s" % (API_BASE, api_uri, _get), post, headers) conn.request(method.upper(), "%s/%s%s" % (API_BASE, api_uri, _get), post, headers)
Context not available.
response = conn.getresponse() response = conn.getresponse()
data = response.read() data = response.read()
log.debug(_("Got response: %r") % (data), level=8) log.debug(_("Got response: %r") % data, level=8)
except (httplib.BadStatusLine, httplib.CannotSendRequest) as e: except (httplib.BadStatusLine, httplib.CannotSendRequest) as e:
if isretry: if isretry:
Context not available.
return response_data return response_data
def resource_add(params=None)Index: pykolab/wap_client/__init__.py def resource_add(params=None)Index: pykolab/wap_client/__init__.py
if params == None: if params is None:
params = get_user_input() params = get_user_input()
return request('POST', 'resource.add', post=json.dumps(params)) return request('POST', 'resource.add', post=json.dumps(params))
def resource_delete(params=None): def resource_delete(params=None):
if params == None: if params is None:
params = { params = {
'id': utils.ask_question("Resource DN to delete", "resource") 'id': utils.ask_question("Resource DN to delete", "resource")
} }
return request('POST', 'resource.delete', post=json.dumps(params)) return request('POST', 'resource.delete', post=json.dumps(params))
def resource_find(params=None): def resource_find(params=None):
post = { 'search': { 'params': {} } } post = {'search': {'params': {}}}
for (k,v) in params.items(): for (k, v) in params.items():
post['search']['params'][k] = { 'value': v, 'type': 'exact' } post['search']['params'][k] = {'value': v, 'type': 'exact'}
return request('POST', 'resource.find', post=json.dumps(post)) return request('POST', 'resource.find', post=json.dumps(post))
def resource_info(resource=None): def resource_info(resource=None):
if resource == None: if resource is None:
resource = utils.ask_question("Resource DN") resource = utils.ask_question("Resource DN")
return request('GET', 'resource.info', get={ 'id': resource }) return request('GET', 'resource.info', get={'id': resource})
def resource_types_list(): def resource_types_list():
return request('GET', 'resource_types.list') return request('GET', 'resource_types.list')
def resources_list(params={}): def resources_list(params={}):
return request('POST', 'resources.list', post=json.dumps(params)) return request('POST', 'resources.list', post=json.dumps(params))
def role_add(params=None): def role_add(params=None):
if params == None: if params is None:
role_name = utils.ask_question("Role name") role_name = utils.ask_question("Role name")
params = { params = {
Context not available.
return request('POST', 'role.add', params) return request('POST', 'role.add', params)
def role_capabilities(): def role_capabilities():
return request('GET', 'role.capabilities') return request('GET', 'role.capabilities')
def role_delete(params=None): def role_delete(params=None):
if params == None: if params is None:
role_name = utils.ask_question("Role name") role_name = utils.ask_question("Role name")
role = role_find_by_attribute({'cn': role_name}) role = role_find_by_attribute({'cn': role_name})
params = { params = {
Context not available.
return request('POST', 'role.delete', post=post) return request('POST', 'role.delete', post=post)
def role_find_by_attribute(params=None): def role_find_by_attribute(params=None):
if params == None: if params is None:
role_name = utils.ask_question("Role name") role_name = utils.ask_question("Role name")
else: else:
role_name = params['cn'] role_name = params['cn']
get = { 'cn': role_name } get = {'cn': role_name}
role = request('GET', 'role.find_by_attribute', get=get) role = request('GET', 'role.find_by_attribute', get=get)
return role return role
def role_info(role_name): def role_info(role_name):
role = role_find_by_attribute({'cn': role_name}) role = role_find_by_attribute({'cn': role_name})
get = { 'role': role['id'] } get = {'role': role['id']}
role = request('GET', 'role.info', get=get) role = request('GET', 'role.info', get=get)
return role return role
def roles_list(): def roles_list():
return request('GET', 'roles.list') return request('GET', 'roles.list')
def sharedfolder_add(params=None): def sharedfolder_add(params=None):
if params == None: if params is None:
params = get_user_input() params = get_user_input()
return request('POST', 'sharedfolder.add', post=json.dumps(params)) return request('POST', 'sharedfolder.add', post=json.dumps(params))
def sharedfolder_delete(params=None): def sharedfolder_delete(params=None):
if params == None: if params is None:
params = { params = {
'id': utils.ask_question("Shared Folder DN to delete", "sharedfolder") 'id': utils.ask_question("Shared Folder DN to delete", "sharedfolder")
} }
return request('POST', 'sharedfolder.delete', post=json.dumps(params)) return request('POST', 'sharedfolder.delete', post=json.dumps(params))
def sharedfolders_list(params={}): def sharedfolders_list(params={}):
return request('POST', 'sharedfolders.list', post=json.dumps(params)) return request('POST', 'sharedfolders.list', post=json.dumps(params))
def system_capabilities(domain=None): def system_capabilities(domain=None):
return request('GET', 'system.capabilities', get={'domain':domain}) return request('GET', 'system.capabilities', get={'domain': domain})
def system_get_domain(): def system_get_domain():
return request('GET', 'system.get_domain') return request('GET', 'system.get_domain')
def system_select_domain(domain=None): def system_select_domain(domain=None):
if domain == None: if domain is None:
domain = utils.ask_question("Domain name") domain = utils.ask_question("Domain name")
get = { 'domain': domain } get = {'domain': domain}
return request('GET', 'system.select_domain', get=get) return request('GET', 'system.select_domain', get=get)
def user_add(params=None): def user_add(params=None):
if params == None: if params is None:
params = get_user_input() params = get_user_input()
params = json.dumps(params) params = json.dumps(params)
return request('POST', 'user.add', post=params) return request('POST', 'user.add', post=params)
def user_delete(params=None): def user_delete(params=None):
if params == None: if params is None:
params = { params = {
'id': utils.ask_question("Username for user to delete", "user") 'id': utils.ask_question("Username for user to delete", "user")
} }
Context not available.
return request('POST', 'user.delete', post=post) return request('POST', 'user.delete', post=post)
def user_edit(user = None, attributes={}):
if user == None: def user_edit(user=None, attributes={}):
if user is None:
get = { get = {
'id': utils.ask_question("Username for user to edit", "user") 'id': utils.ask_question("Username for user to edit", "user")
} }
Context not available.
return user_edit return user_edit
def user_find(attribs=None): def user_find(attribs=None):
if attribs == None: if attribs is None:
post = { post = {
'search': { 'search': {
'params': { 'params': {
utils.ask_question("Attribute") : { utils.ask_question("Attribute"): {
'value': utils.ask_question("value"), 'value': utils.ask_question("value"),
'type': 'exact' 'type': 'exact'
} }
Context not available.
} }
} }
else: else:
post = { 'search': { 'params': {} } } post = {'search': {'params': {}}}
for (k,v) in attribs.items(): for (k, v) in attribs.items():
post['search']['params'][k] = { 'value': v, 'type': 'exact' } post['search']['params'][k] = {'value': v, 'type': 'exact'}
post = json.dumps(post) post = json.dumps(post)
Context not available.
return user return user
def user_form_value_generate(params=None): def user_form_value_generate(params=None):
if params == None: if params is None:
params = get_user_input() params = get_user_input()
post = json.dumps(params) post = json.dumps(params)
return request('POST', 'form_value.generate', post=post) return request('POST', 'form_value.generate', post=post)
def user_form_value_generate_uid(params=None): def user_form_value_generate_uid(params=None):
if params == None: if params is None:
params = get_user_input() params = get_user_input()
params = json.dumps(params) params = json.dumps(params)
return request('POST', 'form_value.generate_uid', params) return request('POST', 'form_value.generate_uid', params)
def user_form_value_generate_userpassword(*args, **kw): def user_form_value_generate_userpassword(*args, **kw):
result = form_value_generate_password() result = form_value_generate_password()
return { 'userpassword': result['password'] } return {'userpassword': result['password']}
def user_info(user=None): def user_info(user=None):
if user == None: if user is None:
user = utils.ask_question("User email address") user = utils.ask_question("User email address")
_params = { 'id': user } _params = {'id': user}
user = request('GET', 'user.info', get=_params) user = request('GET', 'user.info', get=_params)
return user return user
def users_list(params={}): def users_list(params={}):
return request('POST', 'users.list', post=json.dumps(params)) return request('POST', 'users.list', post=json.dumps(params))
def user_types_list(): def user_types_list():
return request('GET', 'user_types.list') return request('GET', 'user_types.list')
Context not available.