如何使用会话而不需传递它 [SqlAlchemy]

Question

我想在数据库里检查不同的值，然后创建一个新的值，所以我需要进行查询，但我不太清楚在我的SQLAlchemy类里是否需要创建一个会话，或者该怎么做？是把会话当作全局使用吗？我在文档里没找到相关信息。

大概是这样的：

class MyClass(Base):

    __tablename__ = 'my_class'
    __table_args__ = (UniqueConstraint('key', 'key2'),
                     {}
                     )
    id =  Column(Integer, Sequence('my_class_id'), primary_key=True)
    key = Column(String(30), nullable= False) #unique together key2
    key2 = Column(String(30), nullable = False) 
    value = Column(Integer, nullable=False)

    def __init__(self, key, key2):
        #check if exist key and key2
        values = session.query(MyClass.value).filter(MyClass.key == self.key).\
            filter(MyClass.key2 == self)

        if values:
            raise IntegrityError

        #get biggest value
        value = session.query(MyClass.value).filter(MyClass.key = self.key).order_by(asc(MyClass.value)) #I'm not shure if i need 'asc'
        #no value new key and key2

        if not value:
            self.key = key
            self.key2 = key2
            self.value = '000'
            return

        #i used a single table beacuse is easier to understand
        #in this example
        self.key = key
        self.key2 = key
        self.value = increment(value.first())

我正在使用SQLAlchemy 6.2和声明式方式

谢谢

Answer 1

我在这里发现，我们可以使用Session.object_session(self)这个方法：

def new_value(self):
    #not really DRY
    #the object has to be binded with some session first.
    session = Session.object_session(self) # << this is the important stuff

    #check if exist key and key2
    values = session.query(MyClass.value).filter(MyClass.key == self.key).\
        filter(MyClass.key2 == self)

    if values:
        return #None

    #get biggest value
    value = session.query(MyClass.value).\
               filter(MyClass.key = self.key).\
               order_by(desc(MyClass.value))

    return increment(value.first())

Answer 2

你需要自己管理会话，也就是说，要定义一个全局的会话对象。

比如，Pylons 应用程序是这样定义它们的会话的：

from sqlalchemy.orm import scoped_session, sessionmaker
Session = scoped_session(sessionmaker())

然后稍后会把它绑定到一个引擎上，使用：

Session.configure(bind=engine)

使用 scoped_session 的话，你的代码就会是线程安全的（每个线程都会使用自己的会话）。