自学教程：Python实现一个自助取数查询工具

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)

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

#建立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

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

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

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)