使用SQLAlchemy操作資料庫表過程解析
阿新 • • 發佈:2020-06-11
需求場景:
使用sqlalchmy從現有的表中獲取資料(不是自己建表)。百度了一下,網上都是使用sqlalchemy自己先建立表,然後匯入資料表的模型類進行增刪改查;現在不是自己建表,該如何操作呢?
操作方案
通過sqlalchmey執行原生的sql語句,增刪改查的原生語句攜帶表名,就不需要匯入資料表的模型類了。
使用的包:
SQLAlchemy (1.3.10) + mysql-connector-python (8.0.19)
提供以下乾貨:
- 演示了向原生sql語句傳遞變數的用法 即動態執行sql語句 更加靈活
- 通過執行原生的sql語句實現操作已有的表
- 演示了sql語句根據多欄位排序的方法等
DEMO
# -*- coding:utf-8 -*- from sqlalchemy import create_engine,MetaData,Table,exists from sqlalchemy.orm import sessionmaker,scoped_session from util.Log import Log from conf.parseConfig import parseConf # 從配置檔案中獲取mysql的配置資訊 host = parseConf.get_conf('MySQLInfo','host') port = parseConf.get_conf('MySQLInfo','port') dbname = parseConf.get_conf('MySQLInfo','dbname') usernm = parseConf.get_conf('MySQLInfo','usernm') passwd = parseConf.get_conf('MySQLInfo','passwd') engine_str = "mysql+mysqlconnector://{0}:{1}@{2}:{3}/{4}".format(usernm,passwd,host,port,dbname) class OpsMysql(object): def __init__(self,log=Log(__file__).getlog()): self.log = log self.session = None try: self.engine = create_engine( engine_str,max_overflow=0,# 超過連線池大小外最多建立的連線 pool_size=5,# 連線池大小 pool_timeout=30,# 池中沒有執行緒最多等待的時間,否則報錯 pool_recycle=-1,# 多久之後對執行緒池中的執行緒進行一次連線的回收(重置) # echo=True,# 顯示相應執行的 sql 指令 encoding='utf-8' ) SessionFactory = sessionmaker(bind=self.engine) self.session = scoped_session(SessionFactory) except Exception as e: self.log.error(str(e)) self.log.error("Connect {0}@{1}:{2} failed!".format(dbname,port)) def get_session(self): return self.session def getEngine(self): return self.engine def init_db(self,base): base.metadata.create_all(self.engine) def drop_db(self,base): base.metadata.drop_all(self.engine) if __name__ == "__main__": log = Log(__file__).getlog() tt = OpsMysql(log) session = tt.get_session() if session: # 通過執行原生的sql語句實現操作已有的表 # 此處演示了向原生sql語句傳遞變數的用法 即動態執行sql語句 更加靈活 mail_id = 1 res = session.execute('select * from tbl_mail_addr where mail_id='" + mail_id + "' and mail_tp="c"') res01 = res.fetchall() # 結果是列表 print(res01[0]) # (1,'c',1,'[email protected]') mail_id = '1' mail_type = 'c' # 查詢id為1,型別是c的郵箱資訊,並按mail_tp降序,addr_no升序排列,限制查詢數量100 sql = "select * from tbl_mail_addr where tbl_mail_addr.oper_flag='1' and tbl_mail_addr.mail_id='" + mail_id + "' and tbl_mail_addr.mail_tp='" + mail_type + "' order by tbl_mail_addr.mail_tp,tbl_mail_addr.addr_no ASC limit 100" result = session.execute(sql) value_list = result.fetchall() print(value_list) # [(1,'[email protected]'),(1,2,'[email protected]),3,'[email protected]')] session.close()
以上就是本文的全部內容,希望對大家的學習有所幫助,也希望大家多多支援我們。