正确使用 sqlalchemy scoped_session 和金字塔模型

Question

我在寻找使用数据库查询的正确方法时遇到问题。我在网络开发方面没有太多经验，就像下班后工作 6 个月（所以每周最多 10 小时）。我选择 Pyramid 主要是因为 Py3k 支持。它没有 django 那么多的社区支持，也没有教程，所以我可能做错了。所以...

我正在使用 Pyramid(1.4) 和 SQLAlchemy (0.8)。我已经按照 Pyramid 教程中的描述创建了项目。

我已经阅读了这篇关于使用 scoped_session 的精彩帖子 (https://stackoverflow.com/a/11547942/1498245)。

但是.. 我一直在努力采用适当的（如果存在的话）数据库对象处理方法。

所以这是我开始的：

## Project layout
myproject
├── models
│   ├── somemodel1.py
│   ├── somemodel2.py
│   ├── __init__.py
│   ├── meta.py
│   └── somemodel3.py
├── schemas
│   ├── some_colander_schema1.py
│   ├── some_colander_schema2.py
│   └── some_colander_schema3.py
├── scripts
│   ├── __init__.py
│   └── initializedb.py
├── static
│   ├── bootstrap
│   ├── favicon.ico
│   ├── jquery-1.9.1.min.js
│   └── transparent.gif
├── templates
│   ├── some_chameleon_template.pt
│   ├── some_chameleon_template1.pt
│   ├── some_chameleon_template2.pt
│   ├── forbidden_view.pt
│   ├── global_layout_noauth.pt
│   └── global_layout.pt
├── views
│   ├── someviewclass1.py
│   ├── someviewclass2.py
│   └── someviewclass3.py
└── __init__.py


# models/meta.py PRE @contextmaker
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import scoped_session, sessionmaker

DBSession   = scoped_session(sessionmaker())
Base        = declarative_base()

现在进入重点。

在我看来，我使用的是这个结构：

class SomeView1(Layout): #in Layout class request is saved to self.request
    @view_config(...)
    def list(self):
        DBSession()
        try:
            somethings = DBSession.query(SomeModel).options(eagerload(...)).filter(...).all()
        except: #it's bad to catch everything I know, but had to close session properly
            somethings = []
            DBSession.rollback()
        DBSession.remove()

        ... #do something more

        DBSession()
        try:
            somethingMores = DBSession.query(SomeModel2).options(eagerload(...)).filter(...).all()
        except: #it's bad to catch everything I know, but had to close session properly
            somethingMores = []
            DBSession.rollback()
        DBSession.remove()

        return {"values" : somethings,
                "others" : somethingMores}

    @view_config(...)
    def edit(self):
        # here's recently implemented "update"
        DBSession()
        try:
            somethings = DBSession.query(SomeModel1)...all()
            somethingsMore = DBSession.query(SomeModel2)...all()
        finally:
            DBSession.remove()
        return {"somethings":somethings,
                "somethingsMore": somethingsMore}

但后来我意识到我有很多这样的查询......所以我把它们移到了模型中。

在我看来，我只需要打电话：

somethings = SomeModel1.all()
#or
something = SomeModel.by_id(some_id)

在SomeModel1 中，例如：

class SomeModel(Base):
    __tablename__ = "somemodel"
    idsomeModel = Column(Integer, primary_key=True)

    @classmethod
    def all(cls):
        DBSession()
        try:
            retval = ...
        except:
            retval = None
        finally:
            DBSession.remove()
        return retval

这一切都很好（不是真的，但是等等..）但是当我必须例如放置排序顺序或 where 子句时它会变得很讨厌。然后我的def all(cls) 变成了def all(cls, **kwargs) 和很多ifs，比如：

query = DBSession.query(SomeModel)
if "forPeriod" in kwargs:
    query = query.filter(SomeModel.date > kwargs["forPeriod"])
if "sortOrder" in kwargs:
    query = query.order_by(kwargs["sort"])

retval = query.all()

没事吧？或者我应该在我的视图中写查询？我想没有“好方法”，但我觉得这种方法更易于维护。

这并不好。现在我在 sqlalchemy 网站上读到，将会话传递给模型要好得多。在我看来，我会这样使用它：

@view_config(...)
def some_function(self):
    with get_db_session as session:
        somethings = SomeModel.all(session, some_arguments)

那个函数get_db_session来自新的models/meta.py：

# models/meta.py
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import scoped_session, sessionmaker
from contextlib import contextmanager

DBSession   = scoped_session(sessionmaker())
Base        = declarative_base()

@contextmanager
def get_scoped_session():
    try:
        yield DBSession()
    finally:
        DBSession.remove()

这是一篇糟糕的帖子，但我希望有人告诉我“好吧，你的方向很好”或“不，f*ck man，你在做什么？”

所以总结一下：

1. 在模型中隐藏查询是个好主意吗？ （有时在kwargs 上有很多ifs；并且模糊异常-如果发生这种情况，我的返回值为None，因为在视图中我不在乎是什么原因，我把它记录下来（在模型中））

2. 将 scoped_session 与@contextmanager 和with 语句一起使用是一种好方法吗？

3. 不，没有第三个问题.. 这么多文字只是为了问两个问题：/

Answer 1

#1：

我会使用 scoped_session.query_property() 进行查询，而不是隐藏太多，设置你的 Base 如下：

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import scoped_session, sessionmaker

DBSession = scoped_session(
    sessionmaker(
        extension='ZopeTransactionExtension'  # for your 2nd question
    )
)

class ORMClass(object):
    query = DBSession.query_property()

Base = declarative_base(cls=ORMClass)

像往常一样从这个基类派生你的模型：

class MyModel(Base):
    ...

现在您将能够：

MyModel.query.filter_by(attr1='something').all()

对于#2： 我同意 umeboshi，使用事务管理器

Answer 2

就#2而言，我使用transaction

with transaction.manager:
    p = Page()
    p.title = "Page Title"
    p.content = "Page Content"
    dbsession.add(p)
p = dbsession.merge(p)