SQLAlchemy的两种操作数据库的方式

本是古典何须时尚 2022-05-14 02:23 230阅读 0赞

工作需要, 用到了SQLAlchemy, 经过了差不多一个月的学习和探索, 整理如下:

SQLAlchemy有两种操作数据库的方式, 一种是session-mapper的方式, 统称为ORM, 另外一种就是有点像原生的sql语句, 称之为core方式.

首先来看第一种方式:  
ORM(session-mapper)

from sqlalchemy import create_engine
    from sqlalchemy.ext.declarative import declarative_base
    from sqlalchemy.dialects.mysql import INTEGER, DATETIME, REAL, TEXT, VARCHAR
    from sqlalchemy import Table, Column, ForeignKey, MetaData
    from sqlalchemy.orm import mapper, sessionmaker
    from sqlalchemy import  func
    import re
    import hashlib
    import os, sys
    import time, datetime
    
    
    class SelectData():
        def __init__(self):
            s1     = "mysql+mysqldb://"
            server = "root:123456@localhost:3306/test"
            s2     = "?use_unicode=0&charset=utf8"
            connectserver = s1 + server + s2
            self.engine = create_engine(connectserver, encoding = 'utf-8', echo = False)
            self.Base = MetaData()
    
            self.user= Table(
                'user', self.Base,
                 Column('id', INTEGER, primary_key = True),
                 Column('name', TEXT)
                 Column('passwd', TEXT)
                 )
            self.client = Table(
                'client', self.Base,
                 Column('id', INTEGER, primary_key = True),
                 Column('name', TEXT),
                 Column('id_user', INTEGER),
                 Column('reserved_int', INTEGER),
                 Column('reserved_varchar', VARCHAR(200))
                 )
        self.Base.create_all(self.engine)#create_all可以自动检测表是否已经建立，非常方便
            self.Session = sessionmaker() #创建了一个自定义了的 Session类
            self.Session.configure(bind = self.engine)  #将创建的数据库连接关联到这个session
    def insert_function(self):
        session = self.Session()
        class User(object): pass
        mapper(User, self.user)
        class Client(object): pass
        mapper(Client, self.client)
    
        #select方式：
        ret = session.query(User).filter(User.name == "admin").first()
        if ret is None:
            #添加新的字段
            user = User()
            user.name = "admin"
            user.passwd = CountMd5("123456")#此处CountMd5()是自己写的一个计算md5值的函数
            session.add(user)
            session.flush()
            #session.commit()
            #获取刚刚插入数据的自增主键id
            id_usr = user.id
        else:
            id_usr = ret.id
    
        ret2 = session.query(Client).filter_by(id_user = id_usr).first()
        if ret2 is None:
            client = Client()
            client.name = "Chocolate"
            client.id_user = id_usr
            client.session.add(client)
            session.flush()
            session.commit()
        else:
            #使用update:
            session.query(Client).filter_by(id_user = id_usr).update({
       "name": "Chocolate", "id_user":id_usr})   
            session.flush()
            session.commit()
            session.close() 
    #多表联合查询
    def testfun2(self, partname, pageindex):
        session = self.Session()
        class User(object): pass
        mapper(User, self.user)
        class Client(object): pass
        mapper(Client, self.client)
    
        pageval = (pageindex - 1) * 10
        #这里面一个是like的用法，一个是只显示某10条
        ret = self.session.query(User)\
                .join(Client, Client.id_user == User.id )\
                .filter(User.name.like('%' + partname + '%'))\
                .offset().limit(10).all()
        if ret is None:
            return None
        else:
            for k in ret:
                print k.name, k.passwd
        session.close()
     #删除表的全部内容，不删除表
    def delete_func(self):
        session = self.Session()
        class User(object): pass
        mapper(User, self.user)
        class Client(object): pass
        mapper(Client, self.client)
        session.query(Client).delete()
        session.query(User).delete()
        session.commit()
        session.close()

上面就是orm的方式了，这个方式要特别注意的地方就是，session的问题，因为session是有个有效期的，如果我们把session在init函数里面做成全局变量self.session,那么就会出现数据库实时更新插入的值这里面查询不到，所以，对于session的态度，我们是用到再新建，用完手动关闭。

Core方式

def sort_all_rule_skus(self, session, rule_id, sku_nos):
         data =[{
       '_order': i, '_sku_no': sku_no} for i, sku_no in enumerate(sku_nos)]
         smt = UnionRuleSku.__table__.update().where(
             and_(
                 UnionRuleSku.rule_id == rule_id,
                 UnionRuleSku.sku_no == bindparam('_sku_no'),    # 绑定参数 _sku_no
             )
         ).values({
             UnionRuleSku.order: bindparam('_order')             # 绑定参数 _order
         })
         session.execute(smt, data)              # 群体插入