ウェブサイト検索

実稼働ビルド用に Flask-RESTPlus Web サービスを構築する方法

グレッグ・オビンナ著

このガイドでは、テスト、開発、運用環境向けに Flask RESTPlus Web アプリケーションを構築するための段階的なアプローチを示します。 Linux ベースの OS (Ubuntu) を使用しますが、ほとんどの手順は Windows と Mac で再現できます。

このガイドを続ける前に、Python プログラミング言語と Flask マイクロ フレームワークについての基本を理解しておく必要があります。これらに詳しくない場合は、入門記事「Python と Flask を使用して Web アプリを構築する方法」を参照することをお勧めします。

このガイドの構成

このガイドは次の部分に分かれています。

  • 特徴
  • Flask-RESTPlus とは何ですか?
  • セットアップとインストール
  • プロジェクトのセットアップと組織化
  • 構成設定
  • フラスコスクリプト
  • データベースモデルと移行
  • テスト
  • 構成
  • ユーザー操作
  • セキュリティと認証
  • ルートの保護と認可
  • 追加のヒント
  • アプリの拡張と結論

特徴

プロジェクト内では次の機能と拡張機能を使用します。

  • Flask-Bcrypt: アプリケーションに bcrypt ハッシュ ユーティリティを提供するFlask 拡張機能です。
  • Flask-Migrate:Alembic を使用して Flask アプリケーションの SQLAlchemy データベース移行を処理する拡張機能。データベース操作は、Flask コマンドライン インターフェイスまたは Flask-Script 拡張機能を通じて利用可能になります。
  • Flask-SQLAlchemy: アプリケーションに SQLAlchemy のサポートを追加する Flask の拡張機能。
  • PyJWT:JSON Web トークン (JWT) をエンコードおよびデコードできる Python ライブラリ。 JWT は、 二者間のクレームを安全に表現するためのオープンな業界標準 (RFC 7519) です。
  • Flask-Script:Flask での外部スクリプトの作成や、Web アプリケーション自体の外部に属するその他のコマンドライン タスクのサポートを提供する拡張機能。
  • 名前空間 (ブループリント)
  • フラスコレストプラス
  • 単体テスト

Flask-RESTPlus とは何ですか?

Flask-RESTPlus は、REST API を迅速に構築するためのサポートを追加する Flask の拡張機能です。 Flask-RESTPlus は、最小限のセットアップでのベスト プラクティスを推奨します。これは、API を記述し、そのドキュメントを (Swagger を使用して) 適切に公開するための、一貫したデコレーターとツールのコレクションを提供します。

セットアップとインストール

コマンド pip --version をターミナルに入力し、Enter キーを押して、pip がインストールされているかどうかを確認します。

pip --version

ターミナルがバージョン番号を応答した場合は、pip がインストールされていることを意味するため、次のステップに進みます。それ以外の場合は、pip をインストールするか、Linux パッケージ マネージャーを使用して、ターミナルで以下のコマンドを実行し、Enter キーを押します。 Python 2.x または 3.x バージョンを選択します。

  • Python 2.x
sudo apt-get install python-pip
  • Python 3.x
sudo apt-get install python3-pip

仮想環境と仮想環境ラッパーをセットアップします (上でインストールされたバージョンに応じて、これらのいずれか 1 つだけが必要です)。

sudo pip install virtualenv

sudo pip3 install virtualenvwrapper

仮想環境ラッパーの完全なセットアップについては、このリンクに従ってください。

新しい環境を作成し、ターミナルで次のコマンドを実行してアクティブ化します。

mkproject name_of_your_project

プロジェクトのセットアップと組織化

関数構造を使用して、プロジェクトのファイルをその内容ごとに整理します。機能的な構造では、テンプレートは 1 つのディレクトリにグループ化され、静的ファイルは別のディレクトリに、ビューは 3 番目のディレクトリにグループ化されます。

プロジェクト ディレクトリに、app という名前の新しいパッケージを作成します。 app 内に、main と test という 2 つのパッケージを作成します。ディレクトリ構造は以下のようになります。

.
├── app
│   ├── __init__.py
│   ├── main
│   │   └── __init__.py
│   └── test
│       └── __init__.py
└── requirements.txt

関数構造を使用してアプリケーションをモジュール化します。
main パッケージ内に、さらに 3 つのパッケージ、controller、service を作成します。 > と モデル。 model パッケージにはすべてのデータベース モデルが含まれ、service パッケージにはアプリケーションのすべてのビジネス ロジックが含まれ、最後に controller パッケージにはすべてのアプリケーション エンドポイントが含まれています。ツリー構造は次のようになります。

.
├── app
│   ├── __init__.py
│   ├── main
│   │   ├── controller
│   │   │   └── __init__.py
│   │   ├── __init__.py
│   │   ├── model
│   │   │   └── __init__.py
│   │   └── service
│   │       └── __init__.py
│   └── test
│       └── __init__.py
└── requirements.txt

次に、必要なパッケージをインストールしましょう。作成した仮想環境がアクティブ化されていることを確認し、ターミナルで次のコマンドを実行します。

pip install flask-bcrypt

pip install flask-restplus

pip install Flask-Migrate

pip install pyjwt

pip install Flask-Script

pip install flask_testing

次のコマンドを実行して、requirements.txt ファイルを作成または更新します。

pip freeze > requirements.txt

生成された requirements.txt ファイルは次のようになります。

alembic==0.9.8
aniso8601==3.0.0
bcrypt==3.1.4
cffi==1.11.5
click==6.7
Flask==0.12.2
Flask-Bcrypt==0.7.1
Flask-Migrate==2.1.1
flask-restplus==0.10.1
Flask-Script==2.0.6
Flask-SQLAlchemy==2.3.2
Flask-Testing==0.7.1
itsdangerous==0.24
Jinja2==2.10
jsonschema==2.6.0
Mako==1.0.7
MarkupSafe==1.0
pycparser==2.18
PyJWT==1.6.0
python-dateutil==2.7.0
python-editor==1.0.3
pytz==2018.3
six==1.11.0
SQLAlchemy==1.2.5
Werkzeug==0.14.1

構成設定

main パッケージで、次の内容を含む config.py というファイルを作成します。

import os

# uncomment the line below for postgres database url from environment variable
# postgres_local_base = os.environ['DATABASE_URL']

basedir = os.path.abspath(os.path.dirname(__file__))

class Config:
    SECRET_KEY = os.getenv('SECRET_KEY', 'my_precious_secret_key')
    DEBUG = False


class DevelopmentConfig(Config):
    # uncomment the line below to use postgres
    # SQLALCHEMY_DATABASE_URI = postgres_local_base
    DEBUG = True
    SQLALCHEMY_DATABASE_URI = 'sqlite:///' + os.path.join(basedir, 'flask_boilerplate_main.db')
    SQLALCHEMY_TRACK_MODIFICATIONS = False


class TestingConfig(Config):
    DEBUG = True
    TESTING = True
    SQLALCHEMY_DATABASE_URI = 'sqlite:///' + os.path.join(basedir, 'flask_boilerplate_test.db')
    PRESERVE_CONTEXT_ON_EXCEPTION = False
    SQLALCHEMY_TRACK_MODIFICATIONS = False


class ProductionConfig(Config):
    DEBUG = False
    # uncomment the line below to use postgres
    # SQLALCHEMY_DATABASE_URI = postgres_local_base


config_by_name = dict(
    dev=DevelopmentConfig,
    test=TestingConfig,
    prod=ProductionConfig
)

key = Config.SECRET_KEY

構成ファイルには、testing、development、および production を含む 3 つの環境セットアップ クラスが含まれています。

Flask オブジェクトの作成にはアプリケーション ファクトリ パターンを使用します。このパターンは、異なる設定でアプリケーションの複数のインスタンスを作成する場合に最も役立ちます。これにより、必須パラメータを指定して create_app 関数を呼び出すことで、テスト、開発、実稼働環境を簡単に切り替えることができます。

main パッケージ内の __init__.py ファイルに、次のコード行を入力します。

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_bcrypt import Bcrypt

from .config import config_by_name

db = SQLAlchemy()
flask_bcrypt = Bcrypt()


def create_app(config_name):
    app = Flask(__name__)
    app.config.from_object(config_by_name[config_name])
    db.init_app(app)
    flask_bcrypt.init_app(app)

    return app

フラスコスクリプト

次に、アプリケーションのエントリ ポイントを作成しましょう。プロジェクトのルート ディレクトリに、次の内容を含む manage.py というファイルを作成します。

import os
import unittest

from flask_migrate import Migrate, MigrateCommand
from flask_script import Manager

from app.main import create_app, db

app = create_app(os.getenv('BOILERPLATE_ENV') or 'dev')

app.app_context().push()

manager = Manager(app)

migrate = Migrate(app, db)

manager.add_command('db', MigrateCommand)

@manager.command
def run():
    app.run()

@manager.command
def test():
    """Runs the unit tests."""
    tests = unittest.TestLoader().discover('app/test', pattern='test*.py')
    result = unittest.TextTestRunner(verbosity=2).run(tests)
    if result.wasSuccessful():
        return 0
    return 1

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

manage.py 内の上記のコードは次のことを行います。

  • 4 行目 と 5 は、それぞれ移行モジュールとマネージャー モジュールをインポートします (移行コマンドはすぐに使用します)。
  • 9 行目 は、最初に作成した create_app 関数を呼び出して、次のいずれかの環境変数から必要なパラメータを使用してアプリケーション インスタンスを作成します - dev、本番、テスト。環境変数に何も設定されていない場合は、デフォルトの dev が使用されます。
  • 13 行目 と 15 は、app インスタンスをそれぞれのコンストラクターに渡すことによってマネージャーをインスタンス化し、クラスを移行します。
  • 17 行目 では、db インスタンスと MigrateCommand インスタンスを manager< の add_command インターフェイスに渡します。 /code>Flask-Script を通じてすべてのデータベース移行コマンドを公開します。
  • line 20 と 25 は、2 つの関数をコマンド ラインから実行可能としてマークします。

Flask-Migrate は、Migrate と MigrateCommand という 2 つのクラスを公開します。 Migrate クラスには、拡張機能のすべての機能が含まれています。 MigrateCommand クラスは、Flask-Script 拡張機能を通じてデータベース移行コマンドを公開する必要がある場合にのみ使用されます。

この時点で、プロジェクトのルート ディレクトリで以下のコマンドを実行してアプリケーションをテストできます。

python manage.py run

すべてが正常であれば、次のような内容が表示されるはずです。

データベースモデルと移行

それでは、モデルを作成しましょう。 sqlalchemy の db インスタンスを使用してモデルを作成します。

db インスタンスには、sqlalchemy と の両方の関数とヘルパーがすべて含まれています。 sqlalchemy.orm <というクラスを提供します。Model は、モデルを宣言するために使用できる宣言ベースです。

model パッケージ内に、次の内容を含む user.py というファイルを作成します。

from .. import db, flask_bcrypt

class User(db.Model):
    """ User Model for storing user related details """
    __tablename__ = "user"

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    email = db.Column(db.String(255), unique=True, nullable=False)
    registered_on = db.Column(db.DateTime, nullable=False)
    admin = db.Column(db.Boolean, nullable=False, default=False)
    public_id = db.Column(db.String(100), unique=True)
    username = db.Column(db.String(50), unique=True)
    password_hash = db.Column(db.String(100))

    @property
    def password(self):
        raise AttributeError('password: write-only field')

    @password.setter
    def password(self, password):
        self.password_hash = flask_bcrypt.generate_password_hash(password).decode('utf-8')

    def check_password(self, password):
        return flask_bcrypt.check_password_hash(self.password_hash, password)

    def __repr__(self):
        return "<User '{}'>".format(self.username)

user.py 内の上記のコードは次のことを行います。

  • 行 3: user クラスは、クラスを sqlalchemy のモデルとして宣言する db.Model クラスを継承します。
  • 7 行目 から 13 までは、ユーザー テーブルに必要な列を作成します。
  • 行 21 はフィールド password_hash のセッターであり、flask-bcrypt を使用して指定されたパスワードを使用してハッシュを生成します。
  • 行 24 は、指定されたパスワードをすでに保存されているpassword_hash と比較します。

ここで、作成したばかりの user モデルからデータベース テーブルを生成するには、manager インターフェイスを通じて mergeCommand を使用します。 manager がモデルを検出するには、以下のコードを manage.py ファイルに追加して、user モデルをインポートする必要があります。

...
from app.main.model import user
...

これで、プロジェクトのルート ディレクトリで次のコマンドを実行して移行の実行に進むことができます。

  1. Alembic が移行を実行するには、init コマンドを使用して移行フォルダーを開始します。
python manage.py db init

2. merge コマンドを使用して、モデル内で検出された変更から移行スクリプトを作成します。これはまだデータベースには影響しません。

python manage.py db migrate --message 'initial database migration'

3. upgrade コマンドを使用して、移行スクリプトをデータベースに適用します。

python manage.py db upgrade

すべてが正常に実行されると、メイン パッケージ内に新しい sqlLite データベース
flask_boilerplate_main.db ファイルが生成されるはずです。

データベース モデルが変更されるたびに、merge コマンドと upgrade コマンドを繰り返します。

テスト

構成

環境構成のセットアップが機能していることを確認するために、いくつかのテストを作成してみましょう。

以下の内容を含む、テスト パッケージ内に test_config.py という名前のファイルを作成します。

import os
import unittest

from flask import current_app
from flask_testing import TestCase

from manage import app
from app.main.config import basedir


class TestDevelopmentConfig(TestCase):
    def create_app(self):
        app.config.from_object('app.main.config.DevelopmentConfig')
        return app

    def test_app_is_development(self):
        self.assertFalse(app.config['SECRET_KEY'] is 'my_precious')
        self.assertTrue(app.config['DEBUG'] is True)
        self.assertFalse(current_app is None)
        self.assertTrue(
            app.config['SQLALCHEMY_DATABASE_URI'] == 'sqlite:///' + os.path.join(basedir, 'flask_boilerplate_main.db')
        )


class TestTestingConfig(TestCase):
    def create_app(self):
        app.config.from_object('app.main.config.TestingConfig')
        return app

    def test_app_is_testing(self):
        self.assertFalse(app.config['SECRET_KEY'] is 'my_precious')
        self.assertTrue(app.config['DEBUG'])
        self.assertTrue(
            app.config['SQLALCHEMY_DATABASE_URI'] == 'sqlite:///' + os.path.join(basedir, 'flask_boilerplate_test.db')
        )


class TestProductionConfig(TestCase):
    def create_app(self):
        app.config.from_object('app.main.config.ProductionConfig')
        return app

    def test_app_is_production(self):
        self.assertTrue(app.config['DEBUG'] is False)


if __name__ == '__main__':
    unittest.main()

以下のコマンドを使用してテストを実行します。

python manage.py test

次の出力が得られるはずです。

ユーザー操作

次に、次のユーザー関連の操作に取り組んでみましょう。

  • 新しいユーザーを作成する
  • public_id を使用して登録ユーザーを取得する
  • すべての登録ユーザーを取得します。

User Service クラス: このクラスは、ユーザー モデルに関連するすべてのロジックを処理します。
service パッケージで、新しいファイル user_service.py を作成します。 には次の内容が含まれます。

import uuid
import datetime

from app.main import db
from app.main.model.user import User


def save_new_user(data):
    user = User.query.filter_by(email=data['email']).first()
    if not user:
        new_user = User(
            public_id=str(uuid.uuid4()),
            email=data['email'],
            username=data['username'],
            password=data['password'],
            registered_on=datetime.datetime.utcnow()
        )
        save_changes(new_user)
        response_object = {
            'status': 'success',
            'message': 'Successfully registered.'
        }
        return response_object, 201
    else:
        response_object = {
            'status': 'fail',
            'message': 'User already exists. Please Log in.',
        }
        return response_object, 409


def get_all_users():
    return User.query.all()


def get_a_user(public_id):
    return User.query.filter_by(public_id=public_id).first()


def save_changes(data):
    db.session.add(data)
    db.session.commit()

user_service.py 内の上記のコードは次のことを行います。

  • 8 行目 から 29 行目 では、最初にユーザーが既に存在するかどうかを確認して新しいユーザーを作成します。ユーザーが存在しない場合は成功 response_object を返し、それ以外の場合はエラー コード 409 を返し、失敗 response_object を返します。
  • 33 行目 と 37 は、それぞれ public_id を指定して、すべての登録済みユーザーのリストとユーザー オブジェクトを返します。
  • 40 行目 から 42 までは、変更をデータベースにコミットします。

オブジェクトを JSON にフォーマットするために jsonify を使用する必要はありません。Flask-restplus が自動的に行います。

main パッケージで、 util という新しいパッケージを作成します。このパッケージには、アプリケーションで必要となる可能性のあるすべてのユーティリティが含まれています。

util パッケージで、新しいファイル dto.py を作成します。名前が示すように、データ転送オブジェクト (DTO) はプロセス間でデータを転送する役割を果たします。私たち自身のケースでは、API 呼び出しのデータをマーシャリングするために使用されます。このことは、進めていくうちにさらに理解できるようになります。

from flask_restplus import Namespace, fields


class UserDto:
    api = Namespace('user', description='user related operations')
    user = api.model('user', {
        'email': fields.String(required=True, description='user email address'),
        'username': fields.String(required=True, description='user username'),
        'password': fields.String(required=True, description='user password'),
        'public_id': fields.String(description='user Identifier')
    })

dto.py 内の上記のコードは次のことを行います。

  • 行 5 は、ユーザー関連の操作用の新しい名前空間を作成します。 Flask-RESTPlus は、Blueprint とほぼ同じパターンを使用する方法を提供します。主なアイデアは、アプリを再利用可能な名前空間に分割することです。名前空間モジュールにはモデルとリソース宣言が含まれます。
  • 行 6 では、行 5 の api 名前空間によって提供される model インターフェイスを介して新しいユーザー dto を作成します。

ユーザー コントローラー: ユーザー コントローラー クラスは、ユーザーに関連するすべての受信 HTTP リクエストを処理します。

controller パッケージの下に、次の内容を含む user_controller.py という新しいファイルを作成します。

from flask import request
from flask_restplus import Resource

from ..util.dto import UserDto
from ..service.user_service import save_new_user, get_all_users, get_a_user

api = UserDto.api
_user = UserDto.user


@api.route('/')
class UserList(Resource):
    @api.doc('list_of_registered_users')
    @api.marshal_list_with(_user, envelope='data')
    def get(self):
        """List all registered users"""
        return get_all_users()

    @api.response(201, 'User successfully created.')
    @api.doc('create a new user')
    @api.expect(_user, validate=True)
    def post(self):
        """Creates a new User """
        data = request.json
        return save_new_user(data=data)


@api.route('/<public_id>')
@api.param('public_id', 'The User identifier')
@api.response(404, 'User not found.')
class User(Resource):
    @api.doc('get a user')
    @api.marshal_with(_user)
    def get(self, public_id):
        """get a user given its identifier"""
        user = get_a_user(public_id)
        if not user:
            api.abort(404)
        else:
            return user

1 行目 から 8 までは、ユーザー コントローラーに必要なすべてのリソースをインポートします。
ユーザー コントローラーに
userList という 2 つの具象クラスを定義しました。 と user。これら 2 つのクラスは、抽象 flask-restplus リソースを拡張します。

具体的なリソースはこのクラスから拡張する必要があり 、サポートされている各 HTTP メソッドのメソッドを公開する必要があります。 サポートされていない HTTP メソッドでリソースが呼び出された場合は、 >API はステータス 405 メソッドが許可されていない応答を返します。 それ以外の場合は、リソースを API に追加するときに使用される URL ルールから適切なメソッドが呼び出され、すべての引数が渡されます インスタンスです。

上記の line 7 の api 名前空間は、以下を含むいくつかのデコレーターをコントローラーに提供します。

  • api.route: リソースをルーティングするデコレーター
  • api.marshal_with:シリアル化に使用するフィールドを指定するデコレーター (ここで<コードを使用します) >userDto 以前に作成しました）
  • api.marshal_list_with: marshal_with のショートカット デコレータ上記とas_list=True
  • api.doc: 装飾されたオブジェクトに API ドキュメントを追加するデコレーター
  • api.response: 期待される応答の 1 つを指定するデコレータ
  • api.expect: 期待される入力モデルを指定するデコレータ ( を引き続き使用します) >userDto (予想される入力用)
  • api.param: 期待されるパラメータの 1 つを指定するデコレータ

これで、ユーザー コントローラーを使用して名前空間を定義しました。次に、これをアプリケーションのエントリ ポイントに追加します。

app パッケージの __init__.py ファイルに次のように入力します。

# app/__init__.py

from flask_restplus import Api
from flask import Blueprint

from .main.controller.user_controller import api as user_ns

blueprint = Blueprint('api', __name__)

api = Api(blueprint,
          title='FLASK RESTPLUS API BOILER-PLATE WITH JWT',
          version='1.0',
          description='a boilerplate for flask restplus web service'
          )

api.add_namespace(user_ns, path='/user')

blueprint.py 内の上記のコードは次のことを行います。

  • 8 行目 では、name と import_name を渡してブループリント インスタンスを作成します。 API は、そのメイン エントリ ポイントです。アプリケーション リソースなので、10 行目 の blueprint で初期化する必要があります。
  • 16 行目 では、ユーザー名前空間 user_ns を API インスタンスの名前空間のリストに追加します。

これでブループリントが定義されました。これを Flask アプリに登録します。
blueprint をインポートし、Flask アプリケーション インスタンスに登録することで、manage.py を更新します。

from app import blueprint
...

app = create_app(os.getenv('BOILERPLATE_ENV') or 'dev')
app.register_blueprint(blueprint)

app.app_context().push()

...

これでアプリケーションをテストして、すべてが正常に動作していることを確認できます。

python manage.py run

次に、ブラウザで URL http://127.0.0.1:5000 を開きます。 Swagger のドキュメントを参照してください。

Swagger テスト機能を使用して、新しいユーザーの作成 エンドポイントをテストしてみましょう。

次の応答が得られるはずです

セキュリティと認証

ブラックリストに登録されたトークンを保存するためのモデル blacklistToken を作成しましょう。 models パッケージで、次の内容を含む blacklist.py ファイルを作成します。

from .. import db
import datetime


class BlacklistToken(db.Model):
    """
    Token Model for storing JWT tokens
    """
    __tablename__ = 'blacklist_tokens'

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    token = db.Column(db.String(500), unique=True, nullable=False)
    blacklisted_on = db.Column(db.DateTime, nullable=False)

    def __init__(self, token):
        self.token = token
        self.blacklisted_on = datetime.datetime.now()

    def __repr__(self):
        return '<id: token: {}'.format(self.token)

    @staticmethod
    def check_blacklist(auth_token):
        # check whether auth token has been blacklisted
        res = BlacklistToken.query.filter_by(token=str(auth_token)).first()
        if res:
            return True
        else:
            return False

変更をデータベースに反映するために忘れずに移行してください。
manage.py に blacklist クラスをインポートします。

from app.main.model import blacklist

merge および upgrade コマンドを実行します。

python manage.py db migrate --message 'add blacklist table'

python manage.py db upgrade

次に、トークンをブラックリストに登録するための次の内容を含むサービス パッケージ内に blacklist_service.py を作成します。

from app.main import db
from app.main.model.blacklist import BlacklistToken


def save_token(token):
    blacklist_token = BlacklistToken(token=token)
    try:
        # insert the token
        db.session.add(blacklist_token)
        db.session.commit()
        response_object = {
            'status': 'success',
            'message': 'Successfully logged out.'
        }
        return response_object, 200
    except Exception as e:
        response_object = {
            'status': 'fail',
            'message': e
        }
        return response_object, 200

トークンをエンコードおよびデコードするための 2 つの静的メソッドを使用して user モデルを更新します。次のインポートを追加します。

import datetime
import jwt
from app.main.model.blacklist import BlacklistToken
from ..config import key
  • エンコーディング
def encode_auth_token(self, user_id):
        """
        Generates the Auth Token
        :return: string
        """
        try:
            payload = {
                'exp': datetime.datetime.utcnow() + datetime.timedelta(days=1, seconds=5),
                'iat': datetime.datetime.utcnow(),
                'sub': user_id
            }
            return jwt.encode(
                payload,
                key,
                algorithm='HS256'
            )
        except Exception as e:
            return e
  • デコード: 認証トークンをデコードするときに、ブラックリストに登録されたトークン、期限切れのトークン、および無効なトークンが考慮されます。
  @staticmethod  
  def decode_auth_token(auth_token):
        """
        Decodes the auth token
        :param auth_token:
        :return: integer|string
        """
        try:
            payload = jwt.decode(auth_token, key)
            is_blacklisted_token = BlacklistToken.check_blacklist(auth_token)
            if is_blacklisted_token:
                return 'Token blacklisted. Please log in again.'
            else:
                return payload['sub']
        except jwt.ExpiredSignatureError:
            return 'Signature expired. Please log in again.'
        except jwt.InvalidTokenError:
            return 'Invalid token. Please log in again.'

次に、user モデルのテストを作成して、encode 関数と decode 関数が適切に動作していることを確認しましょう。

test パッケージで、次の内容の base.py ファイルを作成します。

from flask_testing import TestCase
from app.main import db
from manage import app


class BaseTestCase(TestCase):
    """ Base Tests """

    def create_app(self):
        app.config.from_object('app.main.config.TestingConfig')
        return app

    def setUp(self):
        db.create_all()
        db.session.commit()

    def tearDown(self):
        db.session.remove()
        db.drop_all()

BaseTestCase は、テスト環境を拡張するすべてのテスト ケースの前後に、テスト環境を準備します。

次のテスト ケースを使用して test_user_medol.py を作成します。

import unittest
import datetime

from app.main import db
from app.main.model.user import User
from app.test.base import BaseTestCase


class TestUserModel(BaseTestCase):

    def test_encode_auth_token(self):
        user = User(
            email='test@test.com',
            password='test',
            registered_on=datetime.datetime.utcnow()
        )
        db.session.add(user)
        db.session.commit()
        auth_token = user.encode_auth_token(user.id)
        self.assertTrue(isinstance(auth_token, bytes))

    def test_decode_auth_token(self):
        user = User(
            email='test@test.com',
            password='test',
            registered_on=datetime.datetime.utcnow()
        )
        db.session.add(user)
        db.session.commit()
        auth_token = user.encode_auth_token(user.id)
        self.assertTrue(isinstance(auth_token, bytes))
        self.assertTrue(User.decode_auth_token(auth_token.decode("utf-8") ) == 1)


if __name__ == '__main__':
    unittest.main()

python manage.py test を使用してテストを実行します。すべてのテストに合格するはずです。

ログインとログアウトのための認証エンドポイントを作成しましょう。 >。

  • まず、ログイン ペイロード用の dto が必要です。 login エンドポイントの @expect アノテーションに auth dto を使用します。以下のコードを dto.py に追加します。
class AuthDto:
    api = Namespace('auth', description='authentication related operations')
    user_auth = api.model('auth_details', {
        'email': fields.String(required=True, description='The email address'),
        'password': fields.String(required=True, description='The user password '),
    })
  • 次に、すべての認証関連の操作を処理するための認証ヘルパー クラスを作成します。この auth_helper.py はサービス パッケージ内にあり、login_user と logout_user という 2 つの静的メソッドが含まれます。

ユーザーがログアウトすると、ユーザーのトークンはブラックリストに登録されます。つまり、ユーザーは同じトークンで再度ログインできなくなります。

from app.main.model.user import User
from ..service.blacklist_service import save_token


class Auth:

    @staticmethod
    def login_user(data):
        try:
            # fetch the user data
            user = User.query.filter_by(email=data.get('email')).first()
            if user and user.check_password(data.get('password')):
                auth_token = user.encode_auth_token(user.id)
                if auth_token:
                    response_object = {
                        'status': 'success',
                        'message': 'Successfully logged in.',
                        'Authorization': auth_token.decode()
                    }
                    return response_object, 200
            else:
                response_object = {
                    'status': 'fail',
                    'message': 'email or password does not match.'
                }
                return response_object, 401

        except Exception as e:
            print(e)
            response_object = {
                'status': 'fail',
                'message': 'Try again'
            }
            return response_object, 500

    @staticmethod
    def logout_user(data):
        if data:
            auth_token = data.split(" ")[1]
        else:
            auth_token = ''
        if auth_token:
            resp = User.decode_auth_token(auth_token)
            if not isinstance(resp, str):
                # mark the token as blacklisted
                return save_token(token=auth_token)
            else:
                response_object = {
                    'status': 'fail',
                    'message': resp
                }
                return response_object, 401
        else:
            response_object = {
                'status': 'fail',
                'message': 'Provide a valid auth token.'
            }
            return response_object, 403
  • 次に、login 操作と logout 操作のためのエンドポイントを作成しましょう。
    コントローラ パッケージで、次の内容を含む
    auth_controller.py を作成します。
from flask import request
from flask_restplus import Resource

from app.main.service.auth_helper import Auth
from ..util.dto import AuthDto

api = AuthDto.api
user_auth = AuthDto.user_auth


@api.route('/login')
class UserLogin(Resource):
    """
        User Login Resource
    """
    @api.doc('user login')
    @api.expect(user_auth, validate=True)
    def post(self):
        # get the post data
        post_data = request.json
        return Auth.login_user(data=post_data)


@api.route('/logout')
class LogoutAPI(Resource):
    """
    Logout Resource
    """
    @api.doc('logout a user')
    def post(self):
        # get auth token
        auth_header = request.headers.get('Authorization')
        return Auth.logout_user(data=auth_header)
  • この時点で残っているのは、認証 api 名前空間をアプリケーション Blueprint に登録することだけです。

app パッケージの __init__.py ファイルを次のように更新します。

# app/__init__.py

from flask_restplus import Api
from flask import Blueprint

from .main.controller.user_controller import api as user_ns
from .main.controller.auth_controller import api as auth_ns

blueprint = Blueprint('api', __name__)

api = Api(blueprint,
          title='FLASK RESTPLUS API BOILER-PLATE WITH JWT',
          version='1.0',
          description='a boilerplate for flask restplus web service'
          )

api.add_namespace(user_ns, path='/user')
api.add_namespace(auth_ns)

python manage.py run でアプリケーションを実行し、ブラウザで URL http://127.0.0.1:5000 を開きます。

Swagger ドキュメントには、login および logout エンドポイントを備えた新しく作成された auth 名前空間が反映されているはずです。

認証が期待どおりに機能していることを確認するためのテストを作成する前に、登録が成功したらユーザーが自動的にログインするように登録エンドポイントを変更しましょう。

以下のメソッド generate_token を user_service.py に追加します。

def generate_token(user):
    try:
        # generate the auth token
        auth_token = user.encode_auth_token(user.id)
        response_object = {
            'status': 'success',
            'message': 'Successfully registered.',
            'Authorization': auth_token.decode()
        }
        return response_object, 201
    except Exception as e:
        response_object = {
            'status': 'fail',
            'message': 'Some error occurred. Please try again.'
        }
        return response_object, 401

generate_token メソッドは、ユーザー id をエンコードして認証トークンを生成します。このトークン は です。 > がレスポンスとして返されました。

次に、以下の save_new_user メソッドの return ブロックを置き換えます。

response_object = {
    'status': 'success',
    'message': 'Successfully registered.'
}
return response_object, 201

と

return generate_token(new_user)

ここで、login 機能と logout 機能をテストします。次の内容を含む新しいテスト ファイル test_auth.py をテスト パッケージ内に作成します。

import unittest
import json
from app.test.base import BaseTestCase


def register_user(self):
    return self.client.post(
        '/user/',
        data=json.dumps(dict(
            email='example@gmail.com',
            username='username',
            password='123456'
        )),
        content_type='application/json'
    )


def login_user(self):
    return self.client.post(
        '/auth/login',
        data=json.dumps(dict(
            email='example@gmail.com',
            password='123456'
        )),
        content_type='application/json'
    )


class TestAuthBlueprint(BaseTestCase):

    def test_registered_user_login(self):
            """ Test for login of registered-user login """
            with self.client:
                # user registration
                user_response = register_user(self)
                response_data = json.loads(user_response.data.decode())
                self.assertTrue(response_data['Authorization'])
                self.assertEqual(user_response.status_code, 201)

                # registered user login
                login_response = login_user(self)
                data = json.loads(login_response.data.decode())
                self.assertTrue(data['Authorization'])
                self.assertEqual(login_response.status_code, 200)

    def test_valid_logout(self):
        """ Test for logout before token expires """
        with self.client:
            # user registration
            user_response = register_user(self)
            response_data = json.loads(user_response.data.decode())
            self.assertTrue(response_data['Authorization'])
            self.assertEqual(user_response.status_code, 201)

            # registered user login
            login_response = login_user(self)
            data = json.loads(login_response.data.decode())
            self.assertTrue(data['Authorization'])
            self.assertEqual(login_response.status_code, 200)

            # valid token logout
            response = self.client.post(
                '/auth/logout',
                headers=dict(
                    Authorization='Bearer ' + json.loads(
                        login_response.data.decode()
                    )['Authorization']
                )
            )
            data = json.loads(response.data.decode())
            self.assertTrue(data['status'] == 'success')
            self.assertEqual(response.status_code, 200)

if __name__ == '__main__':
    unittest.main()

より包括的なテスト ケースについては、github リポジトリにアクセスしてください。

ルートの保護と認可

これまでのところ、エンドポイントは正常に作成され、ログインおよびログアウト機能が実装されていますが、エンドポイントは保護されていないままです。

どのエンドポイントが開いているか、または認証や管理者権限が必要かを決定するルールを定義する方法が必要です。

これは、エンドポイント用のカスタム デコレータを作成することで実現できます。

エンドポイントを保護または承認する前に、現在ログインしているユーザーを知る必要があります。これを行うには、フラスコ ライブラリ request を使用して、現在のリクエストのヘッダーから Authorization トークン を取得します。次に、Authorization トークンからユーザーの詳細をデコードします。 。

auth_helper.py ファイルの Auth クラスに、次の静的メソッドを追加します。

@staticmethod
def get_logged_in_user(new_request):
        # get the auth token
        auth_token = new_request.headers.get('Authorization')
        if auth_token:
            resp = User.decode_auth_token(auth_token)
            if not isinstance(resp, str):
                user = User.query.filter_by(id=resp).first()
                response_object = {
                    'status': 'success',
                    'data': {
                        'user_id': user.id,
                        'email': user.email,
                        'admin': user.admin,
                        'registered_on': str(user.registered_on)
                    }
                }
                return response_object, 200
            response_object = {
                'status': 'fail',
                'message': resp
            }
            return response_object, 401
        else:
            response_object = {
                'status': 'fail',
                'message': 'Provide a valid auth token.'
            }
            return response_object, 401

リクエストからログインしているユーザーを取得できるようになったので、次に デコレータ を作成しましょう。

util パッケージ内に次の内容のファイル decorator.py を作成します。

from functools import wraps
from flask import request

from app.main.service.auth_helper import Auth


def token_required(f):
    @wraps(f)
    def decorated(*args, **kwargs):

        data, status = Auth.get_logged_in_user(request)
        token = data.get('data')

        if not token:
            return data, status

        return f(*args, **kwargs)

    return decorated


def admin_token_required(f):
    @wraps(f)
    def decorated(*args, **kwargs):

        data, status = Auth.get_logged_in_user(request)
        token = data.get('data')

        if not token:
            return data, status

        admin = token.get('admin')
        if not admin:
            response_object = {
                'status': 'fail',
                'message': 'admin token required'
            }
            return response_object, 401

        return f(*args, **kwargs)

    return decorated

デコレータとその作成方法の詳細については、このリンクをご覧ください。

有効なトークンと管理者トークンに対してそれぞれデコレータ token_required と admin_token_required を作成したので、あとは保護したいエンドポイントにアノテーションを付けるだけです。 freecodecamp の組織に適したデコレータです。

追加のヒント

現在、アプリケーションでいくつかのタスクを実行するには、アプリの起動、テストの実行、依存関係のインストールなどのさまざまなコマンドを実行する必要があります。Makefile. を使用してすべてのコマンドを 1 つのファイルに配置することで、これらのプロセスを自動化できます。

アプリケーションのルート ディレクトリに、ファイル拡張子のない Makefile を作成します。ファイルには次の内容が含まれている必要があります。

.PHONY: clean system-packages python-packages install tests run all

clean:
   find . -type f -name '*.pyc' -delete
   find . -type f -name '*.log' -delete

system-packages:
   sudo apt install python-pip -y

python-packages:
   pip install -r requirements.txt

install: system-packages python-packages

tests:
   python manage.py test

run:
   python manage.py run

all: clean install tests run

メイクファイルのオプションは次のとおりです。

  1. make install : システム パッケージと Python パッケージの両方をインストールします
  2. make clean : アプリをクリーンアップします。
  3. make testing : すべてのテストを実行します。
  4. make run : アプリケーションを開始します。
  5. make all : clean-up、installation を実行し、 tests を実行し、 starts を実行します。アプリ。

アプリの拡張と結論

現在のアプリケーション構造をコピーして拡張し、アプリに機能やエンドポイントを追加するのは非常に簡単です。実装されている以前のルートを表示するだけです。

ご質問、ご意見、ご提案がございましたら、お気軽にコメントを残してください。また、この投稿が役に立った場合は、拍手アイコンをクリックしてください。そうすれば、他の人もここを見て恩恵を受けることができます。

完全なプロジェクトについては、github リポジトリにアクセスしてください。

読んでいただきありがとうございます、頑張ってください!

関連記事