Skip to content
Snippets Groups Projects
Commit 9c9fa2b3 authored by Julian's avatar Julian
Browse files

API getusers/getgroups performance optimization

parent bfd759bd
No related branches found
No related tags found
No related merge requests found
Show whitespace changes
Inline Side-by-side
tests/test_api.py
+ 122
− 0
View file @ 9c9fa2b3
......@@ -109,3 +109,125 @@ class TestAPICheckPassword(UffdTestCase):
r = self.client.post(path=url_for('api.checkpassword'), data={'loginname': 'testuser', 'password': 'wrongpassword'}, headers=[basic_auth('test', 'test')])
self.assertEqual(r.status_code, 200)
self.assertEqual(r.json, None)
class TestAPIGetusers(UffdTestCase):
def setUpDB(self):
db.session.add(APIClient(service=Service(name='test'), auth_username='test', auth_password='test', perm_users=True))
def fix_result(self, result):
result.sort(key=lambda user: user['id'])
for user in result:
user['groups'].sort()
return result
def test_all(self):
r = self.client.get(path=url_for('api.getusers'), headers=[basic_auth('test', 'test')], follow_redirects=True)
self.assertEqual(r.status_code, 200)
self.assertEqual(self.fix_result(r.json), [
{'displayname': 'Test User', 'email': 'test@example.com', 'id': 10000, 'loginname': 'testuser', 'groups': ['uffd_access', 'users']},
{'displayname': 'Test Admin', 'email': 'admin@example.com', 'id': 10001, 'loginname': 'testadmin', 'groups': ['uffd_access', 'uffd_admin', 'users']}
])
def test_id(self):
r = self.client.get(path=url_for('api.getusers', id=10000), headers=[basic_auth('test', 'test')], follow_redirects=True)
self.assertEqual(r.status_code, 200)
self.assertEqual(self.fix_result(r.json), [
{'displayname': 'Test User', 'email': 'test@example.com', 'id': 10000, 'loginname': 'testuser', 'groups': ['uffd_access', 'users']},
])
def test_id_empty(self):
r = self.client.get(path=url_for('api.getusers', id=0), headers=[basic_auth('test', 'test')], follow_redirects=True)
self.assertEqual(r.status_code, 200)
self.assertEqual(r.json, [])
def test_loginname(self):
r = self.client.get(path=url_for('api.getusers', loginname='testuser'), headers=[basic_auth('test', 'test')], follow_redirects=True)
self.assertEqual(r.status_code, 200)
self.assertEqual(self.fix_result(r.json), [
{'displayname': 'Test User', 'email': 'test@example.com', 'id': 10000, 'loginname': 'testuser', 'groups': ['uffd_access', 'users']},
])
def test_loginname_empty(self):
r = self.client.get(path=url_for('api.getusers', loginname='notauser'), headers=[basic_auth('test', 'test')], follow_redirects=True)
self.assertEqual(r.status_code, 200)
self.assertEqual(r.json, [])
def test_email(self):
r = self.client.get(path=url_for('api.getusers', email='admin@example.com'), headers=[basic_auth('test', 'test')], follow_redirects=True)
self.assertEqual(r.status_code, 200)
self.assertEqual(self.fix_result(r.json), [
{'displayname': 'Test Admin', 'email': 'admin@example.com', 'id': 10001, 'loginname': 'testadmin', 'groups': ['uffd_access', 'uffd_admin', 'users']}
])
def test_email_empty(self):
r = self.client.get(path=url_for('api.getusers', email='foo@bar'), headers=[basic_auth('test', 'test')], follow_redirects=True)
self.assertEqual(r.status_code, 200)
self.assertEqual(r.json, [])
def test_group(self):
r = self.client.get(path=url_for('api.getusers', group='uffd_admin'), headers=[basic_auth('test', 'test')], follow_redirects=True)
self.assertEqual(r.status_code, 200)
self.assertEqual(self.fix_result(r.json), [
{'displayname': 'Test Admin', 'email': 'admin@example.com', 'id': 10001, 'loginname': 'testadmin', 'groups': ['uffd_access', 'uffd_admin', 'users']}
])
def test_group_empty(self):
r = self.client.get(path=url_for('api.getusers', group='notagroup'), headers=[basic_auth('test', 'test')], follow_redirects=True)
self.assertEqual(r.status_code, 200)
self.assertEqual(r.json, [])
class TestAPIGetgroups(UffdTestCase):
def setUpDB(self):
db.session.add(APIClient(service=Service(name='test'), auth_username='test', auth_password='test', perm_users=True))
def fix_result(self, result):
result.sort(key=lambda group: group['id'])
for group in result:
group['members'].sort()
return result
def test_all(self):
r = self.client.get(path=url_for('api.getgroups'), headers=[basic_auth('test', 'test')], follow_redirects=True)
self.assertEqual(r.status_code, 200)
self.assertEqual(self.fix_result(r.json), [
{'id': 20001, 'members': ['testadmin', 'testuser'], 'name': 'users'},
{'id': 20002, 'members': ['testadmin', 'testuser'], 'name': 'uffd_access'},
{'id': 20003, 'members': ['testadmin'], 'name': 'uffd_admin'}
])
def test_id(self):
r = self.client.get(path=url_for('api.getgroups', id=20002), headers=[basic_auth('test', 'test')], follow_redirects=True)
self.assertEqual(r.status_code, 200)
self.assertEqual(self.fix_result(r.json), [
{'id': 20002, 'members': ['testadmin', 'testuser'], 'name': 'uffd_access'},
])
def test_id_empty(self):
r = self.client.get(path=url_for('api.getgroups', id=0), headers=[basic_auth('test', 'test')], follow_redirects=True)
self.assertEqual(r.status_code, 200)
self.assertEqual(r.json, [])
def test_name(self):
r = self.client.get(path=url_for('api.getgroups', name='uffd_admin'), headers=[basic_auth('test', 'test')], follow_redirects=True)
self.assertEqual(r.status_code, 200)
self.assertEqual(self.fix_result(r.json), [
{'id': 20003, 'members': ['testadmin'], 'name': 'uffd_admin'}
])
def test_name_empty(self):
r = self.client.get(path=url_for('api.getgroups', name='notagroup'), headers=[basic_auth('test', 'test')], follow_redirects=True)
self.assertEqual(r.status_code, 200)
self.assertEqual(r.json, [])
def test_member(self):
r = self.client.get(path=url_for('api.getgroups', member='testuser'), headers=[basic_auth('test', 'test')], follow_redirects=True)
self.assertEqual(r.status_code, 200)
self.assertEqual(self.fix_result(r.json), [
{'id': 20001, 'members': ['testadmin', 'testuser'], 'name': 'users'},
{'id': 20002, 'members': ['testadmin', 'testuser'], 'name': 'uffd_access'},
])
def test_member_empty(self):
r = self.client.get(path=url_for('api.getgroups', member='notauser'), headers=[basic_auth('test', 'test')], follow_redirects=True)
self.assertEqual(r.status_code, 200)
self.assertEqual(r.json, [])
uffd/api/views.py
+ 37
− 25
View file @ 9c9fa2b3
......@@ -34,8 +34,11 @@ def apikey_required(permission=None):
return wrapper
def generate_group_dict(group):
return {'id': group.unix_gid, 'name': group.name,
'members': [user.loginname for user in group.members]}
return {
'id': group.unix_gid,
'name': group.name,
'members': [user.loginname for user in group.members]
}
@bp.route('/getgroups', methods=['GET', 'POST'])
@apikey_required('users')
......@@ -44,24 +47,30 @@ def getgroups():
abort(400)
key = (list(request.values.keys()) or [None])[0]
values = request.values.getlist(key)
query = Group.query
if key is None:
groups = Group.query.all()
pass
elif key == 'id' and len(values) == 1:
groups = Group.query.filter_by(unix_gid=values[0]).all()
query = query.filter(Group.unix_gid == values[0])
elif key == 'name' and len(values) == 1:
groups = Group.query.filter_by(name=values[0]).all()
query = query.filter(Group.name == values[0])
elif key == 'member' and len(values) == 1:
user = User.query.filter_by(loginname=values[0]).one_or_none()
groups = [] if user is None else user.groups
query = query.join(Group.members).filter(User.loginname == values[0])
else:
abort(400)
return jsonify([generate_group_dict(group) for group in groups])
# Single-result queries perform better without joinedload
if key is None or key == 'member':
query = query.options(db.joinedload(Group.members))
return jsonify([generate_group_dict(group) for group in query])
def generate_user_dict(user, all_groups=None):
if all_groups is None:
all_groups = user.groups
return {'id': user.unix_uid, 'loginname': user.loginname, 'email': user.mail, 'displayname': user.displayname,
'groups': [group.name for group in all_groups if user in group.members]}
def generate_user_dict(user):
return {
'id': user.unix_uid,
'loginname': user.loginname,
'email': user.mail,
'displayname': user.displayname,
'groups': [group.name for group in user.groups]
}
@bp.route('/getusers', methods=['GET', 'POST'])
@apikey_required('users')
......@@ -70,23 +79,23 @@ def getusers():
abort(400)
key = (list(request.values.keys()) or [None])[0]
values = request.values.getlist(key)
query = User.query
if key is None:
users = User.query.all()
pass
elif key == 'id' and len(values) == 1:
users = User.query.filter_by(unix_uid=values[0]).all()
query = query.filter(User.unix_uid == values[0])
elif key == 'loginname' and len(values) == 1:
users = User.query.filter_by(loginname=values[0]).all()
query = query.filter(User.loginname == values[0])
elif key == 'email' and len(values) == 1:
users = User.query.filter_by(mail=values[0]).all()
query = query.filter(User.mail == values[0])
elif key == 'group' and len(values) == 1:
group = Group.query.filter_by(name=values[0]).one_or_none()
users = [] if group is None else group.members
query = query.join(User.groups).filter(Group.name == values[0])
else:
abort(400)
all_groups = None
if len(users) > 1:
all_groups = Group.query.all()
return jsonify([generate_user_dict(user, all_groups) for user in users])
# Single-result queries perform better without joinedload
if key is None or key == 'group':
query = query.options(db.joinedload(User.groups))
return jsonify([generate_user_dict(user) for user in query])
@bp.route('/checkpassword', methods=['POST'])
@apikey_required('checkpassword')
......@@ -108,8 +117,11 @@ def checkpassword():
return jsonify(generate_user_dict(user))
def generate_mail_dict(mail):
return {'name': mail.uid, 'receive_addresses': list(mail.receivers),
'destination_addresses': list(mail.destinations)}
return {
'name': mail.uid,
'receive_addresses': list(mail.receivers),
'destination_addresses': list(mail.destinations)
}
@bp.route('/getmails', methods=['GET', 'POST'])
@apikey_required('mail_aliases')
......
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment

Impressum - Datenschutzerklärung