pydbapi
Installation
pip install pydbapi
支持的数据库类型
from pydbapi.api import SqliteDB
db = SqliteDB(database=None)
sql = 'select * from [table];'
row, action, result = db.execute(sql)
from pydbapi.api import RedshiftDB
db = RedshiftDB(host, user, password, database, port='5439', safe_rule=True)
sql = 'select * from [schema].[table];'
row, action, result = db.execute(sql)
from pydbapi.api import MysqlDB
db = MysqlDB(host, user, password, database, port=3306, safe_rule=True, isdoris=False)
sql = 'select * from [table];'
row, action, result = db.execute(sql)
from pydbapi.api import TrinoDB
db = TrinoDB(host, user, password, database, catalog, port=8443, safe_rule=True)
sql = 'select * from [table];'
row, action, result = db.execute(sql)
from pydbapi.api import SnowflakeDB
db = SnowflakeDB(user, password, account, warehouse, database, schema, safe_rule=True)
sql = 'select * from [table];'
row, action, result = db.execute(sql)
from pydbapi.api import SqliteDB
db = SqliteDB.get_instance(database=None)
sql = 'select * from [table];'
row, action, result = db.execute(sql)
Column
from pydbapi.col import ColumnModel
-
ColumnModel
- 代码
col = ColumnModel(newname, coltype='varchar', sqlexpr=None, func=None, order=0)
- params
newname
: 新命名;coltype
: 类型sqlexpr
: 查询sql表达式func
: 查询函数,暂时支持'min', 'max', 'sum', 'count'order
: 排序
-
ColumnsModel
- 代码
cols = ColumnsModel(ColumnModel, ColumnModel, ……)
- property
func_cols
: 返回col列表nonfunc_cols
: 返回col列表new_cols
: 返回拼接字符串create_cols
: 返回拼接字符串select_cols
: 返回拼接字符串group_cols
: 返回拼接字符串order_cols
: 返回拼接字符串
- mothed
- get_column_by_name
cols.get_column_by_name(name)
- 返回
ColumnModel
支持的操作
- execute【db/base.py】
- 代码
db.execute(sql, count=None, ehandling=None, verbose=0)
- params
count
: 返回结果的数量;ehandling
: sql执行出错的时候处理方式, default: Noneverbose
: 执行的进度展示方式(0:不打印, 1:文字进度, 2:进度条)
- select
- 代码
db.select(tablename, columns, condition=None, verbose=0)
- params
tablename
: 表名;columns
: 列内容;condition
: sql where 中的条件
- create
- sqlite/redshift
- 代码
db.create(tablename, columns, indexes=None, verbose=0)
- params
tablename
: 表名;columns
: 列内容;indexes
: 索引,sqlite暂不支持索引verbose
: 是否打印执行进度。
- mysql
- 代码
db.create(tablename, columns, indexes=None, index_part=128, ismultiple_index=True, partition=None, verbose=0)
- params
tablename
: 表名;columns
: 列内容;indexes
: 索引index_part
: 索引partismultiple_index
: 多重索引partition
: 分区verbose
: 是否打印执行进度。
- trino
- 代码
db.create(tablename, columns, partition=None, verbose=0)
- params
tablename
: 表名;columns
: 列内容;partition
: 分区verbose
: 是否打印执行进度。
- insert【db/base.py】
- 代码
db.insert(tablename, columns, inserttype='value', values=None, chunksize=1000, fromtable=None, condition=None)
- params
tablename
: 表名;columns
: 列内容;inserttype
: 插入数据类型,支持value、selectvalues
: inserttype='value',插入的数值;chunksize
: inserttype='value', 每个批次插入的量级;fromtable
: inserttype='select',数据来源表;condition
: inserttype='select',数据来源条件;
- drop【db/base.py】
- 代码
db.drop(tablename)
- params
- delete【db/base.py】
- 代码
db.delete(tablename, condition)
- params
tablename
: 表名;condition
: 插入的数值;
- get_columns
- 代码
db.get_columns(tablename)
- params
- add_columns
- 代码
db.add_columns(tablename, columns)
- params
tablename
: 表名;columns
: 列内容;
- get_filesqls【db/fileexec.py】
- 代码
db.get_filesqls(filepath, **kw)
- params
filepath
: sql文件路径;kw
: sql文件中需要替换的参数,会替换sqlfile中的arguments;
- file_exec【db/fileexec.py】
魔法命令
- 使用
load_ext
%load_ext pydbapi
- 配置startup文件夹
- AUTO_RULES
可以自动执行表名(表名包含即可) - REDSHIFT_AUTO_RULES
Amazon Redshift 可以自动执行表名(表名包含即可)
调用日志格式
- 查看每步sql可以使用如下日志格式(如果还出错,同时加上上面的内容)
import logging
dblogger = logging.getLogger('pydbapi.db.base')
dblogger.setLevel(logging.DEBUG)