原生session:
from sqlalchemy.orm import sessionmaker from sqlalchemy import create_engine from sqlalchemy应用.models import Users engine = create_engine( "mysql+pymysql://root:root@127.0.0.1:3306/pro6?charset=utf8", max_overflow=0, # 超过连接池大小外最多创建的连接 pool_size=5, # 连接池大小 ) #from sqlalchemy.orm.session import Session SessionF = sessionmaker(bind=engine) session = SessionF() print(session) obj1 = Users(name=‘ctz‘, email=‘49274573@qq.com‘, extra=‘aaaa‘) session.add(obj1) session.commit() session.close()
问题：由于无法提供线程共享功能，所以在开发时要注意,要给每个线程都创建自己的session
打印sesion可知他是sqlalchemy.orm.session.Session的对象
查看Session的源码 可得到:
class Session(_SessionClassMethods): """Manages persistence operations for ORM-mapped objects. The Session‘s usage paradigm is described at :doc:`/orm/session`. """ public_methods = ( ‘__contains__‘, ‘__iter__‘, ‘add‘, ‘add_all‘, ‘begin‘, ‘begin_nested‘, ‘close‘, ‘commit‘, ‘connection‘, ‘delete‘, ‘execute‘, ‘expire‘, ‘expire_all‘, ‘expunge‘, ‘expunge_all‘, ‘flush‘, ‘get_bind‘, ‘is_modified‘, ‘bulk_save_objects‘, ‘bulk_insert_mappings‘, ‘bulk_update_mappings‘, ‘merge‘, ‘query‘, ‘refresh‘, ‘rollback‘, ‘scalar‘)
2.scoped_session
from sqlalchemy.orm import sessionmaker from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import create_engine from sqlalchemy应用.models import Users from sqlalchemy.orm import scoped_session engine=create_engine( "mysql+pymysql://root:root@127.0.0.1:3306/pro6?charset=utf8", max_overflow=0, # 超过连接池大小外最多创建的连接 pool_size=5, # 连接池大小 ) SessionF=sessionmaker(bind=engine) #scoped_session封装了两个值 Session 和 registry,registry加括号就执行了ThreadLocalRegistry的__call__方法,如果 # 当前本地线程中有session就返回session,没有就将session添加到了本地线程 #self.registry()=session session=scoped_session(SessionF) print(session) obj1=Users(name=‘ctz‘,email=‘49274573@qq.com‘,extra=‘aaaa‘) session.remove() session.query_property()
优点:支持线程安全,为每个线程都创建一个session:
两种方式:通过本地线程Threading.Local()和创建唯一标识的方法(参考flask请求源码)
源码分析：
首先我们在scoped_session中放入了Sesion对象
SessionF=sessionmaker(bind=engine)
session=scoped_session(Session)
一、scoped_session类中
class scoped_session(object): session_factory = None def __init__(self, session_factory, scopefunc=None): #传递过来的那个Session对象 self.session_factory = session_factory #scopefunc唯一标示函数 if scopefunc: self.registry = ScopedRegistry(session_factory, scopefunc) else: self.registry = ThreadLocalRegistry(session_factory)
第一次进来scopefunc唯一标识为None，我们将Session作为参数传递到ThreadLocalRegistry中，
ThreadLocalRegistry类中
class ThreadLocalRegistry(ScopedRegistry): """A :class:`.ScopedRegistry` that uses a ``threading.local()`` variable for storage. """ def __init__(self, createfunc): #传递过来的那个Session对象 self.createfunc = createfunc self.registry = threading.local() #scoped_session.registry()后执行 def __call__(self): try:
#如果本地线程中有值的话直接将值返回， return self.registry.value except AttributeError:
#没有值的话就示例话Session(),并将他存到本地线程中，并把实例的对象返回 #相当于Session()后的对象加到了本地线程中 val = self.registry.value = self.createfunc() return val
其中__call__()只有当scoped_session.registry加括号执行
那我们怎么调用那些方法呢？
def instrument(name): def do(self, *args, **kwargs): return getattr(self.registry(), name)(*args, **kwargs) return do for meth in Session.public_methods: setattr(scoped_session, meth, instrument(meth))
其中 Session就是sqlalchemy.orm.session.Session
public_methods = ( ‘__contains__‘, ‘__iter__‘, ‘add‘, ‘add_all‘, ‘begin‘, ‘begin_nested‘, ‘close‘, ‘commit‘, ‘connection‘, ‘delete‘, ‘execute‘, ‘expire‘, ‘expire_all‘, ‘expunge‘, ‘expunge_all‘, ‘flush‘, ‘get_bind‘, ‘is_modified‘, ‘bulk_save_objects‘, ‘bulk_insert_mappings‘, ‘bulk_update_mappings‘, ‘merge‘, ‘query‘, ‘refresh‘, ‘rollback‘, ‘scalar‘)
在instrument函数中
self.registry()帮我们执行了ThreadLocalRegistry中的___call__方法，拿到了sesion对象
方法源码示例:
def has(self): return hasattr(self.registry, "value")
def remove(self): if self.registry.has(): self.registry().close() self.registry.clear()
实际两种方式原理都是一样的都是第一种，只是第二种将session放到了本地线程中，为每一个进程都设置了一个session，实现了线程安全
