__init__.py 2.21 KB
Newer Older
1
2
# -*- coding: utf-8 -*-

3
4
5
6
import os
import logging
import vispa
import sqlalchemy
7
from sqlalchemy.ext.declarative import declarative_base
8

9
10
logger = logging.getLogger(__name__)

Gero Müller's avatar
Gero Müller committed
11
12
Base = declarative_base()

13
14
__all__ = ['group', 'preference', 'project', 'role', 'shortcuts', 'user',
           'workgroup', 'workspace']
15

16
17
18
19
20
FORBIDDEN_PHRASES = [u"drop ", u"select ", u"dump ", u"insert ", u"delete ",
                     u"update ", u"drop\\ ", u"select\\ ", u"dump\\ ",
                     u"insert\\ ", u"delete\\ ", u"update\\ "]
FORBIDDEN_CHARS = [u"´", u"`"]

Gero Müller's avatar
Gero Müller committed
21

22

Gero Müller's avatar
Gero Müller committed
23
24
25
26
27
28
29
def insertion_safe(*args, **kwargs):
    for arg in list(args) + kwargs.values():
        if isinstance(arg, dict):
            arg = arg.keys()
        if not isinstance(arg, list):
            arg = [arg]
        for elem in arg:
30
31
            if not isinstance(elem, (str, unicode)):
                continue
Gero Müller's avatar
Gero Müller committed
32
33
34
35
36
37
38
39
            # 1. check: forbidden phrases
            for phrase in FORBIDDEN_PHRASES:
                if elem.lower().find(phrase) >= 0:
                    return False, elem
            # 2. check: forbidden chars
            for char in FORBIDDEN_CHARS:
                if elem.lower().find(char) >= 0:
                    return False, elem
40
    return True, None
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69

def open_database(var_dir=None):
    var_dir = var_dir or ""
    sa_identifier = vispa.config('database', 'sqlalchemy.url',
                                 'sqlite:///%s' % os.path.join(var_dir, "vispa.db"))
    pool_size = vispa.config('database', 'sqlalchemy.pool_size', 10)
    max_overflow = vispa.config(
        'database',
        'sqlalchemy.max_overflow',
        10)
    # https://github.com/mitsuhiko/flask-sqlalchemy/issues/2
    # http://docs.sqlalchemy.org/en/latest/core/pooling.html#dealing-with-disconnects
    pool_recycle = vispa.config(
        'database',
        'sqlalchemy.pool_recycle',
        7200)
    logger.info('Use database %s.' % sa_identifier)
    try:
        engine = sqlalchemy.create_engine(
            sa_identifier,
            echo=False,
            pool_size=pool_size,
            pool_recycle=pool_recycle,
            max_overflow=max_overflow)
    except TypeError:
        engine = sqlalchemy.create_engine(
            sa_identifier,
            echo=False)
    return engine