pip install pydbapi
from pydbapi.api import SqliteDB
db = SqliteDB(database=None) # 或者传入路径
sql = 'select * from [table];'
cursor, 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];'
cursor, 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];'
cursor, 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];'
cursor, 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];'
cursor, action, result = db.execute(sql)
from pydbapi.api import SqliteDB
db = SqliteDB.get_instance(database=None) # 或者传入路径
sql = 'select * from [table];'
cursor, 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];'
cursor, action, result = db.execute(sql)
df = result.to_dataframe()
df
from pydbapi.api import TrinoDB
db = TrinoDB(host, user, password, database, catalog, port=8443, safe_rule=True)
sql = 'select * from [table];'
cursor, action, result = db.execute(sql)
result.to_csv(outfile)
from pydbapi.model import ColumnModel
ColumnModel
col = ColumnModel(newname, coltype='varchar', sqlexpr=None, func=None, order=0)newname: 新命名；coltype: 类型sqlexpr: 查询sql表达式func: 查询函数，暂时支持'min', 'max', 'sum', 'count'order: 排序ColumnsModel
cols = ColumnsModel(ColumnModel, ColumnModel, ……)func_cols: 返回col列表nonfunc_cols: 返回col列表new_cols: 返回拼接字符串create_cols: 返回拼接字符串select_cols: 返回拼接字符串group_cols: 返回拼接字符串order_cols: 返回拼接字符串cols.get_column_by_name(name)ColumnModelfrom pydbapi.sql import SqlStatement
sql = 'select * from tablename where part_date >= $part_date;'
sqlstmt = SqlStatement(sql)
属性
方法
sstmt1 = '-- comment'
sqlstmt = SqlStatement.from_sqlsnippets(sstmt1, sql)
sstmt2 = 'and part_date <= $end_date'
sqlstmt += sstmt2
sqlstmt -= sstmt2
sqlstmt = sqlstmt.substitute_params(part_date="'2024-01-01'")
sqlstmt = sqlstmt.get_with_testsql(idx=1)
SqlStatements
from pydbapi.sql import SqlStatements
sql = '''
select * from tablename1 where part_date >= $part_date;
select * from tablename2 where part_date >= $part_date;
'''
sqlstmts = SqlStatements(sql)
属性
方法
sqlstmts = sqlstmts.substitute_params(part_date='2024-01-01')
for stmts in sqlstmts:
stmts
len(sqlstmts)
sqlstmts[0]
sqlstmts[:2]
db.execute(sql, count=None, ehandling=None, verbose=0)count: 返回结果的数量;ehandling: sql执行出错的时候处理方式, default: Noneverbose: 执行的进度展示方式（0：不打印， 1：文字进度， 2：进度条）db.select(tablename, columns, condition=None, verbose=0)tablename: 表名;columns： 列内容;condition: sql where 中的条件db.create(tablename, columns, indexes=None, verbose=0)tablename: 表名;columns： 列内容;indexes: 索引，sqlite暂不支持索引verbose： 是否打印执行进度。db.create(tablename, columns, indexes=None, index_part=128, ismultiple_index=True, partition=None, verbose=0)tablename: 表名;columns： 列内容;indexes: 索引index_part: 索引partismultiple_index: 多重索引partition: 分区verbose： 是否打印执行进度。db.create(tablename, columns, partition=None, verbose=0)tablename: 表名;columns： 列内容;partition: 分区verbose： 是否打印执行进度。db.insert(tablename, columns, inserttype='value', values=None, chunksize=1000, fromtable=None, condition=None)tablename: 表名;columns： 列内容;inserttype: 插入数据类型，支持value、selectvalues: inserttype='value',插入的数值;chunksize: inserttype='value', 每个批次插入的量级;fromtable: inserttype='select',数据来源表;condition: inserttype='select',数据来源条件;db.drop(tablename)tablename: 表名;db.delete(tablename, condition)tablename: 表名;condition: 插入的数值;db.get_columns(tablename)tablename: 表名;db.add_columns(tablename, columns)tablename: 表名;columns： 列内容;db.get_filesqls(filepath, **kw)filepath: sql文件路径;kw： sql文件中需要替换的参数，会替换sqlfile中的arguments;db.file_exec(filepath, ehandling=None, verbose=0, **kw)filepath: sql文件路径; 文件名以test开始或者结尾会打印sql执行的步骤;ehandling: sql执行出错的时候处理方式, default: Noneverbose: 执行的进度展示方式（0：不打印， 1：文字进度， 2：进度条）kw： sql文件中需要替换的参数 在sql文件中用$param, 会替换sqlfile中的arguments;verbose会打印sql执行的步骤;)
#【arguments】#
ts = '2020-06-28'
date = today
date_max = date + timedelta(days=10)
#【arguments】#
###
--【desc1 [verbose]】 #sql描述
--step1
sql1;
--step2
sql2 where name = $name;
###
###
--【desc2 [verbose]】 #sql描述
--step1
sql1;
--step2
sql2;
###
注册方法
命令行中执行pydbapimagic
参数
%dbconfig%dbconfig DBTYPE = 'mysql'
%dbconfig HOST = 'localhost'
%dbconfig USER = 'longfengpili'
%dbconfig PASSWORD = '123456abc'
%dbconfig DATABASE = 'test'
%dbconfig PORT = 3306
%dbconfig DBTYPEFAQs
A simple database API
We found that pydbapi demonstrated a healthy version release cadence and project activity because the last version was released less than a year ago. It has 1 open source maintainer collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Security News
PyPI confirms no security flaws were exploited in the Ultralytics supply chain attack and highlights improvements for safer package publishing.
Security News
Research
The Socket Research Team breaks down a malicious wrapper package that uses obfuscation to harvest credentials and exfiltrate sensitive data.
Research
Security News
Attackers used a malicious npm package typosquatting a popular ESLint plugin to steal sensitive data, execute commands, and exploit developer systems.