Skip to content
Snippets Groups Projects
Select Git revision
  • 3bfa3abee41afd4e870cddac5fd37ecbba63dac2
  • master default protected
  • postgres_integration
  • s3compatible
  • intros
  • bootstrap4
  • modules
7 results

server.py

Blame
  • 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]