一个简单的数据库 API，支持多种数据库类型，包括 SQLite、Amazon Redshift、MySQL 和 Trino。
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 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
+ 代码
`col = ColumnModel(newname, coltype='varchar', sqlexpr=None, func=None, order=0)`
+ 参数
* `newname`: 新名称;
* `coltype`: 类型
* `sqlexpr`: SQL 表达式
* `func`: 查询函数，当前支持 'min'、'max'、'sum'、'count'
* `order`: 排序
+ 代码
`cols = ColumnsModel(ColumnModel, ColumnModel, ……)`
+ 属性
* `func_cols`: 返回列的列表
* `nonfunc_cols`: 返回列的列表
* `new_cols`: 返回拼接字符串
* `create_cols`: 返回拼接字符串
* `select_cols`: 返回拼接字符串
* `group_cols`: 返回拼接字符串
* `order_cols`: 返回拼接字符串
+ 方法
* get_column_by_name
- `cols.get_column_by_name(name)`
- 返回 `ColumnModel`
from 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 执行出错时的处理方式，默认: 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: 索引部分ismultiple_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 文件中需要替换的参数，会替换 SQL 文件中的 arguments;db.file_exec(filepath, ehandling=None, verbose=0, **kw)filepath: SQL 文件路径; 文件名以test开始或者结尾会打印 SQL 执行的步骤;ehandling: SQL 执行出错时的处理方式，默认: Noneverbose: 执行的进度展示方式（0：不打印， 1：文字进度， 2：进度条）kw： SQL 文件中需要替换的参数，在 SQL 文件中用$param，会替换 SQL 文件中的 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
Meet Socket at Black Hat & DEF CON 2025 for 1:1s, insider security talks at Allegiant Stadium, and a private dinner with top minds in software supply chain security.
Security News
CAI is a new open source AI framework that automates penetration testing tasks like scanning and exploitation up to 3,600× faster than humans.
Security News
Deno 2.4 brings back bundling, improves dependency updates and telemetry, and makes the runtime more practical for real-world JavaScript projects.