将查询结果放入Python中的对象中

问题描述 投票:0回答:1

下午好

我正在尝试使用 cx_Oracle 8.3.0 / python 编译良好的信息并将获取的结果存储在对象列表中,如下所示。

import cx_Oracle
def getOraCon(daDsn):
    global oraCon
    oraCon = cx_Oracle.connect(user='my_uid',password='my_pwd',dsn=daDsn)
    global oraCursor
    oraCursor = oraCon.cursor()
    executeDDLOracle("alter session set nls_date_format = 'YYYY-MM-DD HH24:MI:SS'")

def getFetch(sql,cur):
    cur.execute(sql)
    return cur.fetchall()
    
def executeDDLOracle(strSQL):
    oraCursor.execute(strSQL)
    oraCon.commit()

def closeDbObject(dbObj):
    dbObj.close()
    dbObj = None
    del dbObj
    
def closeOra():
    oraCon.commit()
    closeDbObject(oraCursor)
    closeDbObject(oraCon)

def getFromList():
    getOraCon('CORPORATE_DB')
    executeDDLOracle('DROP TYPE T_TABLE')
    executeDDLOracle('create or replace TYPE T_RECORD IS OBJECT (w_id integer,w_lat varchar2(20), 
    sta_dpth integer,end_dpth integer)')
    executeDDLOracle('CREATE OR REPLACE TYPE T_TABLE AS TABLE OF T_RECORD')

    list_type = oraCon.gettype("T_RECORD")
    
    wb_list_main = list_type.newobject()
    wb_fetch=getFetch("select w_id, w_lat,sta_dpth, end_dpth from wb_data where lat_num >0", 
    oraCursor)
    for w_id, w_lat,sta_dpth, end_dpth in wb_fetch:
        wb_list = list_type.newobject()
        wb_list.type.attributes[0]=w_id
        wb_list.type.attributes[1]=w_lat
        wb_list.type.attributes[2]=sta_dpth
        wb_list.type.attributes[3]=end_dpth 
        wb_list_main.append(wb_list)
    closeOra()

我收到错误消息:

<class 'cx_Oracle.NotSupportedError'>: Python value cannot be converted to a database value

在:

        wb_list_main.append(list_type)

我将不胜感激任何帮助。谢谢你。

python cx-oracle
1个回答
0
投票

下午好,

我想我会与你们分享我所发现的一切。我希望这对其他人的工作有所帮助。

import cx_Oracle


def getOraCon(daDsn):
    global oraCon
    oraCon = cx_Oracle.connect(user='minhaszt',password='Dammam_2023',dsn=daDsn)
    global oraCursor
    oraCursor = oraCon.cursor()
    executeDDLOracle("alter session set nls_date_format = 'YYYY-MM-DD HH24:MI:SS'")

def makeDict(sql,cur):
    fetch = getFetch(sql,cur)
    return {key:val for key,val in fetch}   

def getFetch(sql,cur):
    cur.execute(sql)
    return cur.fetchall()
    
def executeDDLOracle(strSQL):
    oraCursor.execute(strSQL)
    oraCon.commit()
def closeDbObject(dbObj):
    dbObj.close()
    dbObj = None
def closeOra():
    oraCon.commit()
    closeDbObject(oraCursor)
    closeDbObject(oraCon)
    
def getData():
    getOraCon('CORPORATE_DB')
    executeDDLOracle('DROP TYPE T_TABLE')
    executeDDLOracle('create or replace TYPE T_RECORD IS OBJECT (w_id integer,w_lat 
    varchar2(20),sta_dpth integer,end_dpth integer)')
    executeDDLOracle('CREATE TYPE T_TABLE AS TABLE OF T_RECORD')
    list_type = oraCon.gettype("T_RECORD")
    wb_list = list_type.newobject()
    
    table_type = oraCon.gettype("T_TABLE")
    wb_table = table_type.newobject()

    wb_fetch=getFetch("select w_id, w_lat,sta_dpth, end_dpth from wb_data where lat_num >0", oraCursor) 
    
    for w_id, w_lat,sta_dpth, end_dpth in wb_fetch:
        wb_list.W_ID=w_id
        wb_list.W_LAT =w_lat
        wb_list.STA_DPTH =sta_dpth
        wb_list.END_DPTH =end_dpth 
        wb_table.append(wb_list)
    wb_fetch = None
    del wb_fetch
    test_fetch= oraCursor.execute("select * from table (:id_values)", 
    id_values=wb_table)
    
    for d in test_fetch.description:
        print(d)
    for r in test_fetch:
        print(r)

顺便说一句,虽然我同意理想情况下我应该将驱动程序更新到 oracledb,但目前我没有选择更新驱动程序。

© www.soinside.com 2019 - 2024. All rights reserved.