0
我創建了使用SQLAlchemy，pyramid_tm，pyramid_beaker和alembic的簡單金字塔應用程序。數據庫是PostgreSQL，適配器是pg8000。現在我試圖實現登錄，但是對數據庫的第一個數據庫查詢創建BEGIN事務並永久掛起。我只想在需要時設置事務（更新，刪除，插入和更復雜的多查詢）。如何不在金字塔中自動啓動交易？
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，但無需用戶名和重定向到/根本沒有會話變量之後。
您不應在請求中共享SQLAlchemy會話。爲每個請求創建（並隨後銷燬）一個新的'Session'。 – univerio