想ひ出16: Flask1.0のチュートリアルをMySQLで対応する

おはようございます、moqrinです。

GCPのドキュメントを読んでいると、サンプルのPythonアプリケーションでFlaskというフレームワークを使っています。
わたくしは時折、PHPやJava、Javascriptのフレームワークで何か書く場合がありますが、
インフラ界隈ではPythonを使うべき機会が多いように思われます。
Pythonと仲良しになる上でも、サンプルのアプリケーションでクラウドサービスを試用する上でも、ちょいとFlaskをつまみ食いしようと思いました。
ということで、Flaskのチュートリアルをやってみました。
バージョンは1.0で、公式ではSQLite3で対応していますが、ここではMySQLで対応します。

チュートリアルを元にしていますが、随所で変更を入れています。
元の内容をベースに変更点や注意点を中心に記載しております。
疑問点や詳細の解説は参考URLや公式ドキュメントを確認頂ければと思います。

やること

1. 準備
2. Project Layout
3. Application Setup
4. Define and Access the Database
5. Blueprints and Views
6. Templates
7. Static Files
8. Blog Blueprint

1. 準備

データベースの接続に必要なモノをローカルにpipインストールしていきます。
(環境として、MacをローカルとしてPython3.X系のvirtualenvなりpyenvで考えています。)

pip install flask-sqlalchemy mysqlclient

すると、エラーが起きます。。わたしは起きました。
こんな感じのエラーです。

Command "python setup.py egg_info" failed with error code 1

結論、下記のように対応します。
mysql_configの場所を確認。

which mysql_config
vi /bin/mysql_config

とりあえず下記のようにmysql_configを編集。

libs="-L$pkglibdir"
#libs="$libs -l "
libs="$libs -lmysqlclient -lssl -lcrypto"

参考URL:

mysqlclientをインストールしようとすると…

2. Project Layout

では順番通り進めていきます。
プロジェクトのレイアウトは以下のようになります。
この例ではマイグレーションをかますため、若干変更しています。
なお、微妙にディレクトリ名も異なっています。

/home/user/Projects/flaskr
├── app/
│   ├── __init__.py
│   ├── models.py
│   ├── auth.py
│   ├── blog.py
│   ├── templates/
│   │   ├── base.html
│   │   ├── auth/
│   │   │   ├── login.html
│   │   │   └── register.html
│   │   └── blog/
│   │       ├── create.html
│   │       ├── index.html
│   │       └── update.html
│   └── static/
│       └── style.css
├── instance/
│   ├── config.py
├──  run.py
└── .gitignore

3. Application Setup

接続するDB情報をセットアップします。
わたくしはDockerのMySQL5.7で対応しました。

# instance/config.py
SECRET_KEY = 'secret'
SQLALCHEMY_DATABASE_URI = 'mysql://root:passwd@127.0.0.1/flask_db'
SQLALCHEMY_TRACK_MODIFICATIONS = True
# app/__init__.py

import os

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

def create_app(test_config=None):
    # create and configure the app
    app = Flask(__name__, instance_relative_config=True)

    if test_config is None:
        # load the instance config, if it exists, when not testing
        app.config.from_pyfile('config.py', silent=True)
    else:
        # load the test config if passed in
        app.config.from_mapping(test_config)

    db.init_app(app)

    # ensure the instance folder exists
    try:
        os.makedirs(app.instance_path)
    except OSError:
        pass

    # a simple page that says hello
    @app.route('/hello')
    def hello():
        return 'Hello, World!'

    return app

変更点として、SQLAlchemyを用いてデータベースに対して接続します。
以下で起動させます。

export FLASK_APP=run
export FLASK_ENV=development
flask run

4. Define and Access the Database

このパートは本家とは異なります。
ここを参考としています。(非常によく出来たチュートリアルです!こんなクソブログ見ている場合じゃありませんw)

後ほどログイン機能を作るので、以下インストールします。
使い方はここの通りです。

pip install flask-login

init.pyに追記します。

# app/__init__.py
...
from flask_login import LoginManager
login_manager = LoginManager()

...

def create_app(config_name):
    ...

    login_manager.init_app(app)
    login_manager.login_message = "You must be logged in to access this page."
    login_manager.login_view = "auth.login"

モデルを作成します。

# app/models.py

from app import db, login_manager
from flask_login import UserMixin

# UserMixinは入れておかないと怒られます
class User(UserMixin, db.Model):

    __tablename__ = 'user'

    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(60), nullable=False, unique=True)
    password = db.Column(db.String(128), nullable=False)


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

class Post(db.Model):

    __tablename__ = 'post'

    id = db.Column(db.Integer, primary_key=True)
    author_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)
    created = db.Column(db.TIMESTAMP, nullable=False)
    title = db.Column(db.String(200), nullable=False)
    body = db.Column(db.String(200), nullable=False)


マイグレーションの準備をします。
以下インストールします。

pip install flask-migrate

使い方はここの通りです。

init.pyに追記します。

# app/__init__.py
...
from flask_migrate import Migrate

...

def create_app(config_name):
    ...

    migrate = Migrate(app, db)
    from app import models

そしたら下記コマンドでマイグレーション対応完了です。

flask db init
flask db migrate
flask db upgrade

これで、DBに必要なテーブルが作成されているはずです。
それでは画面とルーティングに進みます。

5. Blueprints and Views

Flaskでは、画面とルーティングをBlueprintsという枠組みで作っていくようです。
各機能ごとで独立させたBlueprintを作成して、それを大元のファイルに読み込ませるというもののようです。
では公式の通り、ログイン機能となるauthファイルから。
後ほどブログ機能を実装します。

# app/auth.py

from flask import (
    Blueprint, flash, g, redirect, render_template, request, session, url_for
)
from werkzeug.security import check_password_hash, generate_password_hash
from . import db
from .models import User

from flask_login import login_required, login_user, logout_user

bp = Blueprint('auth', __name__, url_prefix='/auth')

init.pyにauthのblueprintを登録します。

# app/__init__.py

def create_app():
    app = ...

    # existing code omitted

    from . import auth
    app.register_blueprint(auth.bp)

    return app

ルーティングと画面の設定をしましょう〜。
まずはユーザー登録画面です。

# app/auth.py

@bp.route('/register', methods=('GET', 'POST'))
def register():
    if request.method == 'POST':
        username = request.form['username']
        password = request.form['password']
        password = generate_password_hash(password)
        error = None
        user = User(username=username,
                    password=password)

        if not username:
            error = 'Username is required.'
        elif not password:
            error = 'Password is required.'

        if error is None:
            db.session.add(user)
            db.session.commit()
            flash('Successfully added a new user.')
            return redirect(url_for('auth.login'))

        flash(error)

    return render_template('auth/register.html')

お次はログイン画面です。

# app/auth.py

@bp.route('/login', methods=('GET', 'POST'))
def login():
    if request.method == 'POST':
        username = request.form['username']
        password = request.form['password']
        error = None
        user = User.query.filter_by(username=username).first()

        if user is None:
            error = 'Incorrect username.'
        elif not check_password_hash(user.password, password):
            error = 'Incorrect password.'

        if error is None:
            session.clear()
            session['user_id'] = user.id
            login_user(user)
            return redirect(url_for('index'))

        flash(error)

    return render_template('auth/login.html')


セッション情報を取得しておくメソッドになります。

# app/auth.py

@bp.before_app_request
def load_logged_in_user():
    user_id = session.get('user_id')

    if user_id is None:
        g.user = None
    else:
        g.user = User.query.get(user_id)

そしてログアウト機能。

# app/auth.py

@bp.route('/logout')
@login_required
def logout():
    session.clear()
    logout_user()
    flash('Logged out.')

    return redirect(url_for('auth.login'))

6. Templates

テンプレートはほぼそのままです。

The Base Layout

# app/templates/base.html

<!doctype html>
<title>{% block title %}{% endblock %} - Flaskr</title>
<link rel="stylesheet" href="{{ url_for('static', filename='style.css') }}">
<nav>
    <h1><a href="{{ url_for('index') }}">Flaskr</a></h1>
    <ul>
        {% if current_user.is_authenticated %}
        <li><span>{{ current_user.username }}</span>
        <li><a href="{{ url_for('auth.logout') }}">Log Out</a>
            {% else %}
        <li><a href="{{ url_for('auth.register') }}">Register</a>
        <li><a href="{{ url_for('auth.login') }}">Log In</a>
            {% endif %}
    </ul>
</nav>
<section class="content">
    <header>
        {% block header %}{% endblock %}
    </header>
    {% for message in get_flashed_messages() %}
    <div class="flash">{{ message }}</div>
    {% endfor %}
    {% block content %}{% endblock %}
</section>

Register

# app/templates/auth/register.htm

{% extends 'base.html' %}

{% block header %}
<h1>{% block title %}Register{% endblock %}</h1>
{% endblock %}

{% block content %}
<form method="post">
    <label for="username">Username</label>
    <input name="username" id="username" required>
    <label for="password">Password</label>
    <input type="password" name="password" id="password" required>
    <input type="submit" value="Register">
</form>
{% endblock %}

Log In

# app/templates/auth/login.html

{% extends 'base.html' %}

{% block header %}
<h1>{% block title %}Log In{% endblock %}</h1>
{% endblock %}

{% block content %}
<form method="post">
    <label for="username">Username</label>
    <input name="username" id="username" required>
    <label for="password">Password</label>
    <input type="password" name="password" id="password" required>
    <input type="submit" value="Log In">
</form>
{% endblock %}

7. Static Files

そのままです。

8. Blog Blueprint

ブログ機能のBlueprintになります。

# app/blog.py

from flask import (
    Blueprint, flash, g, redirect, render_template, request, url_for
)
from flask_login import current_user, login_required

from . import db
from .models import Post

bp = Blueprint('blog', __name__)

init.pyにauthのblueprintを登録します。

# app/__init__.py

def create_app():
    app = ...
    # existing code omitted

    from . import blog
    app.register_blueprint(blog.bp)
    app.add_url_rule('/', endpoint='index')

    return app

では、ルーティングと画面の設定をしましょう。

Index

# app/blog.py

@bp.route('/', methods=['GET', 'POST'])
@login_required
def index():
    posts = db.session.query(Post). \
        filter(Post.author_id == g.user.id). \
        all()
    return render_template('blog/index.html', posts=posts)

# app/templates/blog/index.html
{% extends 'base.html' %}

{% block header %}
<h1>{% block title %}Posts{% endblock %}</h1>
{% if current_user.is_authenticated %}
<a class="action" href="{{ url_for('blog.create') }}">New</a>
{% endif %}
{% endblock %}

{% block content %}
{% for post in posts %}
<article class="post">
    <header>
        <div>
            <h1>{{ post.title }}</h1>
            <div class="about">by {{ post.username }} on {{ post.created.strftime('%Y-%m-%d') }}</div>
        </div>
        <a class="action" href="{{ url_for('blog.update', id=post.id) }}">Edit</a>
    </header>
    <p class="body">{{ post.body }}</p>
</article>
{% if not loop.last %}
<hr>
{% endif %}
{% endfor %}
{% endblock %}

Create

# app/blog.py

@bp.route('/create', methods=('GET', 'POST'))
@login_required
def create():
    if request.method == 'POST':
        title = request.form['title']
        body = request.form['body']
        error = None

        if not title:
            error = 'Title is required.'

        if error is not None:
            flash(error)
        else:
            post = Post(title=title,
                        body=body,
                        author_id=g.user.id)

            db.session.add(post)
            db.session.commit()
            flash('Successfully added a new post.')

            return redirect(url_for('blog.index'))

    return render_template('blog/create.html')

# app/templates/blog/create.html

{% extends 'base.html' %}

{% block header %}
<h1>{% block title %}New Post{% endblock %}</h1>
{% endblock %}

{% block content %}
<form method="post">
    <label for="title">Title</label>
    <input name="title" id="title" value="{{ request.form['title'] }}" required>
    <label for="body">Body</label>
    <textarea name="body" id="body">{{ request.form['body'] }}</textarea>
    <input type="submit" value="Save">
</form>
{% endblock %}

Update

# app/blog.py

@bp.route('/<int:id>/update', methods=('GET', 'POST'))
@login_required
def update(id):

    post = Post.query.get_or_404(id)

    if request.method == 'POST':
        title = request.form['title']
        body = request.form['body']
        error = None

        if not title:
            error = 'Title is required.'

        if error is not None:
            flash(error)
        else:
            post.id = id
            post.title = title
            post.body = body
            db.session.commit()
            flash('Successfully edited the post.')

        return redirect(url_for('blog.index'))

    return render_template('blog/update.html', post=post)

# app/templates/blog/update.html

{% extends 'base.html' %}

{% block header %}
<h1>{% block title %}Edit "{{ post['title'] }}"{% endblock %}</h1>
{% endblock %}

{% block content %}
<form method="post">
    <label for="title">Title</label>
    <input name="title" id="title"
           value="{{ request.form['title'] or post['title'] }}" required>
    <label for="body">Body</label>
    <textarea name="body" id="body">{{ request.form['body'] or post['body'] }}</textarea>
    <input type="submit" value="Save">
</form>
<hr>
<form action="{{ url_for('blog.delete', id=post['id']) }}" method="post">
    <input class="danger" type="submit" value="Delete" onclick="return confirm('Are you sure?');">
</form>
{% endblock %}

Delete

# app/blog.py

@bp.route('/<int:id>/delete', methods=('POST',))
@login_required
def delete(id):
    post = Post.query.get_or_404(id)
    db.session.delete(post)
    db.session.commit()
    flash('Successfully deleted the post.')
    return redirect(url_for('blog.index'))

最後にrun.pyを作成しておきましょう。
次回で使います。

# app/run.py

import os

from app import create_app

config_name = os.getenv('FLASK_CONFIG')
app = create_app()


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

以上です!
とりあえず、CRUDシステムが完了したのでここまでとしました。
続きはデプロイしちゃいます!

参考:

Build a CRUD Web App With Python and Flask
Flask-Login
Flask-Migrate

シェアする

  • このエントリーをはてなブックマークに追加

フォローする

%d人のブロガーが「いいね」をつけました。