Рубрики
Без рубрики

Как структурировать веб-сервис Flask-Restplus для производства сборки

Автор оригинала: FreeCodeCapm Team.

Грег Обинна

В этом руководстве я покажу вам шаг за шагом подход для структурирования веб-приложения Flask Restplus для тестирования, разработки и производства. Я буду использовать OS на основе Linux (Ubuntu), но большинство этапов могут быть реплицированы на Windows и Mac.

Прежде чем продолжить с этим руководством, у вас должно быть базовое понимание языка программирования Python и Flask Micro Framework. Если вы не знакомы с тем, я рекомендую взглянуть на вступительные статьи – Как использовать Python и Flask, чтобы построить веб-приложение.

Как это руководство структурировано

Это руководство разделено на следующие части:

  • Функции
  • Что такое флаб-рестораны?
  • Настройка и установка
  • Настройка и организация проекта
  • Настройки конфигурации
  • Скрипт колбы
  • Модели базы данных и миграция
  • Тестирование
  • Конфигурация
  • Пользовательские операции
  • Безопасность и аутентификация
  • Защита и авторизация маршрута
  • Дополнительные советы
  • Расширение приложения и заключения

Функции

Мы будем использовать следующие функции и расширения в нашем проекте.

  • Flask-Bcrypt : A Расширение флэки, которое обеспечивает утилиты хеширования BCRYPT для вашего приложения Отказ
  • Флэк-мигрировать : Расширение, которое обрабатывает миграции баз данных SQLALCHEMY для приложений в колбе с помощью алемического. Операции базы данных выполнены через интерфейс командной строки Flask или через расширение скрипта Flask-Script.
  • Flask-Sqlalchemy : Расширение для Колбу Это добавляет поддержку для SQLalchemy к вашей заявке.
  • Pyjwt : Библиотека Python, которая позволяет вам кодировать и декодировать JSON Web Tookens (JWT). JWT является открытым, отраслевым стандартом ( RFC 7519 ) для безопасного представления претензий между двумя сторонами.
  • Флэк-скрипт : Расширение, которое обеспечивает поддержку для записи внешних скриптов в колбе и других задачах командной строки, которые принадлежат за пределами самого веб-приложения.
  • Пространства имен ( Чертежи )
  • Flask-Restplus
  • Модульный тест

Что такое флаб-рестораны?

Flask-Restplus – это расширение для колбы, которая добавляет поддержку для быстрого здания API. Flask-Restplus поощряет лучшие практики с минимальной установкой. Он предоставляет когерентную коллекцию декораторов и инструментов для описания вашего API и правильно разоблачить свою документацию (используя чванство).

Настройка и установка

Проверьте, если у вас установлен PIP, набрав команду PIP --version В терминал, затем нажмите Enter.

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

Настройте виртуальную среду и виртуальную обертку окружающей среды (вам нужно только один из них, в зависимости от версии, установленной выше):

sudo pip install virtualenv

sudo pip3 install virtualenvwrapper

Следуйте за Эта ссылка Для полной настройки оболочки виртуальной среды.

Создайте новую среду и активируйте ее, выполняя следующую команду на терминале:

mkproject name_of_your_project

Настройка и организация проекта

Я буду пользоваться Функциональная структура организовать файлы проекта тем, что они делают. В функциональной структуре шаблоны сгруппированы вместе в одном каталоге, статические файлы в другом и представлениях в третьем.

В каталоге проекта создайте новый пакет под названием приложение Отказ Внутри приложение Создайте два пакета Главная и Тест Отказ Структура вашей каталога должна выглядеть аналогичной ниже.

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

Мы собираемся использовать функциональную структуру для модуляции нашего приложения. Внутри Главная Пакет, создайте еще три пакета, а именно: контроллер , Сервис и Модель Отказ Модель Пакет будет содержать все наши модели базы данных, пока Сервис Пакет будет содержать всю бизнес-логику нашего приложения и, наконец, контроллер Пакет будет содержать все наши конечные точки приложения. Структура дерева теперь должна выглядеть следующим образом:

.
├── 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

Создать или обновить требования .txt Файл, запустив команду:

pip freeze > requirements.txt

Сгенерированный требования .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

Настройки конфигурации

В Главная Пакет Создайте файл под названием 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

Файл конфигурации содержит три класса настройки среды, которые включают Тестирование , Развитие и Производство Отказ

Мы будем использовать Применение заводской шаблон для создания нашего объекта колбы. Этот шаблон наиболее полезен для создания нескольких экземпляров нашего приложения с разными настройками. Это облегчает легкость, в которой мы переходим между нашим тестированием, разработкой и производственной средой, позвонив create_app Функция с обязательным параметром.

В __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 Импортирует модули Migrate и Manager соответственно (мы скоро будем использовать команду Migrate).
  • линия 9 называет create_app Функция Мы создали первоначально для создания экземпляра приложения с необходимым параметром из переменной среды, которая может быть одним из следующих – dev , Продукты , Тест Отказ Если никто не установлен в переменной среды, по умолчанию dev используется.
  • линия 13 и 15 Значения менеджера и миграционных классов, передав приложение экземпляр к их соответствующим конструкторам.
  • В линия 17 мы проходим дБ и Migratecommand экземпляры для add_command Интерфейс Менеджер Чтобы открыть все команды миграции базы данных через скрипт Flask.
  • линия 20 и 25 Отмечает две функции в качестве исполняемых файлов из командной строки.

На данный момент мы можем проверить приложение, запустив команду ниже в корневом каталоге проекта.

python manage.py run

Если все в порядке, вы должны увидеть что-то вроде этого:

Модели базы данных и миграция

Теперь давайте создадим наши модели. Мы будем использовать дБ Экземпляр SQLALCHEMY для создания наших моделей.

дБ Экземпляр содержит все функции и помощники от обоих SQLalchemy и sqlalchemy.orm и Это обеспечивает класс под названием Модель Это декларативная база, которая может быть использована для объявления моделей.

В Модель Пакет, создать файл под названием 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 "".format(self.username)

Приведенный выше код внутри user.py делает следующее:

  • Строка 3: Пользователь Класс наследует от db.model Класс, который объявляет класс как модель для SQLalchemy.
  • линия 7 через 13 Создает необходимые столбцы для пользовательской таблицы.
  • линия 21 это установка для поля Password_hash И это использует Flask-Bcrypt Чтобы генерировать хеш, используя предоставленный пароль.
  • линия 24 Сравнивает данный пароль с уже сохраненным Password_hash Отказ

Теперь, чтобы генерировать таблицу базы данных из Пользователь Модель, которую мы только что создали, мы будем использовать Migratecommand через Менеджер интерфейс. Для Менеджер Чтобы обнаружить наши модели, нам придется импортировать Пользователь Модель, добавив ниже код в Manage.py файл:

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

Теперь мы можем приступить к выполнению Миграция Запустив следующие команды на корневом каталоге проекта:

  1. Инициируйте папку миграции, используя init Команда для алемического для выполнения миграций.
python manage.py db init

2. Создайте сценарий миграции от обнаруженных изменений в модели, используя мигрировать команда. Это еще не влияет на базу данных.

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

3. Примените сценарий миграции в базу данных, используя Обновление команда

python manage.py db upgrade

Если все успешно работает, у вас должна быть новая база данных SQLite колбу котельной Main.db Файл, сгенерированный внутри основного пакета.

Тестирование

Конфигурация

Чтобы быть уверенным, что настройка для нашей конфигурации окружающей среды работает, давайте напишем пару тестов для него.

Создайте файл под названием 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.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 Создает новый пользователь путем первой проверки, если пользователь уже существует; Это возвращает успех Ответ_object Если пользователь не существует еще, он возвращает код ошибки 409 и неудача Ответ_object Отказ
  • линия 33 и 37 Верните список всех зарегистрированных пользователей и объекта пользователя, предоставляя public_id соответственно.
  • линия 40 к 42 совершает изменения в базе данных.

В Главная Пакет, создайте новый пакет под названием Util Отказ Этот пакет будет содержать все необходимые утилиты, которые нам могут понадобиться в нашем приложении.

В Util Пакет, создайте новый файл det.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')
    })

Приведенный выше код внутри det.py делает следующее:

  • Линия 5 Создает новое пространство имен для операций, связанных с пользователем. Flask-RestPlus предоставляет способ использовать практически одинаковый шаблон, что и План Отказ Основная идея состоит в том, чтобы разделить приложение в многоразовые пространства имен. Модуль пространства имен будет содержать моделей и декларацию ресурсов.
  • линия 6 Создает новый пользователь DTO через Модель Интерфейс, предоставленный API пространство имен в Линия 5 Отказ

Контроллер пользователя: Класс контроллера пользователя обрабатывает все входящие HTTP-запросы, относящиеся к пользователю.

Под контроллер Пакет, создайте новый файл под названием 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('/')
@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 и Пользователь Отказ Эти два класса расширяют абстрактный ресурс Flask-Restplus.

API пространство имен в линия 7 Выше представлен контроллер несколькими декораторами, которые включают, но не ограничиваются следующим:

  • API. Маршрут : Декоратор по маршруту ресурсов
  • API. Marshal_with : Декоратор, указывающий поля для использования для сериализации (именно здесь мы используем userdto мы создали ранее)
  • API. marshal_list_with : Ярлык декоратор для Marshal_with выше с as_list.
  • API. Док : Декоратор, чтобы добавить некоторую документацию API в оформленный объект
  • API. Ответ: Декоратор, чтобы указать одну из ожидаемых ответов
  • API. Ожидайте: Декоратор, чтобы указать ожидаемую модель ввода (мы все еще используем userdto для ожидаемого ввода)
  • API. paral: Декоратор, чтобы указать один из ожидаемых параметров

Теперь мы определили наше пространство имен с помощью пользовательского контроллера. Теперь его время, чтобы добавить его в точку записи приложения.

В __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 мы создаем экземпляр плана, проходя Имя и Import_name. API является основной точкой входа для ресурсов приложений и, следовательно, необходимо инициализировать с помощью План в линия 10 Отказ
  • В линия 16 Добавляем пользовательское пространство имен user_ns К списку пространств имен в API пример.

Теперь мы определили наш план. Пришло время зарегистрировать его в нашем приложении Flask. Обновить Manage.py импортируя План и регистрация его с помощью экземпляра приложения Flask.

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.

Давайте проверим Создать новый пользователь Конечная точка с использованием функциональности тестирования SWARGER.

Вы должны получить следующий ответ

Безопасность и аутентификация

Давайте создадим модель BlackListtoken Для хранения черной списки токенов. В модели Пакет, создать 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 '

Давайте не забудем мигрировать изменения в силу в нашей базе данных. Импорт Черный список класс в Manage.py Отказ

from app.main.model import blacklist

Запустите мигрировать и Обновление команды

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

Обновите Пользователь Модель с двумя статическими методами кодирования и декодирования токенов. Добавьте следующие импорт:

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.'

Теперь давайте напишем тест на Пользователь Модель, чтобы убедиться, что наши кодировать и декодировать Функции работают должным образом.

В Тест Пакет, создать 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()

Baseetescase Устанавливает нашу тестовую среду, готовую до и после каждого тестового случая, который расширяет его.

Создать 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 Отказ Все тесты должны пройти.

Давайте создадим Конечные точки аутентификации для Вход и Выход из системы Отказ

  • Сначала нам нужен Dто для логинской полезной нагрузки. Мы будем использовать Auth Do для @expect Аннотация в Вход конечная точка. Добавьте код ниже на do.py.py.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.
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
  • Давайте теперь создаем конечные точки для Вход и Выход из системы Операции. В пакете контроллера создайте 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)
  • На данный момент осталось единственное, что зарегистрировать Auth API пространство имен с приложением Чертеж

Обновить __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 теперь должна отражать вновь созданные 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 Метод генерирует аутентификацию токен Кодируя пользователя я бы. Это токен это возвращено как ответ.

Далее замените Возвращение Блок в Save_new_user метод ниже

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

с участием

return generate_token(new_user)

Теперь пришло время проверить Вход и Выход из системы Функциональные возможности. Создать новый тестовый файл 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 Repo для более исчерпывающих тестовых случаев.

Защита и авторизация маршрута

До сих пор мы успешно создали наши конечные точки, реализованные функциональные возможности входа в систему и выходы, но наши конечные точки остаются незащищенными.

Нам нужен способ определить правила, которые определяют, какая из нашей конечной точки открыта или требует аутентификации или даже привилегий администратора.

Мы можем достичь этого, создавая пользовательские декораторы для наших конечных точек.

Прежде чем мы сможем защитить или разрешать любую из наших конечных точек, нам нужно знать текущий вошедший в систему пользователем. Мы можем сделать это, потянув Токен авторизации От заголовка текущего запроса, используя библиотеку Flask запрос. Затем мы декодируем детали пользователей от Токен авторизации Отказ

В Auth класс auth_helper.py Файл, добавьте следующий статический метод:

@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

Теперь, когда мы можем получить войти в систему пользователя по запросу, давайте пойдем вперед и создать декораторы.

Создать файл декоратор .py В Util Пакет со следующим контентом:

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_requireed Для действительного токена и для токена администратора соответственно все, что осталось, это аннотировать конечные точки, которые мы хотим защитить с оргобзойдерами FreeCodeCamp Декоратор Отказ

Дополнительные советы

В настоящее время для выполнения некоторых задач в нашем приложении мы обязаны для запуска различных команд для запуска приложения, запущенных тестов, устанавливающих зависимости и т. Д. Мы можем автоматизировать эти процессы, устраивая все команды в одном файле, используя Makefile.

В корневом каталоге приложения создайте 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. сделать установку : Установка как системных пакетов и Python-Packages
  2. сделать чистый : очищает приложение
  3. сделать тесты : управляет всеми тестами
  4. сделать бегать : запускает приложение
  5. сделать все : Выполняет Уборка , Установка , бегите Тесты и Начинается приложение.

Расширение приложения и заключения

Довольно легко скопировать текущую структуру приложений и продлить его, чтобы добавить дополнительные функции/конечные точки в приложение. Просто просмотрите любой из предыдущих маршрутов, которые были реализованы.

Не стесняйтесь оставлять комментарий, у вас какие-либо вопросы, наблюдения или рекомендации. Кроме того, если этот пост был полезен для вас, нажмите на значок хлопания, чтобы другие увидят это здесь и пользуются.

Посетите Github Repository для полного проекта.

Спасибо за чтение и удачи!

Читайте ещё по теме: