onemap-implementationSupervision-system-python/db/oracle.py

# -*- coding: utf-8 -*-

#设置语言环境
import os
import sys
import json
import log
reload(sys)
sys.setdefaultencoding('utf-8')
os.environ['NLS_LANG'] = 'SIMPLIFIED CHINESE_CHINA.UTF8'

import cx_Oracle

class Oracle:
    def __init__(self, conn):
        """Oracle
        C:\Python27\ArcGISx6410.8\Scripts
        pip install cx-Oracle==7.3.0
        :param conn：数据库连接字符串（'user/password@127.0.0.1/orcl'）
        """
        self._conn = cx_Oracle.connect(conn, encoding="UTF-8", nencoding="UTF-8")
        self.cursor = self._conn.cursor() #游标

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

    def close(self):
        try:
            if hasattr(self,'cursor'): 
                self.cursor.close()
            if hasattr(self,'_conn'): 
                self._conn.close()
        except:
            msg = str(sys.exc_info()).decode('string-escape')
            log.error(msg)

    def __del__(self):
        self.close()

    def query(self, sql):
        """查询所有数据
        :param sql：SQL语句
        """
        self.cursor.execute(sql)
        ds = self.cursor.fetchall()
        cols = [d[0] for d in self.cursor.description]
        list = []
        for row in ds:
            b = dict(zip(cols, row)) 
            list.append(b)
        
        # 处理可能存在的LOB对象，确保可以JSON序列化
        for item in list:
            for key, value in item.items():
                # 如果值是LOB对象，尝试读取其内容
                if hasattr(value, 'read'):
                    try:
                        item[key] = value.read()
                    except:
                        item[key] = str(value)
        
        jobj = json.loads(json.dumps(list, ensure_ascii=False))
        return jobj

    def insert(self, sql, dict = None):
        """查询数据
        :param sql：SQL语句 ==> insert into 表(字段1,字段2) values(:字段1,:字段2)
        :param dict：值字典 ==> {"字段1":"abc", "字段2":"abc"}
        """
        if dict == None:
            self.cursor.execute(sql)
        else:
            self.cursor.execute(sql, dict)
        self.commit()