Pyramid 1.5 で、Jinja2 で、MySQL を使う、というパターンです。
- 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 - 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) - 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 [] - 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 を使ったログインフォームができました。
