基于底層數據來開發不難，無非是將用戶輸入變量作為篩選條件，將參數映射到 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 pd
pd.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 pd
pd.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 os
os.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=False
continue
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_DB
import pandas as pd
from 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 自助取數查詢的資料請關注腳本之家其它相關文章！
您可能感興趣的文章:- Python數據分析之pandas讀取數據
- Python 循環讀取數據內存不足的解決方案
- Python隨機函數random隨機獲取數字、字符串、列表等使用詳解
- python實現scrapy爬蟲每天定時抓取數據的示例代碼
- Python從文件中讀取數據的方法步驟
- python從PDF中提取數據的示例
- python從Oracle讀取數據生成圖表
- python3:excel操作之讀取數據并返回字典 + 寫入的案例
- Python爬取數據并實現可視化代碼解析
- Python定時從Mysql提取數據存入Redis的實現
- 使用Python腳本從文件讀取數據代碼實例
- python3實現從kafka獲取數據,并解析為json格式,寫入到mysql中
