原生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，實現了線程安全