Python操作数据库及应用

Utils.DBHelper 

import cx_Oracle


class OracleHandler(object):

    def __init__(self, name):
        if name == 'PULSE':
            self.user = 'user'
            self.password = 'password'
            self.host = web_mes

        else:
            raise Exception('Can not recognise the user name, please contact to SW administrator')
        self._conn = cx_Oracle.connect(self.user, self.password, self.host)
        self.cursor = self._conn.cursor()

    def get_version(self):
        return self._conn.version

    def query_all(self, query):
        self.cursor.execute(query)
        return self.cursor.fetchall()

    def query_one(self, query):
        self.cursor.execute(query)
        return self.cursor.fetchone()

    def insert_many(self, sql_query, insert_param):
        if len(insert_param) == 0:
            raise Exception('Insert parameters cannot be null')
        self.cursor.prepare(sql_query)
        result = self.cursor.executemany(None, insert_param)
        self.commit()
        return result

    def insert_one(self, sql_query, insert_param):
        if len(insert_param) == 0:
            raise Exception('Insert parameters cannot be null')
        self.cursor.prepare(sql_query)
        result = self.cursor.execute(None, insert_param)
        self.commit()
        return result

    def delete_data(self, sql_query):
        self.cursor.execute(sql_query)
        self.commit()

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

    def __del__(self):
        if hasattr(self, 'cursor'):
            self.cursor.close()
        if hasattr(self, '_conn'):
            self._conn.close()


web_mes = """
        (DESCRIPTION =
            (ADDRESS_LIST =
            (ADDRESS = (PROTOCOL = TCP)(HOST = 10.xxx.xxx.xxx)(PORT = 1521))
            )
            (CONNECT_DATA =
            (SERVICE_NAME = prdinfo)
            )
        )
        """

应用

def try_get_exist_data(y, m,plant_code):
    conn = Utils.DBHelper.OracleHandler("PULSE")
    query = f'select * from t_xxx where year = {y} and month = {m} ' \
            f'and plant = {plant_code}'.format(y=y, m=m, plant_code=plant_code)
    try:
        result = conn.query_one(query)
    except Exception as e:
        log.error('An error occurred while get exist data, Error: ' + e.args[0])
    del conn
    if result is None:
        log.info('Data not exists')
        return False
    else:
        log.info('Data exists')
        return True

Query_One，Query_Many，Insert_One，Inser_Many 按情况使用即可