手把手教你搭建一個 Python 連接數(shù)據(jù)庫，快速取數(shù)工具

1：pd.read_sql_query()
讀取自定義數(shù)據(jù)，返還DataFrame格式，通過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ù)庫連接
index_col:選擇返回結(jié)果集索引的列,文本/文本列表
coerce_float:非常有用，將數(shù)字形式的字符串直接以float型讀入
parse_dates:將某一列日期型字符串轉(zhuǎn)換為datetime型數(shù)據(jù)，與pd.to_datetime函數(shù)功能類似。
params:向sql腳本中傳入的參數(shù)，官方類型有列表，元組和字典。用于傳遞參數(shù)的語法是數(shù)據(jù)庫驅(qū)動程序相關(guān)的。
chunksize：如果提供了一個整數(shù)值，那么就會返回一個generator，每次輸出的行數(shù)就是提供的值的大小

read_sql_query（）中可以接受SQL語句，DELETE,INSERT INTO、UPDATE操作沒有返回值（但是會在數(shù)據(jù)庫中執(zhí)行），程序會拋出SourceCodeCloseError，并終止程序。SELECT會返回結(jié)果。如果想繼續(xù)運行，可以try捕捉此異常。
 
2：pd.read_sql_table()
讀取數(shù)據(jù)庫中的表，返還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()
讀數(shù)據(jù)庫通過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)

import cx_Oracle
# Pandas讀寫操作Oracle數(shù)據(jù)庫
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:
            # #查詢獲取數(shù)據(jù)用sql語句
            # 代傳參數(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)啃畔?/span>
    def search_all(self, sql):
        try:

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

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

            self.engine.close()

        except Exception as e:
            return "Error " + e.args[0]

        return df

#建立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
    
#通過文本傳入待導(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=False
                continue
            tasklist.append(build_code.get(line.strip('\n')))  #鍵值對轉(zhuǎn)換
    return tasklist

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語句
    return sql_collection

def db_extranction(lst,sparm,sql_type):   
    """sql_type--輸入需要操作的sql業(yè)務(wù)序號"""
    sql_=sql_d(lst)[sql_type]  #輸出sql語句
    db = Oracle_DB()  # 實例化一個對象
    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

import threading
# Pandas讀寫操作Oracle數(shù)據(jù)庫
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)
    # 輸入時間查找范圍參數(shù)，可手動修改
    sparm = {'Start_time':'2021-04-01','End_time':'2021-05-01'}
    lst = tuple(list(tasklist))
    
    #業(yè)務(wù)類型序號，可手動修改
    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庫，這是官方基于 threading 封裝，先安裝該庫
    # with futures.ThreadPoolExecutor(len(tasklist)) as executor:
    #     executor.map([db_extranction(lst,sparm,sql_type) for lst in tasklist],tasklist)