python连接数据库使用SQLAlchemy

Posted 2020-11-17 冲扬心法

参考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:UsersWC> 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:[email protected]:port/database
DSNs = {
    ‘mysql‘:‘mysql+pymysql://root:[email protected]/%s‘ % DBNAME,
    ‘sqllite‘:‘sqllite:///:memory:‘,
    ‘sql server‘:‘mssql+pymssql://sa:[email protected]/%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(‘
 %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(‘
%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(‘
 ERROR:%s 数据库不被支持,程序退出‘)
        return
        
    try:
        orm = SQLALchemyTest(DSNs[db])
    except RuntimeError:
        printf(‘
ERROR: %s 数据库不支持，程序退出‘)
    
    printf(‘
***创建users表（）‘)
    orm.drop(checkfirst = True)#
    orm.create()
    printf(‘
***创建users表,成功！！！‘)
    
    printf(‘%%%插入数据‘)
    orm.insert()
    orm.dbDump()
    
    printf(‘
随机将用户进行分组‘)
    orm.update()
    orm.dbDump()
    
    printf(‘
 ！！！随机删除用户‘)
    orm.delete()
    orm.dbDump()
    
    printf(‘
***删除表格‘)
    orm.drop()
    
    printf(‘
关闭数据库连接!‘)
    orm.finish()
    
if __name__ == ‘__main__‘:
    main()
    

 如果，你在运行过程中出现了警告（并不影响程序运行结果），可以做如下设置：

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