Pyramidでトランザクションを自動的に開始する方法はありますか？

Question

私は、SQLAlchemy、pyramid_tm、pyramid_beaker、およびalembicを使用する単純なPyramidアプリケーションを作成しました。データベースは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を削除した場合、私は得る：

私は /loginに投稿する際

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

はまた、私は、ユーザーIDについてと何のセッション変数がすべてではありません/へのリダイレクト後_accessed_timeと_creation_time何もしてDebugToolbarでRequest Varsタブでセッション変数を参照してください。

Answer 1

挿入を実行し、エラー（ロールバック）を処理する適切な方法は、セーブポイントとflush()を使用することです。あなたは、単に余分な定型のいずれかなし.addを使用することができるよう

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 

しかし、あなたもあなたの例ではエラーで何もしていません。

要求の最後に、pyramid_tmは最後のコミットを発行します。フラッシュは、データベース上の開いているトランザクションで保留中のSQLコマンドを実行し、潜在的なエラーをキャッチすることができます。