[ Python ] flask-loginでログイン認証を実装してみた

$ tree
.
├── server.py
├── database.py
└── templates
    ├── login.html
    └── mypage.html

1 directory, 7 files

$ pip install flask-login

#!/home/<user_name>/.virtualenvs/web1/bin/python

import sys
from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.sql.functions import current_timestamp
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, TIMESTAMP
from flask_login import UserMixin # <= 新たに追加

DATABASE = 'mysql://%s:%s@%s/%s?charset=utf8mb4' % (
    "docker",  #ユーザー「docker」の
    "docker",  #パスワード「docker」で
    "127.0.0.1:3306",  #MySQLサーバ(http:\/\/127.0.0.1:3306)に接続し
    "test_db", #データベース「test_db」にアクセスします
)

#DB接続用のインスタンスを作成
ENGINE = create_engine(
    DATABASE,
    convert_unicode=True,
    echo=True  #SQLをログに吐き出すフラグ
)

#上記のインスタンスを使って、MySQLとのセッションを張ります
session = scoped_session(
    sessionmaker(
        autoflush = False,
        autocommit = False,
        bind = ENGINE,
    )
)

#以下に書いていくDBモデルのベース部分を作ります
Base = declarative_base()
Base.query = session.query_property()

#DBとデータをやり取りするためのモデルを定義
class User(Base):
    __tablename__ = 'users'
    id = Column('id', Integer, primary_key = True)
    name = Column('name', String(200))
    age = Column('age', Integer)
    description = Column('description', TEXT)
    topics = Column('topics', TEXT)
    created_at = Column('created_at', TIMESTAMP, server_default=current_timestamp())
    updated_at = Column('updated_at', TIMESTAMP, nullable=False, server_default=text('CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP'))

# 今まで利用していたUserモデルと、ログイン機能用のUserMixinを合成
class LoginUser(UserMixin, User):
    # このモデルを介して認証ユーザーIDを内部で取得するためのメソッド(このメソッド名で作ってあげます)
    def get_id(self):
        return self.id

def main(args):
    Base.metadata.drop_all(bind=ENGINE)
    Base.metadata.create_all(bind=ENGINE)

#このファイルを直接実行したとき、mainメソッドでテーブルを作成する
if __name__ == "__main__":
    main(sys.argv)

def get_id(self):
    return self.id  

class LoginUser(UserMixin, User):
  pass

class UserMixin(object):
    if not PY2:
        __hash__ = object.__hash__

    @property
    def is_active(self):
        return True

    @property
    def is_authenticated(self):
        return True

    @property
    def is_anonymous(self):
        return False

    def get_id(self):
        try:
            return text_type(self.id)
        except AttributeError:
            raise NotImplementedError('No `id` attribute - override `get_id`')

    def __eq__(self, other):
        '''
        Checks the equality of two `UserMixin` objects using `get_id`.
        '''
        if isinstance(other, UserMixin):
            return self.get_id() == other.get_id()
        return NotImplemented

    def __ne__(self, other):
        '''
        Checks the inequality of two `UserMixin` objects using `get_id`.
        '''
        equal = self.__eq__(other)
        if equal is NotImplemented:
            return NotImplemented
        return not equal

from flask import Flask, abort, render_template, request, redirect, url_for
from flask_wtf.csrf import CSRFProtect
import os
from database import *
from flask_login import LoginManager, login_required, login_user, logout_user

login_manager = LoginManager()

app = Flask(__name__)
login_manager.init_app(app)

app.config['SECRET_KEY'] = os.urandom(24) #セッション情報を暗号化するための設定
csrf = CSRFProtect(app)

@login_manager.user_loader
def load_user(user_id):
    return LoginUser.query.filter(LoginUser.id == user_id).one_or_none()

@app.route('/login', methods=['GET'])
def form():
    return render_template('login.html')

@app.route('/login', methods=['POST'])
def login():
    name = request.form.get("name", "")
    try:
        user = LoginUser.query.filter(LoginUser.name == name).one_or_none()
        if user == None:
            return render_template('login.html', error="指定のユーザーは存在しません")
        else:
            login_user(user, remember=True)
    except Exception as e:
        return redirect(url_for('mypage'))
    return redirect(url_for('mypage'))

@app.route("/logout")
@login_required
def logout():
    logout_user()
    return redirect(url_for('login'))

@app.route('/mypage', methods=['GET'])
@login_required
def mypage():
    return render_template('mypage.html')

if __name__=="__main__":
    app.debug = True
    app.run(host='0.0.0.0', port="8888")

login_manager.init_app(app)

@login_manager.user_loader
def load_user(user_id):
    return LoginUser.query.filter(LoginUser.id == user_id).one_or_none()

login_user(セッションに保存するユーザーオブジェクト)

@login_required

<html>
<head>
    <meta charset="utf-8">
</head>
<body>
    <p>{% if error %}{{ error }}{% endif %}</p>
    <form action="/login" method="post">
        <input type="hidden" name="csrf_token" value="{{ csrf_token() }}"/>
        <div>
            name <input type="text" name="name">
        </div>
        <div>
            <input type="submit">
        </div>
    </form>
</body>
</html>

<html>
<head></head>
<body>
    <div>ログインに成功しました。ここはあなたのページです。</div>
    <a href="/logout">ログアウト</a>
</body>
</html>

logout_user()