Python数据核对系列之3-Python读取数据库

# 官方给出的案例 https://cx-oracle.readthedocs.io/en/latest/user_guide/connection_handling.html
userpwd = ". . ." # Obtain password string from a user prompt or environment variable
# hr：用户名；userpwd：密码；dbhost.example.com/orclpdb1：数据库地址，例如192.168.0.1:1521/orcl。oracle的默认端口为1521，默认数据库名称orcl
connection = cx_Oracle.connect("hr", userpwd, "dbhost.example.com/orclpdb1", encoding="UTF-8")

import cx_Oracle # 导入oracle操作库
conn = cx_Oracle.connect("user",
                         "password",
                         "IP:1521/orcl",
                         encoding="UTF-8")

import pymysql
conn = pymysql.connect(host="localhost",
                       user="user",
                       password="passwd",
                       db="db",
                       charset="utf-8")

import pymssql
conn = pymssql.connect('IP',
                       'user',
                       'password',
                       'database')
                       
# 例如：
conn = pymssql.connect('192.168.0.1','user','123456','dbname')

建立数据库连接-连接池连接

class PooledDB(creator,  # 数据库接口，上面提到的cx_Oracle、pymssql、pymysql
               mincached=0, # 启动时开启的空连接数量
               maxcached=0, # 连接池最大可用连接数量
               maxshared=0, # 连接池最大可共享连接数量
               maxconnections=0, # 最大允许连接数量
               blocking=False, # 最大允许连接数量
               maxusage=None, # 单个连接最大复用次数
               setsession=None, # 用于传递到数据库的准备会话，如 [”set name UTF-8″] 。
               reset=True, 
               failures=None, 
               ping=1, 
               *args, 
               **kwargs)
# After you have created the connection pool, 
# you can use connection() to get pooled, 
# steady DB-API 2 connections


from DBUtils.PooledDB import PooledDB
pool = PooledDB(cx_Oracle,
                mincached=2,
                maxconnections=6,
                user='user',
                password='123456',
                dsn='192.168.0.1:1521/orcl')
conn = pool.connection() # 通过连接池建立连接

# 先
sql = '''
  select * 
  from tablename
''' # 三引号可以编辑文本块
with conn.cursor() as cur:
    cur.execute(sql) # 让数据库执行SQL语句
    # SQL执行后，cursor默认是在第一条数据上
    cur.fetchone() # 获取SQL执行后的表的一条数据。执行后cursor移动到第2条数据上。
    cur.fetchmany(100) 
    # 获取SQL执行后的表的100条数据。
    # 此时是从第2条数据开始读取的，读到101条结束。
    # cursor移动到102条数据上
    cur.fetchall() # 获取第102及之后的所有数据
    cur.description # 获取SQL执行结果的列信息
    cur.commit() # 如果sql是insert/delet/update操作必须进行commit操作，不然操作不会入库。

column = []
for i in range(len(colunms)):
    column.append(colunms[i][0])
print(column)

import pandas as pd
class DataFrame(data=None, # 要放入DataFrame的数据，可以是2维数组、字典或者DataFrame数据
                index: Optional[Axes]=None, # 索引列名称
                columns: Optional[Axes]=None, # 各列名称
                dtype: Optional[Dtype]=None, # 各列数据类型
                copy: bool=False) # 
df = pd.DataFrame(data=data,columns=column)

pd.read_sql()获取数据库数据

pandas.read_sql(sql,  # 要执行的sql
                con,  # 数据库连接对象
                index_col='None', # 索引列
                coerce_float='True', # 是否将非字符串、非数字对象(如decimal.Decimal)的值转换为浮点值，这对SQL结果集很有用。
                params='None', 
                parse_dates='None', # 要解析为date类型的数据列
                columns='None', # 列名称
                chunksize: None = 'None')  # 执行SQL查询生成DataFrame，如果chunksize不为None，
                # 则生成可迭代的DataFrame对象，每个DataFrame块包含chunksize设置的行数的数据。
                # 也就是通过chunksize将SQL执行后的数据分成几块。
df2 = pd.read_sql(sql,BI_conn)