Commit 483cffed authored by jens.kleineheismann's avatar jens.kleineheismann

Initial commit.

parents
*.pyc
# build package
*.egg-info
dist/
build/
# .hgignore
.*\.pyc
^build/
^dist/
^.*\.egg-info/
include README.rst
ABOUT
=====
This package provide a python binding to the KIT Gruppenverwaltung REST API.
INSTALLATION
============
$ pip install -e .
or
$ python setup.py develop
or any other method, you use to install your python packages.
USAGE
=====
from kitgvapi import basicConfig, Oe, Group, User
basicConfig('myusername', 'mypassword')
iism = Oe('IISM')
for g in iism.get_groups():
print g.name
staff = Group('IISM-EM-OU-Staff')
for u in staff.get_users(effective=True):
print "%s %s (%s)" % (u.firstname, u.lastname, u.samaccount)
for g in staff.get_groups():
print g.name
me = User('zy8373')
others = me.oe.get_users()
LICENCE
=======
Author: Jens Kleineheismann <kleineheismann@kit.edu>
Permission to use, copy, modify, and/or distribute this software
for any purpose with or without fee is hereby granted.
from .client import Client
from .config import basicConfig
from .models import Group, Oe, User
import json
import requests
from .config import CONFIG
from .exceptions import NotFoundError, MissingConfigError
class ClientConfig(object):
def __init__(self):
self.url_schema = CONFIG['URL_SCHEMA']
self.server = CONFIG['SERVER']
self.base_uri = CONFIG['BASE_URI']
self.username = CONFIG['USERNAME']
self.password = CONFIG['PASSWORD']
if not self.username:
self.credentials = None
elif isinstance(self.username, MissingConfigError):
raise self.username
elif isinstance(self.password, MissingConfigError):
raise self.password
else:
self.credentials = requests.auth.HTTPBasicAuth(self.username, self.password)
class Client(object):
def __init__(self, config=None):
self.config = config
if self.config is None:
self.config = ClientConfig()
def _get(self, url):
r = requests.get(url, verify=True, auth=self.config.credentials)
if r.status_code != requests.codes.ok:
r.raise_for_status()
return r
def _post(self, url, data=None):
if data is None:
data = {}
r = requests.post(url, verify=True, auth=self.config.credentials, data=data)
if r.status_code != requests.codes.ok:
r.raise_for_status()
return r
def read(self, path):
url = self.config.url_schema + self.config.server + self.config.base_uri + path
try:
r = self._get(url)
except requests.exceptions.HTTPError as e:
if e.response.status_code == 404:
raise NotFoundError("Endpoint returned Not Found")
else:
raise e
j = r.json()
return j
def write(self, path, data=None):
url = self.config.url_schema + self.config.server + self.config.base_uri + path
if data is not None:
data = json.dumps(data)
r = self._post(url, data)
def get_oes(self, pk_only=False):
path = '/oe/list'
l = self.read(path)
if pk_only:
l = [d[CONFIG['OE_PK_ATTRIB']] for d in l]
return l
def get_oe(self, pk):
path = '/oe/name/{pk}'.format(pk=pk)
return self.read(path)
def get_oe_groups(self, pk, pk_only=False):
path = '/groups/{pk}'.format(pk=pk)
l = self.read(path)
if pk_only:
l = [d[CONFIG['GROUP_PK_ATTRIB']] for d in l]
return l
def get_oe_users(self, pk, pk_only=False):
path = '/oe/name/{pk}/identities'.format(pk=pk)
l = self.read(path)
if pk_only:
l = [d[CONFIG['USER_PK_ATTRIB']] for d in l]
return l
def get_user(self, pk):
path = '/identities/get/{pk}'.format(pk=pk)
return self.read(path)
def get_user_groups(self, pk, pk_only=False):
path = '/identities/get/{pk}/memberof'.format(pk=pk)
l = self.read(path)
if pk_only:
l = [d[CONFIG['GROUP_PK_ATTRIB']] for d in l]
return l
def get_group(self, pk):
group_pk = pk
oe_pk = pk.partition('-')[0]
path = '/groups/{oe}/{group}'.format(oe=oe_pk, group=group_pk)
return self.read(path)
def get_group_users(self, pk, pk_only=True, effective=False):
if not pk_only:
raise NotImplemented('Not supported')
group_pk = pk
oe_pk = pk.partition('-')[0]
quant = 'members'
if effective:
quant = 'effective' + quant
path = '/groups/{oe}/{group}/{quant}'.format(oe=oe_pk, group=group_pk, quant=quant)
return self.read(path)
def get_group_groups(self, pk, pk_only=True):
if not pk_only:
raise NotImplemented('Not supported')
group_pk = pk
oe_pk = pk.partition('-')[0]
path = '/groups/{oe}/{group}/subgroups'.format(oe=oe_pk, group=group_pk)
return self.read(path)
def add_group_group(self, super_pk, sub_pk):
oe_pk = super_pk.partition('-')[0]
path = '/groups/{oe}/{superg}/subgroups/{subg}'.format(oe=oe_pk, superg=super_pk, subg=sub_pk)
self.write(path)
def add_group_user(self, group_pk, user_pk):
oe_pk = group_pk.partition('-')[0]
path = '/groups/{oe}/{group}/members/{user}'.format(oe=oe_pk, group=group_pk, user=user_pk)
self.write(path)
def create_group(self, name, description):
oe_pk = name.partition('-')[0]
path = '/groups/{oe}'.format(oe=oe_pk)
data = {'name': name, 'description': description}
self.write(path, data=data)
from .exceptions import MissingConfigError
CONFIG = {
'GROUP_PK_ATTRIB': 'name',
'OE_PK_ATTRIB': 'kurz',
'USER_PK_ATTRIB': 'samaccount',
'URL_SCHEMA': 'https://',
'SERVER': 'kit-idm-03.scc.kit.edu',
'BASE_URI': '/itbportal-rest-war/rest',
'USERNAME': MissingConfigError('No username set'),
'PASSWORD': MissingConfigError('No password set'),
}
def basicConfig(username, password):
CONFIG['USERNAME'] = username
CONFIG['PASSWORD'] = password
class KitGvApiError(Exception):
pass
class MissingConfigError(KitGvApiError):
pass
class MissingClientError(KitGvApiError):
pass
class MissingPkError(KitGvApiError):
pass
class NotFoundError(KitGvApiError):
pass
import logging
logging.basicConfig(format='%(levelname)s: %(message)s')
logger = logging.getLogger('kitgvapi')
from .client import Client
from .config import CONFIG
from .exceptions import MissingClientError, MissingPkError
class Model(object):
def __init__(self, pk=None, client=None, data=None):
self.pk = pk
self.client = client
self.is_loaded = False
if self.client is None:
self.client = Client()
if data is not None:
self._load_data_attribs(data)
self.is_loaded = True
if self.pk is not None:
self.load()
def _load_data_attribs(self, data=None):
raise NotImplementedError('Abstract method')
def load(self, reload=False):
if self.is_loaded and not reload:
return
if self.client is None:
raise MissingClientError()
if self.pk is None:
raise MissingPkError()
self._load_data_attribs()
self.is_loaded = True
def _save_data_attribs(self):
raise NotImplementedError('Abstract method')
def save(self):
if self.client is None:
raise MissingClientError()
self._save_data_attribs()
class Group(Model):
def _load_data_attribs(self, data=None):
if data is None:
data = self.client.get_group(pk=self.pk)
self.name = data['name']
self.description = data['description']
self.gidNumber = data['gidNumber']
self.readOnly = data['readOnly']
self.verteiler = data['verteiler']
self.eMailAddress = data['eMailAddress']
self.oe_pk = data['oe']
self.pk = getattr(self, CONFIG['GROUP_PK_ATTRIB'])
def _save_data_attribs(self):
raise NotImplemented('Not supported')
@property
def oe(self):
return Oe(client=self.client, pk=self.oe_pk)
def get_users(self, effective=False):
self.load()
l = self.client.get_group_users(pk=self.pk, effective=effective)
for i in l:
y = User(client=self.client, pk=i)
yield y
def add_user(self, pk):
self.client.add_group_user(group_pk=self.pk, user_pk=pk)
def remove_user(self, pk):
raise NotImplemented('Not yet')
def get_groups(self):
self.load()
l = self.client.get_group_groups(pk=self.pk)
for i in l:
y = Group(client=self.client, pk=i)
yield y
def add_group(self, pk):
self.client.add_group_group(super_pk=self.pk, sub_pk=pk)
def remove_group(self, pk):
raise NotImplemented('Not yet')
class Oe(Model):
def _load_data_attribs(self, data=None):
if data is None:
data = self.client.get_oe(pk=self.pk)
self.kurz = data['kurz']
self.lang = data['lang']
self.pk = getattr(self, CONFIG['OE_PK_ATTRIB'])
def _save_data_attribs(self):
raise NotImplemented('Not supported')
def get_groups(self):
self.load()
l = self.client.get_oe_groups(pk=self.pk)
for d in l:
y = Group(client=self.client, data=d)
yield y
def get_users(self):
self.load()
l = self.client.get_oe_users(pk=self.pk)
for d in l:
y = User(client=self.client, data=d)
yield y
def create_group(self, name, description):
self.client.create_group(name, description)
def delete_group(self, pk):
raise NotImplemented('Not yet')
class User(Model):
def _load_data_attribs(self, data=None):
if data is None:
data = self.client.get_user(pk=self.pk)
self.samaccount = data['samaccount']
self.firstname = data['firstname']
self.lastname = data['lastname']
self.oe = Oe(client=self.client, data=data['oe'])
self.pk = getattr(self, CONFIG['USER_PK_ATTRIB'])
def _save_data_attribs(self):
raise NotImplemented('Not supported')
def get_groups(self):
self.load()
l = self.client.get_user_groups(pk=self.pk)
for d in l:
y = Group(client=self.client, data=d)
yield y
#!/usr/bin/env python
from setuptools import setup, find_packages
def readme():
with open('README.rst') as f:
return f.read()
setup(
name='python-kitgvapi',
version='0.1.dev0',
description=('Python bindings for KIT Gruppenverwaltung REST API'),
long_description=readme(),
author='Jens Kleineheismann',
author_email='kleineheismann@kit.edu',
url='https://em-dub.iism.kit.edu/vcs/hg/python-kitgvapi',
packages=find_packages(),
include_package_data=True,
install_requires=[
'requests',
],
)
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment