【重磅推荐】SQLAlchemy 中的 Session、sessionmaker、scoped_session及多线程之间的用法【使用SQLAlchemy和flask-SQLAlchemy的区别】

朴灿烈づ我的快乐病毒、 2021-08-26 18:15 622阅读 0赞

参考别人的文档：[https://www.cnblogs.com/ChangAn223/p/11277468.html][https_www.cnblogs.com_ChangAn223_p_11277468.html]

参考Session的创建过程：[https://www.cnblogs.com/ybjourney/p/11876595.html][https_www.cnblogs.com_ybjourney_p_11876595.html]

### **因为Session创建的问题会导致以下常见错误：** ###

1、(mysql.connector.errors.OperationalError) MySQL Connection not available【其实就是session已经失效,而无法新创建】

### **分享一个本人自己封装的Python-SQLAlchemy【基于单实例Session,和Flask的封装类似】** ###

from sqlalchemy import create_engine
    from sqlalchemy.sql.schema import Column
    from sqlalchemy import ForeignKey
    from sqlalchemy.orm import relationship
    from sqlalchemy.sql.sqltypes import String, Integer, Text, DateTime, Float, Boolean
    from sqlalchemy.orm import sessionmaker, scoped_session
    from sqlalchemy import orm
    from sqlalchemy.ext.declarative import declarative_base
    from app.libs.mysql.own_column_type import OwnUnsignedDecimal
    from app.config.global_secure import GlobalSecure
    from app.libs.mysql.own_base_model import create_own_base_model
    
    
    class OwnQuery(orm.query.Query):
        # 重写父类原生SQLalchemy.orm.query.Query中的filter_by方法！！！
        # 数据库Base模型中,status=1,表面这条数据是真实有效的(非软删除)
        def filter_by(self, **kwargs):
            if "status" not in kwargs.keys():
                kwargs.update(status=1)
            return super().filter_by(**kwargs)
    
    
    class OwnSQLAlchemy:
        def __init__(self, query_class=None, **kwargs):
            # 初始化数据库连接:
            # uri格式为"mysql+mysqlconnector://用户名:密码@localhost:3306/数据库名称"
            self.engine = create_engine(GlobalSecure.get_sqlalchemy_database_mysql_uri())
            # 创建Model实例的基类:
            self.Model = declarative_base()
            """
                1、这里的self.session是全局共享的对象,由于对象内部实现了__call__方法,支持session()创建新的实例【单线程必须使用单例模式】
                2、为了兼容和Flask同步封装,这里和Flask一样都使用小写的session,这里的实现和flask是一样的!
            """
            self.session = self.create_scoped_session(query_class=query_class, **kwargs)
            self.Column = Column
            self.ForeignKey = ForeignKey
            self.relationship = relationship
            # 常用数据类型快捷键
            self.String = String
            self.Integer = Integer
            self.Decimal = OwnUnsignedDecimal
            self.Text = Text
            self.DateTime = DateTime
            self.Float = Float
            self.Integer = Integer
            self.Boolean = Boolean
    
        def create_create_all_table(self):
            """
                功能：创建所有的mysql表Table【在所有表创建完成之后,第一次必须执行此方法来创建所有的表】
                **非常非常重要**：每个Model第一次初始化必须调用此方法创建Table,否则会报错【Table '数据库名称.表名称' doesn't exist】
                最佳位置：把所有的Model放入一个__init__.py文件初始化,统一调用一次此方法！
            """
            self.Model.metadata.create_all(bind=self.engine)
    
        def create_session(self, query_class=None, **kwargs):
            """
                功能：通过sessionmaker来创建session,此session创建出的实例是可变的【称之为session工厂】
                特殊情况：由于sessionmaker内部实现了__call__方法【兼容对象支持函数的调用方式】session()等价于session.__call__(*args,**kwargs)
                         这里的session变量是全局共享的,仅执行session()才会创建一个新的session实例【sessionmaker每次创建的session()实例都是不一样的】
                关于重写：如何重写sqlalchemy内部的结构,比如Query类
                    query_cls参数默认值为【sqlalchemy.orm.query.Query】具体参数请参考sessionmaker方法的官方文档
                    如果要修改query_cls的值,可以继承sqlalchemy.orm.query.Query重写内部方法即可
                参考官方文档：
                    1、https://docs.sqlalchemy.org/en/13/orm/session_api.html?highlight=sessionmaker#sqlalchemy.orm.session.sessionmaker
                    2、https://docs.sqlalchemy.org/en/13/orm/session_api.html?highlight=sessionmaker#sqlalchemy.orm.session.Session
            """
            return sessionmaker(bind=self.engine, query_cls=query_class, **kwargs)
    
        def create_scoped_session(self, query_class=None, **kwargs):
            """
                功能：通过scoped_session方法来创建session【Flask内部也是封装此方法创建session】
                特殊情况：由于scoped_session内部实现了__call__方法【兼容对象支持函数的调用方式】session()等价于session.__call__(*args,**kwargs)
                         这里的session变量是全局共享的,仅执行session()才会创建一个新的session实例【scoped_session每次创建的session()的实例是单例模式】
                         所谓单例模式：如果存在则共享,不存在则重新创建【在单线程中,单例模式非常重要,避免同时多个不同的session实例同时操作一个表格造成数据混乱】
                关于线程：
                    单线程：必须此方法创建session(),确保每次session()都仅存在一个活跃的实例【类似单例模式】
                    多线程：多线程创建session,用scoped_session创建或直接sessionmaker创建session没多大影响【多线程不可能共享一个实例,一个线程开启一个实例,互不干扰】
                    SQLAlchemy中的 Session、sessionmaker、scoped_session参考文档：
                        我总结的文档：https://blog.csdn.net/weixin_43343144/article/details/104434620
                        官方文档：https://docs.sqlalchemy.org/en/13/orm/contextual.html?highlight=scoped_session#sqlalchemy.orm.scoping.scoped_session
               问题汇总：
                    1、不管使用sessionmaker或scoped_session创建都是一个session实例,为何在实际使用则必须使用session()继续创建实例呢？
                        答：查看源码,会发现以上2个类内部都实现了__call__特殊方法,这个方法就是让实例兼容函数的调用【具体参数__call__的用法】
                    2：session的生命周期？以及如何关闭?
                        生命周期：参考别人经典总结：https://www.cnblogs.com/ChangAn223/p/11277468.html
                        如何关闭session：scoped_session创建的实例,关闭必须使用sopped_session的remove()方法,首先会调用.close()方法,
                                它的作用是释放会话所拥有的任何链接、事务资源的效果， 然后丢弃会话本身。“释放” 意味着链接呗返回到他们的连接池,
                                任何事务状态都会被回滚,最终使用低层的ABAPI链接rollback()的方法。
            """
            session_factory = self.create_session(query_class=query_class, **kwargs)
            return scoped_session(session_factory)
    
    
    db = OwnSQLAlchemy(query_class=OwnQuery)

from contextlib import contextmanager
    from app.libs.errors.own_global_exception import OwnGlobalException
    
    
    class OwnDBSessionRollback:
        """
            功能：基于数据库的创建session实例、提交更新、异常回滚统一的类【必须结合with才可以调用】
        """
    
        def __init__(self, db_session):
            """
                功能：
                    db_session：全局共享一个session会话管理对象
                    db_session_instance：由当前session创建出来的实例
                温馨提示：
                    1、每一次执行数据库CRUD都将创建一个全新的session实例,执行完毕之后,会立即remove移除!
                    2、单线程必须使用scoped_session方法来创建,确保同一个线程仅存在一个活跃的实例【避免数据混乱出错】
                    3、必须避免单线程同时存在多个不一样的session实例而导致操作表格的时候某些成功,而某些可能失败的情况!
            """
            self.db_session = db_session
            # 1、基于scoped_session创建的session实例是类单例模式【db_session本地线程如果存在实例则共享,否则新创建】
            # 2、每次新创建的session实例会自动加入db_session的本地线程,而通过remove方法可以从线程中移除实例!
            self.db_session_instance = db_session()
            print(f"db_session_instance_id={id(self.db_session_instance)},"
                  f"start_is_activate={self.db_session_instance.is_active}")
    
        @contextmanager
        def auto_commit_db(self, is_query=False):
            """
                功能：利用contextlib.contextmanager模块自动封装一个支持with语句的
                关于装饰器contextmanager模块【要调用的函数必须返回generator --iterator(使用yield语句)】：
                    使用装饰器模块：contextlib是为了加强with语句，提供上下文机制的模块，它是通过Generator实现的。
                    传统with方式：通过定义类以及写__enter__和__exit__来进行上下文管理虽然不难，但是很繁琐。
                            contextlib中的contextmanager作为装饰器来提供一种针对函数级别的上下文管理机制。
                参数：
                    is_query：兼容仅执行增删改等更新操作才需要commit操作【查询无数据改变,无需执行commit】
                实现原理：参考我总结的文档：https://blog.csdn.net/weixin_43343144/article/details/104438805
                温馨提示：
                    1、执行commit提交方法首先会自动执行flush方法【将所有对象更改刷新到数据库】
                    2、关于数据库更新必须遵循以下规则：
                       每次修改一张表,就必须先执行更新一次【千万不能同时修改多张表然后在统一执行更新,会严重导致数据错乱或丢失的情况】
                    3、具体原因：因为_session.commit提交才会更新数据库生效,
                       如果多张表交叉一起更新,可能导致部分表更新成功,另外部分根本没更新的问题,
                       所以为了数据更安全【尤其资金问题】,每修改一张表必须先执行commit更新成功后,再去修改另一张表继续更新！
                装饰器contextmanager执行顺序：
                    【第一步】执行yield语句以上的代码
                    【第二步】执行yield语句中的代码【也就是as db_session_instance:内部这部分代码】
                    【第三步】执行yield语句后部分代码【处理commit提交、remove移除、及异常回滚】
                细节问题：
                    1、为什么db_session_instance必须等待失效抛出异常后才能remove【而不能每次请求前新创建的新session实例,请求结束后立即remove】?
                       答案：
                            A、首先基于scoped_session新创建的session()实例会自动加入本地线程,只要没有执行remove之前,都会是同一个实例【类单例模式】
                            一旦remove则重新创建一个,全新的session()实例,重新添加到db_session的本地线程中,来确保单线程永远仅有一个活跃的session实例！
                            B、比如执行先查询后更新操作：首先通过一个db_session_instance执行query获取一个model,然后必须使用同一个使用同一个session实例去更新,
                            而这时如果你当前的session实例已经被remove,再一次创建的实例肯定不一样了,此时的model在新的session已失效,
                            最终导致更新失败【这也是为什么执行更新失败的根本原因,因为查询后更新必须是同一个session实例】
            """
            try:
                # 【非常重要】这里生成器返回的结果会传递给as后的对象【比如： with OwnDBSessionRollback() as db_session_instance】
                yield self.db_session_instance
                if not is_query:
                    self.db_session_instance.commit()
            except BaseException as e:
                """
                    关于session异常处理：
                        1、一旦当前db_session_instance实例连接出异常了,就要从db_session本地线程中移除,
                           当下一次重新请求数据库时,在创建一个新的new_db_session_instance实例加入db_session本地线程,
                           确保每一次正常运行单线程中有且仅有一个活跃的db_session_instance实例
                        2、一旦session出现异常,必须先回滚,然后在移除【确保本次请求的数据恢复原始状态】
                """
                self.db_session_instance.rollback()
                self.db_session.remove()
                raise OwnGlobalException(
                    f"由于本次mysql请求连接失败,出现了异常,强制执行事务回滚后," +
                    f"从本地线程remove此session实例,并创建一个新的实例并添加到本地线程" +
                    f"【确保单线程永远只有一个活跃的db_session_instance】" +
                    f"本次请求具体异常原因如下：{e}")
            finally:
                """
                    1、每次请求之后必须关闭db_session_instance实例释放资源,避免资源浪费引发【异常：达到大小<x>溢出<y>的QueuePool限制,连接超时】
                    2、只要db_session_instance没有从db_session本地线程移除,单线程就会共享此session实例
                    3、执行db_session()打开db_session_instance实例,而执行db_session_instance.close关闭实例
                """
                self.db_session_instance.close()

[https_www.cnblogs.com_ChangAn223_p_11277468.html]: https://www.cnblogs.com/ChangAn223/p/11277468.html
[https_www.cnblogs.com_ybjourney_p_11876595.html]: https://www.cnblogs.com/ybjourney/p/11876595.html