Package tlib :: Package base :: Module MysqlConnector |
[hide private]
[frames] | no frames]

# Source Code for Module tlib.base.MysqlConnector

      1  # @author dumitru.fostic@ypg.com
      2  # @date   May , 2014
      3  import MySQLdb
      4  import warnings
      5  import pytest
      6
      7
      8 -class MysqlConnector(object):
      9      """
     10      Connects to MySQL DB, executes queries
     11      Autocommit is disabled by default, commit is done with commit function
     12      Warnings are turned into exceptions
     13      """
     14 -    def __init__(self, config_data):
     15          """
     16          Initialize DB connection. Turning warning into exceptions
     17          """
     18          warnings.filterwarnings('error', category=MySQLdb.Warning)  # turn warnings into exceptions
     19          self.db = None
     20          try:
     21              self.db = MySQLdb.connect(**config_data)
     22              self.cursor = self.db.cursor(MySQLdb.cursors.DictCursor)
     23              self.db.autocommit(False)   # in case of autocommit will change by default ot ON in future
     24          except MySQLdb.Error, err:
     25              print "Error %d: %s" % (err.args[0], err.args[1])
     26              pytest.fail('FAILED connecting to DB with error: %d: %s' % (err.args[0], err.args[1]))
     27
     28 -    def get_autocommit_status(self, var_type='local'):
     29          """
     30          Returns status of autocommit LOCAL/GLOBAL/SESSION variables (local by default)
     31          :return: tuple(dict) i.e. ({'Value': 'OFF', 'Variable_name': 'autocommit'},)
     32          """
     33          if (var_type in ('local', 'global', 'session')) is False:
     34              pytest.fail("Wrong argument '%s' passed in <get_auto_commit> function. Valid args are str: 'local', 'global' or 'session'" % var_type)
     35
     36          status_query = "show " + var_type + " variables where variable_name = 'autocommit'"
     37          return self.select_query(status_query)
     38
     39 -    def select_query(self, query, fetch='all'):
     40          """
     41          Send/execute queries
     42          :param: str MySQL query
     43          :param: str 'one'/'all' Rows to be fetched
     44          :return: fetch = one  dictionary i.e. {row_1}
     45          :return: fetch != one tuple with arguments dictionaries, each dictionary is a row i.e. ({row_first}, {row_second}, ... ,{row_last})
     46          """
     47          try:
     48              self.cursor.execute(query)
     49              data = self.cursor.fetchone() if fetch == 'one' else self.cursor.fetchall()
     50              return data
     51          except MySQLdb.Error, err:
     52              print "Error %d: %s" % (err.args[0], err.args[1])
     53              pytest.fail('<select_query()> function FAILURE with error: %d: %s' % (err.args[0], err.args[1]))
     54
     55 -    def update_query(self, query):
     56          """
     57          Sends query to DB.
     58          :param: str query
     59          :return: int Nr of rows updated (in case of updating with same value, query is successful but nr of rows updated  = 0
     60          """
     61          try:
     62              self.cursor.execute(query)
     63              rows_updated = self.cursor.rowcount
     64              print "Number of rows updated: %s" % rows_updated
     65          except MySQLdb.Error, err:
     66              print "Error %d: %s" % (err.args[0], err.args[1])
     67              pytest.fail('<update_query()> function FAILURE with error: %d: %s' % (err.args[0], err.args[1]))
     68
     69          return rows_updated
     70
     71 -    def delete_query(self, query):
     72          """
     73          to be updated: don't have permission to delete on current working DB
     74          """
     75          try:
     76              self.cursor.execute(query)
     77              rows_deleted = self.cursor.rowcount
     78              print "Number of rows deleted: %s" % rows_deleted
     79          except MySQLdb.Error, err:
     80              print "Error %d: %s" % (err.args[0], err.args[1])
     81              pytest.fail('<delete_query()> function FAILURE with error: %d: %s' % (err.args[0], err.args[1]))
     82
     83          return rows_deleted
     84
     85 -    def commit(self):
     86          self.db.commit()
     87
     88 -    def rollback(self):
     89          self.db.rollback()
     90
     91 -    def close_cursor(self):
     92          self.cursor.close()
     93
     94 -    def close_db(self):
     95          self.db.close()
     96
     97 -    def __del__(self):
     98          self.cursor.close()
     99          self.db.close()
    100