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

sac*_*che 5 python sqlalchemy

我想检查数据库中的不同值并创建一个新值,所以我需要查询,但我不知道是否必须在 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())
Run Code Online (Sandbox Code Playgroud)

我正在使用 SQLALchemy 6.2 和声明式

谢谢

Mar*_*nas 1

您必须进行自己的会话管理——例如定义一个模块全局会话对象。

例如,Pylons 应用程序如下定义其会话:

from sqlalchemy.orm import scoped_session, sessionmaker
Session = scoped_session(sessionmaker())
Run Code Online (Sandbox Code Playgroud)

然后将其绑定到引擎

Session.configure(bind=engine)
Run Code Online (Sandbox Code Playgroud)

使用scoped_session意味着您的代码是线程安全的(每个线程将使用自己的会话)。