之前研究了一段時間的Flask促王,覺得使用ORM操作數據庫真滴爽,于是便去翻了翻SQLAlchemy模塊的文檔唐片。一開始還想著是不是要從table類開始寫污朽,看了一眼數據庫里的一百多張表拦耐,頓時感覺壓力山大......
好在SQLAlchemy可以使用命令生成表類,如下:
# 依賴sqlacodegen模塊
# pip install sqlacodegen -i http://pypi.douban.com/simple --trusted-host pypi.douban.com
# 使用前可以先sqlacodegen –help看一下具體用法
# 一句話生成models.py文件
# sqlacodegen mysql+oursql://user:password@localhost/dbname
# 數據表反向生成sqlalchemy ORM:
# sqlacodegen --tables tablename(表名) --outfile ./filename.py(輸出的文件名) mysql+pymysql://user:password@host/dbname?charset=yourencode
# windows 環(huán)境下cmd窗口 執(zhí)行 即可.(注意輸出的文件)
但這樣操作仍然有些不方便孔轴,尤其是在數據表經常發(fā)生變化的時候,每次都要去同步數據表結構碎捺,感覺就很麻煩路鹰。
雖然數據庫里有一百多張表,但筆者經常用的表也就十多個收厨,總有一種殺雞焉用牛刀的感覺晋柱,個人也還是更喜歡pymysql那樣的連接方式。
好在SQLAlchemy也提供了反射ORM的方法--automap_base诵叁。
from sqlalchemy.ext.automap import automap_base
from sqlalchemy.orm import Session
from sqlalchemy import create_engine
# 連接url
client_url = f'mysql+pymysql://user:password@host:port/dbname?charset=yourencode'
# 連接引擎
engine = create_engine(client_url, pool_size=10)
# 生成所有的映射關系為Base
Base = automap_base()
# 設置被映射的類和關系并執(zhí)行映射
Base.prepare(engine, reflect=True)
# 獲取會話連接
session = Session(bind=engine)
# 獲取表對象
table = Base.classes.table_name
接下來就可以用ORM方式操作數據庫了雁竞。
# select 全表查詢
datas = session.query(table).all()
# insert 插入記錄
user = table(username='xxx', password='xxx')
session.add(user)
session.commit()
# update 更新記錄
session.query(table).filter_by(username='xxx').update({'password':'xxxxxx'})
session.commit()
# delete 刪除記錄
session.query(table).filter_by(username='xxx').delete()
session.commit()
還可以根據自己的需求進一步封裝api。
首先拧额,先將上述操作寫入到類中碑诉。
class SqlAlchemyScript:
def __init__(self, host='xx.xx.xx.xx', user='xxxx', passwd='xxxxxx', port=3306, db='xx', charset='UTF8MB4'):
self.host = host
self.user = user
self.passwd = passwd
self.port = port
self.db = db
self.charset = charset
# 數據庫連接url
self.client_url = f'mysql+pymysql://{self.user}:{self.passwd}@{self.host}:{self.port}/{self.db}?charset={self.charset}&autocommit=true'
# 數據庫連接引擎
self.engine = create_engine(self.client_url, pool_size=5)
# 生成所有的映射關系為Base
self.Base = automap_base()
# 設置被映射的類和關系并執(zhí)行映射
self.Base.prepare(self.engine, reflect=True)
# 獲取會話連接
self.session = Session(bind=self.engine)
在這個類中定義2個獲取表對象的方法。get_table方法獲取單獨的表侥锦,get_tables方法獲取列表中的所有表并返回1個包含所有表對象的列表联贩。
def get_table(self, table_name):
"""
獲取表對象
:param table_name: 表名
:return: table
"""
if isinstance(table_name, str):
strs = f'self.Base.classes.{table_name}'
# 反射得到ORM
table = eval(strs)
return table
else:
raise TypeError('傳入的類型必須是str!!!')
def get_tables(self, tables):
"""
獲取表對象
:param tables: 存儲表名的列表或元組
:return: list
"""
lst = list()
if isinstance(tables, list) or isinstance(tables, tuple):
for t in tables:
table_obj = self.get_table(t)
lst.append(table_obj)
else:
raise TypeError('傳入的類型必須是list or tuple!!!')
return lst
接下來可以封裝任意你想使用的功能。
比如:定義1個簡單插入方法捎拯。
def simple_insert(self, table, data_dict):
"""
插入記錄
:param table: 表對象
:param data_dict: 插入的數據(dcit)
:return: 新插入的實例
"""
result = table(**data_dict)
self.session.add(result)
self.session.commit()
print(f'{data_dict} 已插入表@峄稀Cぱ帷!')
return result
使用方法如下:
if __name__ == '__main__':
test = SqlAlchemyScript()
table = test.get_table('table_name')
result = test.simple_insert(table, {'username': 'xxx', 'password': 'xxx'})
print(f'新插入了1條記錄祸泪,該記錄的id為:{result.id}')