基于底層數(shù)據(jù)來(lái)開(kāi)發(fā)不難，無(wú)非是將用戶輸入變量作為篩選條件，將參數(shù)映射到 sql 語(yǔ)句，并生成一個(gè) sql 語(yǔ)句然后再去數(shù)據(jù)庫(kù)執(zhí)行

最后再利用 QT 開(kāi)發(fā)一個(gè) GUI 界面，用戶界面的點(diǎn)擊和篩選條件，信號(hào)觸發(fā)對(duì)應(yīng)按鈕與綁定的傳參槽函數(shù)執(zhí)行

一、數(shù)據(jù)庫(kù)連接類

此處利用 pandas 讀寫操作 oracle 數(shù)據(jù)庫(kù)

二、主函數(shù)模塊

1）輸入?yún)?shù)模塊，外部輸入條件參數(shù)，建立數(shù)據(jù)庫(kù)關(guān)鍵字段映射

--注：讀取外部 txt 文件，將篩選字段可能需要進(jìn)行鍵值對(duì)轉(zhuǎn)換

2）sql 語(yǔ)句集合模塊，將待執(zhí)行的業(yè)務(wù) sql 語(yǔ)句統(tǒng)一存放到這里

3）數(shù)據(jù)處理函數(shù)工廠

4）使用多線程提取數(shù)據(jù)

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

Pandas 是基于 NumPy 開(kāi)發(fā)，為了解決數(shù)據(jù)分析任務(wù)的模塊，Pandas 引入了大量庫(kù)和一些標(biāo)準(zhǔn)的數(shù)據(jù)模型，提供了高效地操作大型數(shù)據(jù)集所需的方法類和函數(shù)

pandas 調(diào)用數(shù)據(jù)庫(kù)主要有 read_sql_table，read_sql_query，read_sql 三種方式

本文主要介紹一下 Pandas 中 read_sql_query 方法的使用

1：pd.read_sql_query()讀取自定義數(shù)據(jù)，返還DataFrame格式，通過(guò)SQL查詢腳本包括增刪改查。pd.read_sql_query(sql, con, index_col=None,coerce_float=True, params=None, parse_dates=None,chunksize=None)sql：要執(zhí)行的sql腳本，文本類型con：數(shù)據(jù)庫(kù)連接index_col:選擇返回結(jié)果集索引的列,文本/文本列表coerce_float:非常有用，將數(shù)字形式的字符串直接以float型讀入parse_dates:將某一列日期型字符串轉(zhuǎn)換為datetime型數(shù)據(jù)，與pd.to_datetime函數(shù)功能類似。params:向sql腳本中傳入的參數(shù)，官方類型有列表，元組和字典。用于傳遞參數(shù)的語(yǔ)法是數(shù)據(jù)庫(kù)驅(qū)動(dòng)程序相關(guān)的。chunksize：如果提供了一個(gè)整數(shù)值，那么就會(huì)返回一個(gè)generator，每次輸出的行數(shù)就是提供的值的大小read_sql_query（）中可以接受SQL語(yǔ)句，DELETE,INSERT INTO、UPDATE操作沒(méi)有返回值（但是會(huì)在數(shù)據(jù)庫(kù)中執(zhí)行），程序會(huì)拋出SourceCodeCloseError，并終止程序。SELECT會(huì)返回結(jié)果。如果想繼續(xù)運(yùn)行，可以try捕捉此異常。 2：pd.read_sql_table()讀取數(shù)據(jù)庫(kù)中的表，返還DataFrame格式（通過(guò)表名）import pandas as pdpd.read_sql_table(table_name, con, schema=None,index_col=None, coerce_float=True, parse_dates=None, columns=None,chunksize=None) 3:pd.read_sql()讀數(shù)據(jù)庫(kù)通過(guò)SQL腳本或者表名import pandas as pdpd.read_sql(sql, con, index_col=None,coerce_float=True, params=None, parse_dates=None, columns=None, chunksize=None)

以下創(chuàng)建連接 oracel 數(shù)據(jù)庫(kù)的連接類 Oracle_DB

主要提供 2 種操作數(shù)據(jù)的函數(shù)方法。

import cx_Oracle# Pandas讀寫操作Oracle數(shù)據(jù)庫(kù)import pandas as pd# 避免編碼問(wèn)題帶來(lái)的亂碼import osos.environ[’NLS_LANG’] = ’SIMPLIFIED CHINESE_CHINA.UTF8’class Oracle_DB(object): def __init__(self):try: # 連接oracle # 方法1：sqlalchemy 提供的create_engine() # from sqlalchemy import create_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’)except cx_Oracle.Error as e: print('Error %d：%s' % (e.args[0], e.args[1])) exit()# 查詢部分信息 def search_one(self, sql,sparm):try: # #查詢獲取數(shù)據(jù)用sql語(yǔ)句 # 代傳參數(shù)：sparm--查詢指定字段參數(shù) df = pd.read_sql_query(sql, self.engine,params=sparm) self.engine.close()except Exception as e: return 'Error ' + e.args[0]return df # 查詢?nèi)啃畔? def search_all(self, sql):try: # #查詢獲取數(shù)據(jù)用sql語(yǔ)句 df = pd.read_sql_query(sql, self.engine) self.engine.close()except Exception as e: return 'Error ' + e.args[0]return df二、數(shù)據(jù)提取主函數(shù)模塊

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

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

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

#建立ID——編號(hào)字典def buildid(): sqlid = '''select * from b_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() return ID_bUILDCODE #通過(guò)文本傳入待導(dǎo)出數(shù)據(jù)清單def read_task_list(): build_code=buildid() tasklist=[] is_first_line=True with open('./b_lst.txt') as lst:for line in lst: if is_first_line:is_first_line=Falsecontinue tasklist.append(build_code.get(line.strip(’n’))) #鍵值對(duì)轉(zhuǎn)換 return tasklist

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

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

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

def sql_d(lst): # 逐月數(shù)據(jù) sql_d_energy_item_month = '''select * from d_energy_item_month where recorddate >= to_date(:Start_time, ’yyyy-MM-dd’) and recorddate < to_date(:End_time, ’yyyy-MM-dd’) and buildid in {0} order by recorddate asc'''.format(lst) # 逐月數(shù)據(jù) sql_d_energy_month = '''select d.*,t.name from d_energy_month d join t_device_info t on d.branchid = t.id where d.recorddate >= to_date(:Start_time, ’yyyy-MM-dd’) and d.recorddate < to_date(:End_time, ’yyyy-MM-dd’) and d.buildid = ’{0}’ order by d.recorddate asc'''.format(lst) # 查詢當(dāng)日數(shù)據(jù) sql_energy_item_hour_cheak = '''select * from d_energy_item_hour where trunc(sysdate)=trunc(recorddate) order by recorddate asc'''.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語(yǔ)句 return sql_collection

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

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

def db_extranction(lst,sparm,sql_type): '''sql_type--輸入需要操作的sql業(yè)務(wù)序號(hào)''' sql_=sql_d(lst)[sql_type] #輸出sql語(yǔ)句 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) return RES

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

import threading# Pandas讀寫操作Oracle數(shù)據(jù)庫(kù)from tools.Data_Update_oracle import Oracle_DBimport pandas as pdfrom concurrent import futures if __name__ == ’__main__’: #外部傳入 tasklist= read_task_list() print(tasklist) # 輸入時(shí)間查找范圍參數(shù)，可手動(dòng)修改 sparm = {’Start_time’:’2021-04-01’,’End_time’:’2021-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)) for lst in tasklist] # [threads[i].start() for i in range(len(threads))]方法二：使用python的concurrent庫(kù)，這是官方基于 threading 封裝，先安裝該庫(kù) # with futures.ThreadPoolExecutor(len(tasklist)) as executor: # executor.map([db_extranction(lst,sparm,sql_type) for lst in tasklist],tasklist)

到此整個(gè)數(shù)據(jù)庫(kù)取數(shù)工具開(kāi)發(fā)流程介紹完畢，就差最后一步分享給小伙伴使用了，做成 GUI 應(yīng)用此處不做詳細(xì)介紹，構(gòu)建獨(dú)立的 python 環(huán)境，快速發(fā)布你的應(yīng)用

以上就是Python實(shí)現(xiàn)一個(gè)自助取數(shù)查詢工具的詳細(xì)內(nèi)容，更多關(guān)于python 自助取數(shù)查詢的資料請(qǐng)關(guān)注好吧啦網(wǎng)其它相關(guān)文章！