flask-login とは？

ユーザー認証を簡単に扱うための Flask の拡張パッケージ。

インストール

pip install flask-login

Flask-Login documnetation

設定

以下のコマンドを実行し、必要なパッケージをインストールします。

pip install flask sqlalchemy flask-sqlalchemy flask-wtf mysql-connector-python pymysql

project
- templates
  - index.html
  - login.html
- app.py

ファイル

app.py

from flask import Flask, render_template, request, redirect, url_for, session, flash
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.exc import NoResultFound, MultipleResultsFound
from flask_login import LoginManager, UserMixin, login_user, login_required, logout_user, current_user
from flask_wtf.csrf import CSRFProtect
from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField, SubmitField, BooleanField
import bcrypt

app = Flask(__name__)

# session を使う際に SECRET_KEY を設定
app.config['SECRET_KEY'] = 'secret_key'

# CSRF 対策
csrf = CSRFProtect(app)

# データベースに MySQL を使う
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://root:password@localhost/sample'

# SQLALchemy を初期化
db = SQLAlchemy(app)

# User というクラス(データベースにテーブル情報)
class User(UserMixin, db.Model):
    __tablename__ = 'User'
    id = db.Column(db.Integer, primary_key=True)
    mail = db.Column(db.String(255), unique=True)
    password = db.Column(db.String(255), unique=True)

    def __init__(self, mail, password):
        self.mail = mail
        self.password = password

class LoginForm(FlaskForm):
    mail = StringField('mail')
    password = PasswordField('Password')
    remember_me = BooleanField('Remember Me')
    submit = SubmitField()

with app.app_context():
    # テーブルが存在しなければ自動で作成
    db.create_all()

    password = b"password"
    salt = b'$2b$12$ZaUSY.Z0X1sEzaEqMoKFtu'
    hashed = bcrypt.hashpw(password,salt) # パスワードのハッシュ化
    
    # DB 何もレコードがない場合 test ユーザーを作成する
    user = User.query.filter_by(mail='test').first()
    if user is None:
        testuser = User(mail='test', password=hashed)
        db.session.add(testuser)
        db.session.commit()

# <-- comment.login-manager
# flask-login の初期化
# クラスの作成
login_manager = LoginManager()
login_manager.init_app(app)
# /comment.login-manager -->

# <-- comment.login_required
# リダイレクトするビュー関数と、エラーメッセージを定義
# ビュー関数 login にリダイレクトするように指定する
login_manager.login_view = 'login'

# 未ログインユーザーにメッセージ表示
def localize_callback(*args, **kwarg):
    return 'このページにアクセスするには、ログインが必要です'

login_manager.localize_callback = localize_callback
# /comment.login_required -->


@login_manager.user_loader
def load_user(user_id):
    return User.query.get(int(user_id))

@app.route('/login', methods=['GET', 'POST'])
def login():
    form = LoginForm()

    # POST の処理
    if request.method == "POST":
        mail = request.form.get('mail')
        password = request.form.get('password')

        try:
            # Userテーブルから mail に一致するユーザを取得
            user = User.query.filter_by(mail=mail).one_or_none()
            
            if user is None:
                flash('ログインできませんでした。') 
                return redirect(url_for('login'))
            
            if bcrypt.checkpw(password.encode("utf8"), user.password.encode("utf8")):
                # ユーザー情報をセッションに格納
                login_user(user)
                return redirect('index')
            else:
                flash('ログインできませんでした。') 
                return redirect(url_for('login'))
        except NoResultFound:
            flash('データがありません。') 
            return redirect(url_for('login'))
        except MultipleResultsFound:
            flash('データが複数あります。') 
            return redirect(url_for('login'))
    else:
        # GET の処理
        return render_template('login.html', form = form)

# ログアウト処理
@app.route('/logout')
@login_required
def logout():
    logout_user()
    flash('ログアウトしました。') 
    return redirect(url_for('login'))

# 認証が必要なページ
@app.route('/index')
@login_required
def index():
    return render_template('index.html')

if __name__ == '__main__':
    app.run(debug=True)

login.html

<!DOCTYPE html>
<html lang="en">
<head>
  <meta charset="UTF-8">
  <meta name="viewport" content="width=device-width, initial-scale=1.0">
  <title>Document</title>
</head>
<body>
  <ul>
    {% with messages = get_flashed_messages() %}
      {% for message in messages %}
        <li>{{ message }}</li>
      {% endfor %}
    {% endwith %}
  </ul>
  
  <form action="/login" method="post">
    <input type="hidden" name="csrf_token" value="{{ csrf_token() }}"/>
    <div>
      <p>mail <input type="text" name="mail"></p>
      <p>password <input type="password" name="password"></p>
      <input type="submit">
    </div>
  </form>
</body>
</html>

index.html

<!DOCTYPE html>
<html lang="en">
<head>
  <meta charset="UTF-8">
  <meta name="viewport" content="width=device-width, initial-scale=1.0">
  <title>Document</title>
</head>
<body>
  <h1>flask app/index</h1>
  <p><a href="/logout">ログアウト</a></p>
  <p>{{ message }}</p>
</body>
</html>

@login_required デコレータ

login_required をインポートするには以下のようにします。

from flask_login import login_required

@login_required をつけるとログインしていない場合エラーになります。

Flask 内部での処理の順番は以下の通りです。

@login_manager.user_loader が呼び出される
@login_manager.user_loader の中で user_id の値が渡されるので、user_id をもとにユーザー情報を取得して返す
@login_manager.ueer_loader からのレスポンスが None の場合はログインをしていないとして @login_manager.unauthorized_handler が呼び出される

@login_required を付けることで、未ログインユーザーはログインページに転送できます。

@app.route('/index')
@login_required
def index():
    return render_template('index.html')

また、転送された際のエラーメッセージを設定することができます。

# ビュー関数 login にリダイレクトするように指定する
login_manager.login_view = 'login'

# 未ログインユーザーにメッセージ表示
def localize_callback(*args, **kwarg):
    return 'このページにアクセスするには、ログインが必要です'

login_manager.localize_callback = localize_callback

User モデル

Flask-Login 拡張機能は、アプリケーションのクラス 'User' が定義され、特定のプロパティとメソッドがアプリケーションに実装されます。

メソッド	内容
is_authenticated	ユーザーが有効な資格情報を持っている場合は True、そうでない場合は False
is_active	ユーザーのアカウントがアクティブな場合は True、それ以外の場合は False
is_anonymous	通常のユーザーの場合は False、ゲストユーザーの場合は True
get_id()	ユーザーで重複しない識別子を文字列として返す

以下のようにすると、取得したレコードの ID が取得できます。

user = User.query.filter_by(mail=mail).one_or_none()
  print(user.get_id())

loader_user()

コールバック関数の loader_user() を定義しておく必要があります。

定義しておかないとエラーになります。

このコールバック関数は、セッションに保存されているユーザーIDからユーザーオブジェクトを読み込みするために使用します。

ユーザIDに該当するユーザが存在しない場合は、この関数が None を返します。

def load_user(user_id):
    return User.query.get(int(user_id))