原生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，实现了线程安全