pip3 install pydbclib

from pydbclib import connect
# 使用with上下文，可以自動提交，自動關(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ù)有個driver參數(shù)決定你是通過哪個數(shù)據(jù)庫驅(qū)動包去連接的
# driver參數(shù)默認(rèn)值是sqlalchemy，即通過sqlalchemy驅(qū)動包連接數(shù)據(jù)庫
>>> db = pydbclib.connect("sqlite:///:memory:")
>>> db = pydbclib.connect(":memory:", driver='sqlite3')
# 也可以傳入驅(qū)動包連接對象
>>> 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返回迭代器類型，用get方法獲取前幾條記錄，使用map對每條記錄進行數(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對象, 前提已經(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'}
# 也可以直接寫成sql條件表達式，其他接口的條件參數(shù)類似都可以是表達式
>>> 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'}]

# 將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ù)庫驅(qū)動連接，driver參數(shù)指定驅(qū)動包名稱
# 例如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")
# 通過odbc方式連接
db = pydbclib.connect('DSN=mysqldb;UID=user;PWD=password', driver="pyodbc") 
# 通過已有驅(qū)動連接方式連接
import pymysql
con = pymysql.connect(user="user", password="password", database="test")
db = pydbclib.connect(driver=con)

# 使用Sqlalchemy包來連接數(shù)據(jù)庫，drvier參數(shù)默認(rèn)為'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)