aurweb/db.py

   1 import mysql.connector
   2 import sqlite3
   3
   4 import aurweb.config
   5
   6
   7 class Connection:
   8     _conn = None
   9     _paramstyle = None
  10
  11     def __init__(self):
  12         aur_db_backend = aurweb.config.get('database', 'backend')
  13
  14         if aur_db_backend == 'mysql':
  15             aur_db_host = aurweb.config.get('database', 'host')
  16             aur_db_name = aurweb.config.get('database', 'name')
  17             aur_db_user = aurweb.config.get('database', 'user')
  18             aur_db_pass = aurweb.config.get('database', 'password')
  19             aur_db_socket = aurweb.config.get('database', 'socket')
  20             self._conn = mysql.connector.connect(host=aur_db_host,
  21                                                  user=aur_db_user,
  22                                                  passwd=aur_db_pass,
  23                                                  db=aur_db_name,
  24                                                  unix_socket=aur_db_socket,
  25                                                  buffered=True)
  26             self._paramstyle = mysql.connector.paramstyle
  27         elif aur_db_backend == 'sqlite':
  28             aur_db_name = aurweb.config.get('database', 'name')
  29             self._conn = sqlite3.connect(aur_db_name)
  30             self._paramstyle = sqlite3.paramstyle
  31         else:
  32             raise ValueError('unsupported database backend')
  33
  34     def execute(self, query, params=()):
  35         if self._paramstyle in ('format', 'pyformat'):
  36             query = query.replace('%', '%%').replace('?', '%s')
  37         elif self._paramstyle == 'qmark':
  38             pass
  39         else:
  40             raise ValueError('unsupported paramstyle')
  41
  42         cur = self._conn.cursor()
  43         cur.execute(query, params)
  44
  45         return cur
  46
  47     def commit(self):
  48         self._conn.commit()
  49
  50     def close(self):
  51         self._conn.close()