python連接數據庫使用SQLAlchemy

參考python核心編程

ORM（Object Relational Mapper），如果你是一個更願意操作Python對象而不是SQL查詢的程序員，並且仍然希望使用關系型數據庫作為你的后端，那么你可能更傾向於使用ORM。

ORM將純SQL語句進行了抽象畫處理，將其實現為python中的對象，這樣你只操作這些對象就像完成於SQL語句相同的任務——數據庫中的表被轉化為Python類，其中的數據列作為屬性，而數據庫操作則會作為方法。一般來說，ORM為了代替你執行很多工作，會把一些事情變得復雜，或者比直接使用適配器更多的代碼。但是，這點額外工作可以獲得更高的生產效率。

目前，最知名的Python ORM是SQLAlchemy和SQLobject。而后者目前只支持python2的版本。

本節，我們探討SQLAlchemy的使用。SQLALchemy幫助文檔

SQLAlchemy不在python的標准庫中需要我們手動安裝，這里采用pip的安裝方法：

PS C:\Users\WC> pip install sqlalchemy
Collecting sqlalchemy
  Downloading https://files.pythonhosted.org/packages/c1/c8/392fcd2d01534bc871c65cb964e0b39d59f
/SQLAlchemy-1.2.7.tar.gz (5.6MB)
    100% |████████████████████████████████| 5.6MB 299kB/s
Installing collected packages: sqlalchemy
  Running setup.py install for sqlalchemy ... done
Successfully installed sqlalchemy-1.2.7

 下面的腳本簡單實現了一個SQLAlchemy的連接到Mysql和sql server 的用戶洗牌應用。運行該腳本時請注意我們使用到上一節編寫的腳本！！！

#本腳本兼容了python2.x和3.x版本，使用SQLAlchemy ORM連接sqlserver或者MySQL，實現一個用戶洗牌的應用

from os.path import dirname
import pymysql
from pymysql.err import InternalError
from random import randrange
from distutils.log import warn as printf
from sqlalchemy import Column,Integer,String,create_engine,exc,orm
from sqlalchemy.ext.declarative import declarative_base
#導入本地應用模塊（ushuffle_dbU）必要的常量和工具函數，這是為了避免到處復制、黏貼相同的代碼
from ushuffle_dbU import DBNAME,NAMELEN,randName,FIELDS,tformat,cformat,setup

#dialect+driver://username:password@host:port/database
DSNs = {
    'mysql':'mysql+pymysql://root:Jwxjs123456@localhost/%s' % DBNAME,
    'sqllite':'sqllite:///:memory:',
    'sql server':'mssql+pymssql://sa:Jwxjs123456@localhost/%s' % DBNAME,
}

#SQLALchemy的聲明層。
base = declarative_base()
class Users(base):#繼承Base類
    __tablename__ = 'users' #映射的數據庫表名
    login = Column(String(NAMELEN))
    userid = Column(Integer,primary_key = True)
    projid = Column(Integer)
    def __str__(self): #返回易於閱讀的數據行的字符串格式
        return ''.join(map(tformat,(self.login,self.userid,self.projid)))
    
class SQLALchemyTest(object):
    #初始化函數盡可能的得到一個可用的數據庫，保存其連接
    def __init__(self,dsn):
        try:
            eng = create_engine(dsn)#,echo = True)#嘗試使用dsn創建數據庫引擎,echo設置為True可使得我們能看到ORM生成的SQL語句
            printf('*****創建數據庫引擎成功')
        except ImportError:
            raise RuntimeError()#創建失敗。一般來說，引擎創建失敗意味着SQLAlchemy不支持所選的數據庫，會拋出ImportError
        try :
            eng.connect()#如果數據庫引擎創建成功，則嘗試創建數據庫連接
            printf('*****連接數據庫%s成功' % DBNAME)
        except (exc.InternalError,InternalError):#如果創建連接失敗，一般意味着數據庫本身不可達。此例中是因為目標數據庫不存在
            eng = create_engine( dirname(dsn),echo = True)#dirname返回dsn的目錄。其實就是os.path.split(path)的第一個元素。效果相當於dsn去掉數據庫 
            eng.execute('create database %s' % DBNAME).close()
            printf('新建數據庫%s成功' % DBNAME)
            eng.create_engine(dsn,echo = True)
            printf('連接數據庫%s成功' % DBNAME)
        #創建數據庫引擎連接之后，需要創建一個會話對象Session，其實是一個factory。
        #sessionmaker()也可以在create_engine之前創建，然后等到數據庫引擎連接創建之后，調用Session.configure(bind=)實現綁定
        Session = orm.sessionmaker(bind=eng)
        #Session()可以創建一個綁定到數據庫的對象。但是到此為止，它還沒有打開任何的連接。
        #當它第一次被調用的時候，會嘗試從數據庫引擎連接池中檢索一個鏈接，該連接會一直被持有直到所有的任務都被提交或者Session對象被關閉
        self.ses = Session()
        #將用戶的表和數據庫引擎作為 實例的屬性保存下來。意味着這張表的所有操作都會被綁定到這個指定的引擎中。
        self.users = Users.__table__
        self.eng = self.users.metadata.bind = eng#引擎和表的元數據進行了額外的綁定，意味着這張表的所有操作都會綁定到這個指定的引擎中
    def insert(self):
        self.ses.add_all(Users(login= who,userid=userid,projid=randrange(1,5)) for who,userid in randName())
        self.ses.commit()
    
    def update(self):
        fr = randrange(1,5)
        to = randrange(1,5)
        i = -1
        users = self.ses.query(Users).filter_by(projid = fr).all()
        for i ,user in enumerate(users):
            user.projid = to
        self.ses.commit()
        printf('\n %s 個用戶名從 %s 更改到 %s' % (i+1,fr,to))
    
    def delete(self):
        rm = randrange(1,5)
        i = -1
        users = self.ses.query(Users).filter_by(projid = rm).all()
        for i,user in enumerate(users):
            self.ses.delete(user)
        self.ses.commit()
        printf('組號為%s的共%s個用戶被刪除' % (rm,i+1))
    
    def finish(self):
        self.ses.connection().close()
    
    def dbDump(self):
        printf('\n%s' % ''.join(map(cformat,FIELDS)))
        users = self.ses.query(Users).all()
        for user in users:
            printf(user)
        self.ses.commit()
    
    def __getattr__(self,attr):#drop和create方法實際上只需要調用表的drop()和create()方法即可，不用我們自己單獨編寫。這里使用委托機制——當屬性查找失敗的時候會調用__getattr__()方法。
        return getattr(self.users,attr)#使用help(getattr)來獲取幫助。getattr()獲取對象的屬性，getattr(x,'y')相當於執行x.y。
    
def main():
    printf('連接%s數據庫' % DBNAME)
    db = setup()
    
    if db not in DSNs:
        printf('\n ERROR:%s 數據庫不被支持,程序退出')
        return
        
    try:
        orm = SQLALchemyTest(DSNs[db])
    except RuntimeError:
        printf('\nERROR: %s 數據庫不支持，程序退出')
    
    printf('\n***創建users表（）')
    orm.drop(checkfirst = True)#
    orm.create()
    printf('\n***創建users表,成功！！！')
    
    printf('%%%插入數據')
    orm.insert()
    orm.dbDump()
    
    printf('\n隨機將用戶進行分組')
    orm.update()
    orm.dbDump()
    
    printf('\n ！！！隨機刪除用戶')
    orm.delete()
    orm.dbDump()
    
    printf('\n***刪除表格')
    orm.drop()
    
    printf('\n關閉數據庫連接!')
    orm.finish()
    
if __name__ == '__main__':
    main()
    

 如果，你在運行過程中出現了警告（並不影響程序運行結果），可以做如下設置：

import warnings
warnings.filterwarnings("ignore")#忽略警告