pydbclib是一個通用的python關係型資料庫操作工具包，使用統一的介面操作各種關係型資料庫（如 oracle、mysql、postgres、hive、impala等）進行增刪改查，它是對各個python資料庫連接驅動包（如sqlalchemy、pymysql、cx_Oracle、pyhive ...
pydbclib是一個通用的python關係型資料庫操作工具包，使用統一的介面操作各種關係型資料庫（如 oracle、mysql、postgres、hive、impala等）進行增刪改查，它是對各個python資料庫連接驅動包（如sqlalchemy、pymysql、cx_Oracle、pyhive、pyodbc、impala等）的封裝，依照python最簡原則SQL占位符統一成 ':[name]' 這一種形式，這點和sqlalchemy是一樣的
安裝
pip3 install pydbclib
簡單使用
看下簡單的查詢示例
from pydbclib import connect # 使用with上下文，可以自動提交，自動關閉連接 with connect("sqlite:///:memory:") as db: db.execute('create table foo(a integer, b varchar(20))') # 統一使用’:[name]'形式的SQL的占位符 db.execute("insert into foo(a,b) values(:a,:b)", [{"a": 1, "b": "one"}]*4) print(db.read("select * from foo").get_one()) print(db.read("select * from foo").get_all()) print(db.read("select * from foo").to_df()) db.get_table("foo").insert({"a": 2, "b": "two"}) print(db.get_table("foo").find_one({"a": 2})) print(db.get_table("foo").find().get_all()) print(db.get_table("foo").find().to_df())
查詢結果記錄是以字典形式展現，向庫里寫入記錄也是字典形式，如果要使用原生元祖形式，查詢函數read里添加as_dict=False參數
介面文檔
資料庫連接，更多常用資料庫連接方式參考文章結尾
# connect函數有個driver參數決定你是通過哪個資料庫驅動包去連接的 # driver參數預設值是sqlalchemy，即通過sqlalchemy驅動包連接資料庫 >>> db = pydbclib.connect("sqlite:///:memory:") >>> db = pydbclib.connect(":memory:", driver='sqlite3') # 也可以傳入驅動包連接對象 >>> import sqlite3 >>> db = pydbclib.connect(driver=sqlite3.connect(":memory:")) >>> from sqlalchemy import create_engine >>> db = pydbclib.connect(driver=create_engine("sqlite:///:memory:"))
原生SQL介面
1. 使用execute方法執行SQL，和各資料庫連接包基本一致，不同點是它既可以單條執行，也可以批量執行(相當於executemany)，另外該方法的SQL占位符是':[name]'形式
>>> record = {"a": 1, "b": "one"}
>>> db.execute('create table foo(a integer, b varchar(20))')
# 插入單條記錄，結果返回影響行數
>>> db.execute("insert into foo(a,b) values(:a,:b)", record)
1
# 插入多條記錄
>>> db.execute("insert into foo(a,b) values(:a,:b)", [record, record])
2
2. 查詢數據
# 查詢結果只返回一條記錄 >>> db.read_one("select * from foo") {'a': 1, 'b': 'one'} #read返回迭代器類型，用get方法獲取前幾條記錄，使用map對每條記錄進行數據清洗 >>> db.read("select * from foo").map(lambda x: {f"foo.{k}": v for k,v in x.items()}).get(2) # as_dict=False返回原生元祖記錄 >>> db.read("select * from foo", as_dict=False).get(2) [(1, 'one'), (1, 'one')] # 也可以直接for遍歷 >>> for r in db.read("select * from foo"): ... print(r) ... {'a': 1, 'b': 'one'} {'a': 1, 'b': 'one'} {'a': 1, 'b': 'one'} # 轉換成pandas dataframe對象, 前提已經安裝了pandas >>> db.read("select * from foo").to_df() a b 0 1 one 1 1 one 2 1 one
3. 提交、回滾、關閉連接
>>> db.rollback() >>> db.commit() >>> db.close()
表級別操作的SQL介面封裝
1. 插入記錄
# 插入單條和插入多條，輸入參數字典的鍵值必須和表中欄位同名 >>> db.get_table("foo").insert({"a": 1, "b": "one"}) 1 >>> db.get_table("foo").insert([{"a": 1, "b": "one"}]*10) 10
2. 查詢記錄
# 查詢欄位a=1第一條記錄 >>> db.get_table("foo").find_one({"a": 1}) {'a': 1, 'b': 'one'} # 也可以直接寫成sql條件表達式，其他介面的條件參數類似都可以是表達式 >>> db.get_table("foo").find_one("a=1") {'a': 1, 'b': 'one'} # 查詢欄位a=1所有記錄，find返回迭代器對象同上面read方法 >>> db.get_table("foo").find({"a": 1}).get_all() [{'a': 1, 'b': 'one'},...{'a': 1, 'b': 'one'}]
3. 更新記錄
# 將a=1那條記錄的b欄位值更新為"first" >>> db.get_table("foo").update({"a": 1}, {"b": "first"}) 11 >>> db.get_table("foo").find({"a": 1}).get_one() {'a': 1, 'b': 'first'}
4. 刪除記錄
# 將a=1那條記錄刪除 >>> db.get_table("foo").delete({"a": 1}) 11 >>> db.get_table("foo").find({"a": 1}).get_all() []
常用資料庫連接
1. Common Driver
# 使用普通資料庫驅動連接，driver參數指定驅動包名稱 # 例如pymysql包driver='pymysql',connect函數其餘的參數和driver參數指定的包的創建連接參數一致 # 連接mysql db = pydbclib.connect(user="user", password="password", database="test", driver="pymysql") # 連接oracle db = pydbclib.connect('user/password@local:1521/xe', driver="cx_Oracle") # 通過odbc方式連接 db = pydbclib.connect('DSN=mysqldb;UID=user;PWD=password', driver="pyodbc") # 通過已有驅動連接方式連接 import pymysql con = pymysql.connect(user="user", password="password", database="test") db = pydbclib.connect(driver=con)
2. Sqlalchemy Driver
# 使用Sqlalchemy包來連接資料庫，drvier參數預設為'sqlalchemy' # 連接oracle db = pydbclib.connect("oracle://user:password@local:1521/xe") # 連接mysql db = pydbclib.connect("mysql+pyodbc://:@mysqldb") # 通過已有engine連接 from sqlalchemy import create_engine engine = create_engine("mysql+pymysql://user:password@localhost:3306/test") db = pydbclib.connect(driver=engine)
使用過程中有任何疑問，歡迎評論交流
項目地址pydbclib
