import psycopg2 class PgSQLConnection(object): """aaaaaaa""" __conection_object = None __db_config = {'dbname': 'testsrv', 'host': '192.168.2.171', 'password': 'Idkfa2009', 'port': 5432, 'user': 'admin'} def __init__(self, connect_str=None): self.__conection_object = None if connect_str is not None: self.__db_config = connect_str else: self.__db_config = {'dbname': 'testsrv', 'host': '192.168.2.171', 'password': 'Idkfa2009', 'port': 5432, 'user': 'admin'} def db_connect(self, connect_str=None): result = False try: if connect_str == None: self.__conection_object = psycopg2.connect(**self.__db_config) else: self.__db_config = connect_str self.__conection_object = psycopg2.connect(**self.__db_config) self.__conection_object.autocommit = True result = True except Exception as error: print(error) return result def getConfig(self): return self.__db_config def db_execute_query(self, query): cur = self.__conection_object.cursor() cur.execute(query) data = cur.fetchall() cur.close() return data def db_upload_large_file(self, filepath): # a new connection must be created to use large objects __conn = psycopg2.connect(**self.__db_config) lobj = __conn.lobject(oid=0, mode="rw", new_oid=0, new_file=filepath) fileoid = lobj.oid lobj.close() __conn.commit() __conn.close() return fileoid def commit(self): self.__conection_object.commit() def db_close(self): if self.__conection_object is not None: self.__conection_object.close() del self.__conection_object self.__conection_object = None def __del__(self): if self.__conection_object is not None: self.__conection_object.close() # print("disconnected")