db.py 1.84 KB
Newer Older
1
2
3
4
5
6
7
8
9
try:
    import mysql.connector
except ImportError:
    pass

try:
    import sqlite3
except ImportError:
    pass
10

11
import aurweb.config
12
13
14
15


class Connection:
    _conn = None
16
    _paramstyle = None
17
18

    def __init__(self):
19
        aur_db_backend = aurweb.config.get('database', 'backend')
20
21

        if aur_db_backend == 'mysql':
22
23
24
25
26
            aur_db_host = aurweb.config.get('database', 'host')
            aur_db_name = aurweb.config.get('database', 'name')
            aur_db_user = aurweb.config.get('database', 'user')
            aur_db_pass = aurweb.config.get('database', 'password')
            aur_db_socket = aurweb.config.get('database', 'socket')
27
28
29
30
31
32
33
34
            self._conn = mysql.connector.connect(host=aur_db_host,
                                                 user=aur_db_user,
                                                 passwd=aur_db_pass,
                                                 db=aur_db_name,
                                                 unix_socket=aur_db_socket,
                                                 buffered=True)
            self._paramstyle = mysql.connector.paramstyle
        elif aur_db_backend == 'sqlite':
35
            aur_db_name = aurweb.config.get('database', 'name')
36
37
38
39
            self._conn = sqlite3.connect(aur_db_name)
            self._paramstyle = sqlite3.paramstyle
        else:
            raise ValueError('unsupported database backend')
40
41

    def execute(self, query, params=()):
42
        if self._paramstyle in ('format', 'pyformat'):
43
44
45
46
47
            query = query.replace('%', '%%').replace('?', '%s')
        elif self._paramstyle == 'qmark':
            pass
        else:
            raise ValueError('unsupported paramstyle')
48
49
50
51
52
53
54
55
56
57
58

        cur = self._conn.cursor()
        cur.execute(query, params)

        return cur

    def commit(self):
        self._conn.commit()

    def close(self):
        self._conn.close()