如何在SQLAlchemy的Base类方法中直接访问Session？

方案1：使用SQLAlchemy内置的object_session #

SQLAlchemy提供了sqlalchemy.orm.session.object_session函数，能直接从模型实例获取它绑定的Session——只要这个实例是通过Session查询出来、或者已经被添加到Session中，就能拿到对应的Session。

修改你的代码：

from sqlalchemy.orm.session import object_session

class Job(Base):
    __tablename__ = 'jobs'

    # id = ..., etc.
  
    def evaluate(self):  
        session = object_session(self)
        keywords = session.query(Keyword).all()
        # Do something with keywords

方案2：给Base类绑定线程安全的全局Session #

如果是多线程应用，建议用threading.local()存储Session，保证每个线程有独立的Session实例，再给Base类添加属性供子类调用：

import threading
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

engine = create_engine('your-db-url')
Session = sessionmaker(bind=engine)
local_storage = threading.local()

# 给当前线程初始化Session
local_storage.session = Session()

Base = declarative_base()

# 给Base类添加Session属性
class BaseWithSession(Base):
    __abstract__ = True  # 标记为抽象基类，不会生成数据库表

    @property
    def session(self):
        return local_storage.session

class Job(BaseWithSession):
    __tablename__ = 'jobs'

    # id = ..., etc.
  
    def evaluate(self):  
        keywords = self.session.query(Keyword).all()
        # Do something with keywords

方案3：利用模型实例的query属性（仅限声明式基类） #

如果你的Base是declarative_base()生成的，已绑定Session的实例可以直接通过self.query执行查询，底层会自动调用绑定的Session：

class Job(Base):
    __tablename__ = 'jobs'

    # id = ..., etc.
  
    def evaluate(self):  
        keywords = self.query(Keyword).all()
        # Do something with keywords

注：这个方法要求实例必须已和Session绑定（即已被添加到Session或从Session查询得到），否则self.query会抛出异常。

内容的提问来源于stack exchange，提问作者Zack Plauché