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

Posted on by

基於底層數據來開發不難，無非是將用戶輸入變量作為篩選條件，將參數映射到 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 自助取數查詢的資料請關註WalkonNet其它相關文章！

推薦閱讀：