我想检查数据库中的不同值并创建一个新值,所以我需要查询,但我不知道是否必须在 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 和声明式
谢谢
您必须进行自己的会话管理——例如定义一个模块全局会话对象。
例如,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意味着您的代码是线程安全的(每个线程将使用自己的会话)。
| 归档时间: |
|
| 查看次数: |
2978 次 |
| 最近记录: |