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