Select Git revision
Code owners
Assign users and groups as approvers for specific file changes. Learn more.
db.py 3.41 KiB
from server import *
import sqlite3
import re
if config['DB_ENGINE'] == 'sqlite':
created = not os.path.exists(config['SQLITE_DB'])
db = sqlite3.connect(config['SQLITE_DB'])
cur = db.cursor()
if config['SQLITE_INIT_SCHEMA']:
cur.executescript(open(config['DB_SCHEMA']).read())
if config['SQLITE_INIT_DATA'] and created:
cur.executescript(open(config['DB_DATA']).read())
db.commit()
db.close()
# Row wrapper for sqlite
def dict_factory(cursor, row):
d = {}
for idx, col in enumerate(cursor.description):
if type(row[idx]) == str:
d[col[0].split('.')[-1]] = row[idx].replace('\\n','\n').replace('\\r','\r')
else:
d[col[0].split('.')[-1]] = row[idx]
return d
def query(operation, *params):
if config['DB_ENGINE'] == 'mysql':
import mysql.connector
if 'db' not in g or not g.db.is_connected():
g.db = mysql.connector.connect(user=config['MYSQL_USER'], password=config['MYSQL_PASSWD'], host=config['MYSQL_HOST'], database=config['MYSQL_DB'])
if not hasattr(request, 'db'):
request.db = g.db.cursor(dictionary=True)
request.db.execute(operation.replace('?', '%s'), params)
elif config['DB_ENGINE'] == 'sqlite':
if 'db' not in g:
g.db = sqlite3.connect(config['SQLITE_DB'])
g.db.row_factory = dict_factory
g.db.isolation_level = None
if not hasattr(request, 'db'):
request.db = g.db.cursor()
request.db.execute(operation, params)
else:
return []
try:
rows = request.db.fetchall()
except:
rows = []
return rows
@app.teardown_request
def commit_db(*args):
if hasattr(request, 'db'):
request.db.close()
g.db.commit()
def searchquery(text, columns, match, tables, suffix, *suffixparams):
params = []
subexprs = []
words = text.split(' ')
prio = len(words)+1
for word in words:
if word == '' or word.isspace():
continue
matchexpr = ' OR '.join(['%s LIKE ?'%column for column in match])
subexprs.append('SELECT %s, %s AS _prio FROM %s WHERE %s'%(columns, str(prio), tables, matchexpr))
params += ['%'+word+'%']*len(match)
prio -= 1
if subexprs == []:
return []
expr = 'SELECT *,SUM(_prio) AS _score FROM (%s) AS _tmp %s'%(' UNION '.join(subexprs), suffix)
return query(expr, *params, *suffixparams)
LDAP_USERRE = re.compile(r'[^a-z0-9]')
notldap = {
'videoag':('videoag', ['users','videoag'], {'uid': 'videoag', 'givenName': 'Video', 'sn': 'Geier'}),
'gustav':('passwort', ['users'], {'uid': 'gustav', 'givenName': 'Gustav', 'sn': 'Geier'})
}
def ldapauth(user, password):
user = LDAP_USERRE.sub(r'', user.lower())
if 'LDAP_HOST' in config:
import ldap3
try:
conn = ldap3.Connection(config['LDAP_HOST'], 'uid=%s,ou=users,dc=fsmpi,dc=rwth-aachen,dc=de'%user, password, auto_bind=True)
if conn.search("ou=groups,dc=fsmpi,dc=rwth-aachen,dc=de", "(&(cn=*)(memberUid=%s))"%user, attributes=['cn']):
groups = [e.cn.value for e in conn.entries]
conn.unbind()
return user, groups
except ldap3.core.exceptions.LDAPBindError:
pass
elif config.get('DEBUG') and user in notldap and password == notldap[user][0]:
return user, notldap[user][1]
return None, []
def ldapget(user):
user = LDAP_USERRE.sub(r'', user.lower())
if 'LDAP_HOST' in config:
import ldap3
conn = ldap3.Connection('ldaps://rumo.fsmpi.rwth-aachen.de', auto_bind=True)
conn.search("ou=users,dc=fsmpi,dc=rwth-aachen,dc=de", "(uid=%s)"%user,
attributes=ldap3.ALL_ATTRIBUTES)
e = conn.entries[0]
return {'uid': user, 'givenName': e.givenName.value, 'sn':e.sn.value}
else:
return notldap[user][2]