Python實(shí)現(xiàn)一個(gè)自助取數(shù)查詢工具

第Python實(shí)現(xiàn)一個(gè)自助取數(shù)查詢工具#fromsqlalchemyimportcreate_engine

#engine=create_engine('oracle+cx_oracle://username:password@ip:1521/ORCL')

##方法2：cx_Oracle.connect()

self.engine=cx_Oracle.connect('username','password','ip:1521/database')

exceptcx_Oracle.Errorase:

print("Error%d：%s"%(e.args[0],e.args[1]))

exit()

#查詢部分信息

defsearch_one(self,sql,sparm):

try:

##查詢獲取數(shù)據(jù)用sql語句

#代傳參數(shù)：sparm--查詢指定字段參數(shù)

df=pd.read_sql_query(sql,self.engine,params=sparm)

self.engine.close()

exceptExceptionase:

return"Error"+e.args[0]

returndf

#查詢?nèi)啃畔?/p>

defsearch_all(self,sql):

try:

##查詢獲取數(shù)據(jù)用sql語句

df=pd.read_sql_query(sql,self.engine)

self.engine.close()

exceptExceptionase:

return"Error"+e.args[0]

returndf

二、數(shù)據(jù)提取主函數(shù)模塊

cx_Oracle是一個(gè)Python擴(kuò)展模塊，相當(dāng)于python的Oracle數(shù)據(jù)庫(kù)的驅(qū)動(dòng)，通過使用所有數(shù)據(jù)庫(kù)訪問模塊通用的數(shù)據(jù)庫(kù)API來實(shí)現(xiàn)Oracle數(shù)據(jù)庫(kù)的查詢和更新。

1）外部輸入?yún)?shù)模塊

txt文本中，就包含一列數(shù)據(jù)，第一行列名，讀取的時(shí)候忽略第一行

#建立ID——編號(hào)字典

defbuildid():

sqlid="""select*fromb_build_info"""

db=Oracle_DB()#實(shí)例化一個(gè)對(duì)象

b_build_info=db.search_all(sqlid)

ID_bUILDCODE=b_build_info.set_index("BUILDCODE")["ID"].to_dict()

returnID_bUILDCODE

#通過文本傳入待導(dǎo)出數(shù)據(jù)清單

defread_task_list():

build_code=buildid()

tasklist=[]

is_first_line=True

withopen("./b_lst.txt")aslst:

forlineinlst:

ifis_first_line:

is_first_line=False

continue

tasklist.append(build_code.get(line.strip('\n')))#鍵值對(duì)轉(zhuǎn)換

returntasklist

2）業(yè)務(wù)sql語句集合

注意in后面{0}不要加引號(hào)，這里傳入為元組，params參數(shù)傳入sparm

={'Start_time':'2025-04-01','End_time':'2025-05-01'}，此處參數(shù)可根據(jù)需要改變

defsql_d(lst):

#逐月數(shù)據(jù)

sql_d_energy_item_month="""select*fromd_energy_item_month

whererecorddate=to_date(:Start_time,'yyyy-MM-dd')

andrecorddateto_date(:End_time,'yyyy-MM-dd')

andbuildidin{0}

orderbyrecorddateasc""".format(lst)

#逐月數(shù)據(jù)

sql_d_energy_month="""selectd.*,fromd_energy_monthdjoint_device_infotond.branchid=t.id

whered.recorddate=to_date(:Start_time,'yyyy-MM-dd')

andd.recorddateto_date(:End_time,'yyyy-MM-dd')

andd.buildid='{0}'

orderbyd.recorddateasc""".format(lst)

#查詢當(dāng)日數(shù)據(jù)

sql_energy_item_hour_cheak="""select*fromd_energy_item_hour

wheretrunc(sysdate)=trunc(recorddate)

orderbyrecorddateasc""".format(lst)

sql_collection=[sql_d_energy_item_month,sql_d_energy_item_day,sql_d_energy_item_hour,sql_d_energy_month,

sql_d_energy_day,sql_d_energy_hour,sql_energy_hour_cheak]

#此處省略部分sql語句

returnsql_collection

3）業(yè)務(wù)數(shù)據(jù)處理

業(yè)務(wù)數(shù)據(jù)處理流程，原始數(shù)據(jù)后處理，這里不作介紹：

defdb_extranction(lst,sparm,sql_type):

"""sql_type--輸入需要操作的sql業(yè)務(wù)序號(hào)"""

sql_=sql_d(lst)[sql_type]#輸出sql語句

db=Oracle_DB()#實(shí)例化一個(gè)對(duì)象

res=db.search_one(sql_,sparm)

#數(shù)據(jù)處理加工

RES=Data_item_factory(res)#此處省略

#res=db.search_all(sql_d_energy_item_month)

print(RES)

returnRES

多線程提取數(shù)據(jù)部分，這里tasklist列表多線程提取數(shù)據(jù)

importthreading

#Pandas讀寫操作Oracle數(shù)據(jù)庫(kù)

fromtools.Data_Update_oracleimportOracle_DB

importpandasaspd

fromconcurrentimportfutures

if__name__=='__main__':

#外部傳入

tasklist=read_task_list()

print(tasklist)

#輸入時(shí)間查找范圍參數(shù)，可手動(dòng)修改

sparm={'Start_time':'2025-04-01','End_time':'2025-05-01'}

lst=tuple(list(tasklist))

#業(yè)務(wù)類型序號(hào)，可手動(dòng)修改

sql_type=0

#全部提取

db_extranction(lst,sparm,sql_type)

#多線程按字段分批提取

方法一：使用threading模塊的Thread類的構(gòu)造器創(chuàng)建線程

#threads=[threading.Thread(target=db_extranction,args=(lst,sparm,sql_type))forlstintasklist]

#[threads[i].start()foriinrange(len(threads))]

方法二：使用python的concurrent庫(kù)，這是官方基于threading封裝，先安裝該庫(kù)

#withfutures.ThreadPoolExecutor(len(tasklist))asexecutor:

#executor.map([db_extranction(lst,sp