Python實現一個自助取數查詢工具

基于底層數據來開發不難，無非是將用戶輸入變量作為篩選條件，將參數映射到 sql 語句，并生成一個 sql 語句然后再去數據庫執行

最后再利用 QT 開發一個 GUI 界面，用戶界面的點擊和篩選條件，信號觸發對應按鈕與綁定的傳參槽函數執行

一、數據庫連接類

此處利用 pandas 讀寫操作 oracle 數據庫

二、主函數模塊

1）輸入參數模塊，外部輸入條件參數，建立數據庫關鍵字段映射

--注：讀取外部 txt 文件，將篩選字段可能需要進行鍵值對轉換

2）sql 語句集合模塊，將待執行的業務 sql 語句統一存放到這里

3）數據處理函數工廠

4）使用多線程提取數據

cx_Oracle 是一個 Python 擴展模塊，相當于 python 的 Oracle 數據庫的驅動，通過使用所有數據庫訪問模塊通用的數據庫 API 來實現 Oracle 數據庫的查詢和更新

Pandas 是基于 NumPy 開發，為了解決數據分析任務的模塊，Pandas 引入了大量庫和一些標準的數據模型，提供了高效地操作大型數據集所需的方法類和函數

pandas 調用數據庫主要有 read_sql_table，read_sql_query，read_sql 三種方式

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

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

以下創建連接 oracel 數據庫的連接類 Oracle_DB

主要提供 2 種操作數據的函數方法。

import cx_Oracle# Pandas讀寫操作Oracle數據庫import pandas as pd# 避免編碼問題帶來的亂碼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: # #查詢獲取數據用sql語句 # 代傳參數：sparm--查詢指定字段參數 df = pd.read_sql_query(sql, self.engine,params=sparm) self.engine.close()except Exception as e: return 'Error ' + e.args[0]return df # 查詢全部信息 def search_all(self, sql):try: # #查詢獲取數據用sql語句 df = pd.read_sql_query(sql, self.engine) self.engine.close()except Exception as e: return 'Error ' + e.args[0]return df二、數據提取主函數模塊

cx_Oracle 是一個 Python 擴展模塊，相當于 python 的 Oracle 數據庫的驅動，通過使用所有數據庫訪問模塊通用的數據庫 API 來實現 Oracle 數據庫的查詢和更新。

1）外部輸入參數模塊

txt 文本中，就包含一列數據，第一行列名，讀取的時候忽略第一行

#建立ID——編號字典def buildid(): sqlid = '''select * from b_build_info''' db = Oracle_DB() # 實例化一個對象 b_build_info = db.search_all(sqlid) ID_bUILDCODE = b_build_info.set_index('BUILDCODE')['ID'].to_dict() return ID_bUILDCODE #通過文本傳入待導出數據清單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’))) #鍵值對轉換 return tasklist

2）業務 sql 語句集合

注意in后面{0}不要加引號，這里傳入為元組，params 參數傳入sparm

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

def sql_d(lst): # 逐月數據 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) # 逐月數據 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) # 查詢當日數據 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語句 return sql_collection

3）業務數據處理

業務數據處理流程，原始數據后處理，這里不作介紹：

def db_extranction(lst,sparm,sql_type): '''sql_type--輸入需要操作的sql業務序號''' sql_=sql_d(lst)[sql_type] #輸出sql語句 db = Oracle_DB() # 實例化一個對象 res=db.search_one(sql_,sparm) # 數據處理加工 RES=Data_item_factory(res) #此處省略 # res = db.search_all(sql_d_energy_item_month) print(RES) return RES

多線程提取數據部分，這里 tasklist 列表多線程提取數據

import threading# Pandas讀寫操作Oracle數據庫from tools.Data_Update_oracle import Oracle_DBimport pandas as pdfrom concurrent import futures if __name__ == ’__main__’: #外部傳入 tasklist= read_task_list() print(tasklist) # 輸入時間查找范圍參數，可手動修改 sparm = {’Start_time’:’2021-04-01’,’End_time’:’2021-05-01’} lst = tuple(list(tasklist))#業務類型序號，可手動修改 sql_type=0#全部提取 db_extranction(lst,sparm,sql_type) #多線程按字段分批提取 方法一：使用threading模塊的Thread類的構造器創建線程 #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庫，這是官方基于 threading 封裝，先安裝該庫 # with futures.ThreadPoolExecutor(len(tasklist)) as executor: # executor.map([db_extranction(lst,sparm,sql_type) for lst in tasklist],tasklist)

到此整個數據庫取數工具開發流程介紹完畢，就差最后一步分享給小伙伴使用了，做成 GUI 應用此處不做詳細介紹，構建獨立的 python 環境，快速發布你的應用

以上就是Python實現一個自助取數查詢工具的詳細內容，更多關于python 自助取數查詢的資料請關注好吧啦網其它相關文章！