如何不在 Pyramid 中自动启动交易？

Question

我创建了简单的 Pyramid 应用程序，它使用 SQLAlchemy、pyramid_tm、pyramid_beaker 和 alembic。数据库是 PostgreSQL，适配器是 pg8000。现在我正在尝试实现登录，但是对数据库的第一个 DB 查询会创建 BEGIN 事务并永远挂起。我只想在需要时设置事务（UPDATE、DELETE、INSERT 和更复杂的多查询）。

models/user.py:

from sqlalchemy import Column
from sqlalchemy import Unicode
from sqlalchemy import Sequence
from sqlalchemy import Integer
from sqlalchemy import Index
from sqlalchemy import CheckConstraint
from sqlalchemy import text
from sqlalchemy import func
from sqlalchemy.dialects.postgresql import TIMESTAMP

from pyramid.security import Allow

import sqlalchemy.orm.exc as a_exc

import logging

log = logging.getLogger(__name__)


from ..models import DBSession
from ..models import Base

class UserNotFoundException(ValueError):
    pass


class User(Base):
    __tablename__ = 'users'

    __table_args__ = (
        CheckConstraint("login ~* '^[a-z]{3,}$'", name = __tablename__ + "_chk_login"),
        CheckConstraint("login != ''", name = __tablename__ + "_chk_login_not_empty"),
        CheckConstraint("password != ''", name = __tablename__ + "_chk_pw_not_empty"),
        Index(__tablename__ + "_idx_lower_login", text("lower(login)"), unique = True),
    )

    id = Column(Integer, Sequence('users_id_seq'), primary_key = True)
    login = Column(Unicode(64), unique = True, nullable = False, server_default = text("''"))
    password = Column(Unicode(255), nullable = False, server_default = text("''"))
    added = Column(TIMESTAMP, nullable = False, server_default = text("NOW()"))


    @property
    def __acl__(self):
        return [(Allow, self.login, 'view'), ]

    def __init__(self, login, password):
        self.login = login
        self.password = password

    @classmethod
    def get_user(self, login):
        try:
            u = DBSession.query(User).filter(User.login == login).one()
            DBSession.flush()
            return u
        except a_exc.NoResultFound as exc:
            raise UserNotFoundException(exc)

    @classmethod
    def get_user_count(self):
        u = DBSession.query(func.count(User.id)).scalar()
        DBSession.flush()
        return u

    @classmethod
    def create_session(self, login: str, password: str) -> object:
        u = self.get_user(login)

        import bcrypt
        password = password.encode('utf-8')

        try:
            verified = bcrypt.checkpw(password = password, hashed_password = u.password.encode('utf-8'))
        except Exception as exc:
            raise

        if verified != True:
            raise Exception("Coulnd't verify password hash")

        return {'userid': u.id}

    @classmethod
    def add_user(self, login, password):
        import bcrypt
        password = password.encode('utf-8')

        encrypted_pw = bcrypt.hashpw(password, bcrypt.gensalt())
        verified = False

        log.debug("Encrypted PW: '%s'", encrypted_pw)

        try:
            verified = bcrypt.checkpw(password = password, hashed_password = encrypted_pw)
        except Exception:
            raise

        if verified != True:
            raise Exception("Coulnd't verify password hash")

        try:
            DBSession.begin(subtransactions=True)
            DBSession.add(User(login = login, password = encrypted_pw.decode()))
            DBSession.commit()
            log.debug("User added: '%s'", login)
        except Exception as exc:
            DBSession.rollback()
            log.debug("User add failed for user '%s'", login)
            raise

views/views.py:

@view_config(route_name = 'login', renderer = 'templates/login.pt')
def app_login_view(request: Request):
    if request.authenticated_userid:
        # Already logged in -> redirect
        import pyramid.httpexceptions as exc
        return exc.HTTPFound(request.route_path('home'))

    user_not_found_error = {
        'page_background': 'warning',
        'page_title':      _(u"Login failed"),
        'page_text':       _(u"Check username and password."),
    }

    form_user = request.POST.get('user')
    form_password = request.POST.get('password')

    from ..models import User, UserNotFoundException

    if User.get_user_count() == 0:
        # No users in DB
        log.debug("Creating admin user")
        User.add_user(u"admin", u"admin")

    try:
        ses = User.create_session(form_user, form_password)
        request.session['userid'] = ses['userid']
        request.session.save()
        remember(request, ses['userid'])
    except UserNotFoundException as exc:
        log.debug("User '%s' not found in database", form_user)
        return user_not_found_error
    except:
        raise

    # Redirect to front page
    import pyramid.httpexceptions as exc
    return exc.HTTPFound(request.route_path('home'))

日志：

INFO sqlalchemy.engine.base.Engine.dbconn BEGIN (implicit)
INFO sqlalchemy.engine.base.Engine.dbconn SELECT count(users.id) AS count_1 
FROM users
INFO sqlalchemy.engine.base.Engine.dbconn ()
DEBUG [waitress] Creating admin user
DEBUG [user][waitress] Encrypted PW: 'b'$2b$12$n6mN973Gz0wwX7B0kWI.Ae099h7mvLo.mEI.D2NFjZKaLKbGebK16''
INFO sqlalchemy.engine.base.Engine.dbconn INSERT INTO users (id, login, password) VALUES (nextval('users_id_seq'), %s, %s) RETURNING users.id
INFO sqlalchemy.engine.base.Engine.dbconn ('admin', '$2b$12$n6mN973Gz0wwX7B0kWI.Ae099h7mvLo.mEI.D2NFjZKaLKbGebK16')
INFO  [sqlalchemy.engine.base.Engine.dbconn:109][waitress] INSERT INTO users (id, login, password) VALUES (nextval('users_id_seq'), %s, %s) RETURNING users.id
INFO  [sqlalchemy.engine.base.Engine.dbconn:109][waitress] ('admin', '$2b$12$n6mN973Gz0wwX7B0kWI.Ae099h7mvLo.mEI.D2NFjZKaLKbGebK16')
... Hangs here forever ...

如果我从add_user() 中删除subtransactions=True，我会得到：

sqlalchemy.exc.InvalidRequestError: A transaction is already begun.  Use subtransactions=True to allow subtransactions.

此外，当我发布到 /login 时，我在 DebugToolbar 的 Request Vars 选项卡中看到带有 _accessed_time 和 _creation_time 的会话变量，但没有关于用户 ID 的信息，并且在重定向到 / 之后根本没有会话变量.

Answer 1

执行插入和处理错误（回滚）的适当方法是使用保存点和flush()。

sp = request.tm.savepoint()
try:
    DBSession.add(User(login = login, password = encrypted_pw.decode()))
    DBSession.flush()
    log.debug("User added: '%s'", login)
except Exception as exc:
    sp.rollback()
    log.debug("User add failed for user '%s'", login)
    raise

但是，您甚至没有对示例中的错误做任何事情，因此您可以简单地使用 .add 而不需要任何额外的样板。

在请求结束时，pyramid_tm 将发出最终提交。刷新在数据库上的一个打开的事务中执行挂起的 SQL 命令，允许您捕获潜在的错误。