Pyramid 1.5 + jinja2-alchemy-starter で MySQL の authentication

前回の「Pyramid 1.5 + jinja2-alchemy-starter で sqlite の authentication」では、sqlite でしたが、今回はデータベースを MySQL にしてみます。
Pyramid 1.5 で、Jinja2 で、MySQL を使う、というパターンです。

1. development.ini
   
   普通はこんな感じの接続なのですが、これではつながりませんでした。
   

   sqlalchemy.url=mysql+mysqldb://root:root123@localhost/test
   

   
   私の環境は /tmp/mysql.sock で接続するタイプなので、こうなります。
   

   sqlalchemy.url=mysql+pymysql://pyramid:pyramidpass@localhost/pyramid_myproject?unix_socket=/tmp/mysql.sock
   

   
   ほかに、mysqldb ではなくて、pymysql にしました。pip install で pymysql をインストールしておいてください。
   データベースの user は pyramid、そのパスワードは pyramidpass にしてあります。
   データベース名は pyramid_myproject です。
   
   その近辺を抜粋します。
   

   jinja2.directories = myproject:templates
   
   #SQlalchemy configuration (The prefix `sqlalchemy.` can be changed IF you change it in __init__.py's engine_from_config() call too
   # $ bin/pip install pymysql
   sqlalchemy.url=mysql+pymysql://pyramid:pyramidpass@localhost/pyramid_myproject?unix_socket=/tmp/mysql.sock
   sqlalchemy.pool_recycle = 3600
   sqlalchemy.pool_size = 20
   
   [pipeline:main]
   pipeline =
       MyProject
   
   

   
   
2. models.py
   
   基本的には変わっていませんが、 Text というところを String にしました。Text のままでも、あるいは VARCHAR でも良いかも。
   def initialize_sql(engine): が __init__.py から使われるところは変わりません。
   

   #!/usr/bin/env python
   # coding: UTF-8
   from pyramid.security import (
       Allow,
       Everyone,
       )
   from sqlalchemy import (
       Column,
       Integer,
       String,
       create_engine,
       )
   from sqlalchemy.ext.declarative import declarative_base
   from sqlalchemy.orm import (
       scoped_session,
       sessionmaker,
       )
   
   DBSession = scoped_session(sessionmaker(autocommit=False,
                                            autoflush=False))
   Base = declarative_base()
   Base.query = DBSession.query_property()
   
   # Table Class
   class User(Base):
       """ The SQLAlchemy declarative model class for a User object. """
       __tablename__ = 'users'
       __table_args__ = {  
           'mysql_engine': 'InnoDB',  
           'mysql_charset': 'utf8'  
       }   
       id = Column('id', Integer, primary_key=True)
       name = Column('name', String(255), unique=True)
       password = Column('password', String(255))
       group = Column('group', String(255))
   
       def __init__(self, name, password, group):
           self.name = name
           self.password = password
           self.group = group
   
   class RootFactory(object):
       __acl__ = [ (Allow, Everyone, 'view'),
                   (Allow, 'group:editors', 'edit') ]
       def __init__(self, request):
           pass
   
   def initialize_sql(engine):
       DBSession.configure(bind=engine)
   

   
   
3. security.py
   
   テーブルの作成など、前回は SQL 文を書きましたが、今回は sqlalchemy です。
   myproject.my_util はもう使いません。
   

   #!/usr/bin/env python
   # coding: UTF-8
   import pymysql
   import sqlalchemy
   from myproject.models import (DBSession, User)
   
   class MyProjectDB():
       def __init__(self):
           self.session = DBSession
           try:
               self.session.query(User).all()
           except Exception as ex:
               # テーブルがないので作成する。
               print('@MyProjectDB self.session.query(User).all()')
               print(ex)
               user = User.__table__
               engine = self.session.bind
               try:
                   user.create(engine)
               except Exception as ex:
                   print('@MyProjectDB user.create(engine)')
                   print(ex)
               new_user = User('editor', 'editor', 'editors')
               self.session.add(new_user)
               self.session.commit()
           # group='editors' の存在確認。なければつくる。
           try:
               num = self.session.query(User).filter_by(group='editors').count()
               if num == 0:
                   new_user = User('editor', 'editor', 'editors')
                   self.session.add(new_user)
                   self.session.commit()
           except Exception as ex:
               print('@MyProjectDB self.session.add(new_user)')
               print(ex)
       
       def get_group_and_password_of_user(self, user):
           group = None
           password = None
           if self.session is not None:
               try:
                   records = self.session.query(User).filter_by(name=user)
                   for row in records:
                       group = row.group
                       password = row.password
               except Exception as ex:
                   print('@get_group_and_password_of_user')
                   print(ex)
                   group = None
                   password = None
           return (group, password)
       
       def put_group_and_password_of_user(self, user, group, password):
           stmt = sqlalchemy.update(User).where(User.c.name==user).values(group=group, password=password)
       
   def groupfinder(userid, request):
       db = MyProjectDB()
       (group, password) = db.get_group_and_password_of_user(userid)
       the_str = None
       # ['group:editors']
       if group is not None:
           the_str = 'group:' + group
           return [the_str]
       else:
           return []
   

   
   

   
   
4. views.py の def login(request): のところ
   
   models.py の USERS はもう使えません。
   security.py の get_group_and_password_of_user() を代わりに使います。
   ＜修正前＞
   

       if 'form.submitted' in request.params:
           login = request.params['login']
           password = request.params['password']
           if USERS.get(login) == password:
               headers = remember(request, login)
               return HTTPFound(location = came_from,
                                headers = headers)
           message = 'Failed login'
   

   
   ＜修正後＞
   

       if 'form.submitted' in request.params:
           login = request.params['login']
           password = request.params['password']
           my_db = myproject.security.MyProjectDB()
           (group, pwd) = my_db.get_group_and_password_of_user(login)
           if pwd == password:
               headers = remember(request, login)
               return HTTPFound(location = came_from,
                                headers = headers)
           message = 'Failed login'
   

   
   

   
   

これで MySQL を使ったログインフォームができました。