pip3 install pydbclib

from pydbclib import connect
# 使用with上下文，可以自動(dòng)提交，自動(dòng)關(guān)閉連接
with connect("sqlite:///:memory:") as db:
  db.execute('create table foo(a integer, b varchar(20))')
  # 統(tǒng)一使用':[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())

# connect函數(shù)有個(gè)driver參數(shù)決定你是通過(guò)哪個(gè)數(shù)據(jù)庫(kù)驅(qū)動(dòng)包去連接的
# driver參數(shù)默認(rèn)值是sqlalchemy，即通過(guò)sqlalchemy驅(qū)動(dòng)包連接數(shù)據(jù)庫(kù)
>>> db = pydbclib.connect("sqlite:///:memory:")
>>> db = pydbclib.connect(":memory:", driver='sqlite3')
# 也可以傳入驅(qū)動(dòng)包連接對(duì)象
>>> import sqlite3
>>> db = pydbclib.connect(driver=sqlite3.connect(":memory:"))
>>> from sqlalchemy import create_engine
>>> db = pydbclib.connect(driver=create_engine("sqlite:///:memory:"))

>>> record = {"a": 1, "b": "one"}
>>> db.execute('create table foo(a integer, b varchar(20))')
# 插入單條記錄，結(jié)果返回影響行數(shù)
>>> 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

# 查詢結(jié)果只返回一條記錄
>>> db.read_one("select * from foo")
{'a': 1, 'b': 'one'}
#read返回迭代器類(lèi)型，用get方法獲取前幾條記錄，使用map對(duì)每條記錄進(jìn)行數(shù)據(jù)清洗
>>> 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'}
# 轉(zhuǎn)換成pandas dataframe對(duì)象, 前提已經(jīng)安裝了pandas
>>> db.read("select * from foo").to_df()
  a  b
0 1 one
1 1 one
2 1 one

>>> db.rollback()
>>> db.commit()
>>> db.close()

# 插入單條和插入多條，輸入?yún)?shù)字典的鍵值必須和表中字段同名
>>> db.get_table("foo").insert({"a": 1, "b": "one"})
1
>>> db.get_table("foo").insert([{"a": 1, "b": "one"}]*10)
10

# 查詢字段a=1第一條記錄
>>> db.get_table("foo").find_one({"a": 1})
{'a': 1, 'b': 'one'}
# 也可以直接寫(xiě)成sql條件表達(dá)式，其他接口的條件參數(shù)類(lèi)似都可以是表達(dá)式
>>> db.get_table("foo").find_one("a=1")
{'a': 1, 'b': 'one'}
# 查詢字段a=1所有記錄，find返回迭代器對(duì)象同上面read方法
>>> db.get_table("foo").find({"a": 1}).get_all()
[{'a': 1, 'b': 'one'},...{'a': 1, 'b': 'one'}]

# 將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'}

# 將a=1那條記錄刪除
>>> db.get_table("foo").delete({"a": 1})
11
>>> db.get_table("foo").find({"a": 1}).get_all()
[]

# 使用普通數(shù)據(jù)庫(kù)驅(qū)動(dòng)連接，driver參數(shù)指定驅(qū)動(dòng)包名稱(chēng)
# 例如pymysql包driver='pymysql',connect函數(shù)其余的參數(shù)和driver參數(shù)指定的包的創(chuàng)建連接參數(shù)一致
# 連接mysql
db = pydbclib.connect(user="user", password="password", database="test", driver="pymysql")
# 連接oracle
db = pydbclib.connect('user/password@local:1521/xe', driver="cx_Oracle")
# 通過(guò)odbc方式連接
db = pydbclib.connect('DSN=mysqldb;UID=user;PWD=password', driver="pyodbc") 
# 通過(guò)已有驅(qū)動(dòng)連接方式連接
import pymysql
con = pymysql.connect(user="user", password="password", database="test")
db = pydbclib.connect(driver=con)

# 使用Sqlalchemy包來(lái)連接數(shù)據(jù)庫(kù)，drvier參數(shù)默認(rèn)為'sqlalchemy'
# 連接oracle
db = pydbclib.connect("oracle://user:password@local:1521/xe")
# 連接mysql
db = pydbclib.connect("mysql+pyodbc://:@mysqldb")
# 通過(guò)已有engine連接
from sqlalchemy import create_engine
engine = create_engine("mysql+pymysql://user:password@localhost:3306/test")
db = pydbclib.connect(driver=engine)