كيفية هيكلة خدمة ويب Flask-RESTPlus لبناءات الإنتاج

في هذا الدليل ، سأوضح لك نهجًا تدريجيًا لإنشاء تطبيق ويب Flask RESTPlus للاختبار والتطوير وبيئات الإنتاج. سأستخدم نظام تشغيل يعمل بنظام Linux (Ubuntu) ، ولكن يمكن تكرار معظم الخطوات على نظامي التشغيل Windows و Mac.

قبل متابعة هذا الدليل ، يجب أن يكون لديك فهم أساسي للغة برمجة Python وإطار عمل Flask الصغير. إذا لم تكن معتادًا على هؤلاء ، فإنني أوصي بإلقاء نظرة على مقالة تمهيدية - كيفية استخدام Python و Flask لإنشاء تطبيق ويب.

كيف يتم تنظيم هذا الدليل

ينقسم هذا الدليل إلى الأجزاء التالية:

  • المميزات
  • ما هو Flask-RESTPlus؟
  • الإعداد والتثبيت
  • إعداد المشروع وتنظيمه
  • إعدادات التكوين
  • نص قارورة
  • نماذج قواعد البيانات والهجرة
  • اختبارات
  • ترتيب
  • عمليات المستخدم
  • الأمان والمصادقة
  • حماية الطريق والترخيص
  • نصائح اضافية
  • تمديد التطبيق والاستنتاج

المميزات

سنستخدم الميزات والإضافات التالية في مشروعنا.

  • Flask-Bcrypt: ملحق Flask يوفر أدوات مساعدة لتجزئة bcrypt لتطبيقك .
  • Flask-Migrate: ملحق يعالج عمليات ترحيل قاعدة بيانات SQLAlchemy لتطبيقات Flask باستخدام Alembic. يتم توفير عمليات قاعدة البيانات من خلال واجهة سطر أوامر Flask أو من خلال ملحق Flask-Script.
  • Flask-SQLAlchemy: امتداد لـ Flask يضيف دعمًا لـ SQLAlchemy إلى تطبيقك.
  • PyJWT: مكتبة Python تسمح لك بتشفير وفك تشفير JSON Web Tokens (JWT). JWT هو معيار صناعي مفتوح (RFC 7519) لتمثيل المطالبات بشكل آمن بين طرفين.
  • Flask-Script: ملحق يوفر الدعم لكتابة البرامج النصية الخارجية في Flask ومهام سطر الأوامر الأخرى التي تنتمي إلى خارج تطبيق الويب نفسه.
  • مساحات الأسماء (مخططات)
  • ريست بلس القارورة
  • وحدة الاختبار

ما هو Flask-RESTPlus؟

Flask-RESTPlus هو امتداد لـ Flask يضيف دعمًا لبناء واجهات برمجة تطبيقات REST بسرعة. يشجع Flask-RESTPlus أفضل الممارسات مع الحد الأدنى من الإعداد. يوفر مجموعة متماسكة من أدوات الديكور والأدوات لوصف API الخاص بك وفضح وثائقها بشكل صحيح (باستخدام Swagger).

الإعداد والتثبيت

تحقق مما إذا كان لديك نقطة مثبتة ، عن طريق كتابة الأمر pip --versionفي Terminal ، ثم اضغط على Enter.

pip --version

إذا استجابت المحطة الطرفية برقم الإصدار ، فهذا يعني أن النقطة مثبتة ، لذا انتقل إلى الخطوة التالية ، أو قم بتثبيت pip أو باستخدام مدير حزمة Linux ، وقم بتشغيل الأمر أدناه على الجهاز واضغط على إدخال. اختر إصدار 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. في الداخل app، قم بإنشاء حزمتين main و test. يجب أن تبدو بنية الدليل الخاص بك مماثلة لتلك الموجودة أدناه.

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

سنستخدم بنية وظيفية لنمذجة تطبيقنا.

داخل mainالحزمة ، قم بإنشاء ثلاث حزم أخرى وهي: controller، serviceو model. و 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.

سنستخدم نموذج مصنع التطبيق لإنشاء كائن Flask الخاص بنا. هذا النمط هو الأكثر فائدة لإنشاء مثيلات متعددة لتطبيقنا بإعدادات مختلفة. هذا يسهل السهولة التي ننتقل بها بين بيئة الاختبار والتطوير والإنتاج لدينا من خلال استدعاء create_appالوظيفة بالمعامل المطلوب.

في __init__.pyالملف داخل mainالحزمة ، أدخل سطور التعليمات البرمجية التالية:

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يقوم بما يلي:

  • line 4و 5اردات وحدات ترحيل ومدير التوالي (سنقوم باستخدام الأمر ترحيل قريبا).
  • line 9يستدعي create_appالوظيفة التي أنشأناها في البداية لإنشاء مثيل التطبيق بالمعامل المطلوب من متغير البيئة والذي يمكن أن يكون أيًا مما يلي - dev، prod، test. إذا لم يتم تعيين أي شيء في متغير البيئة ، فسيتم devاستخدام الافتراضي .
  • line 13ويقوم 15بإنشاء مثيل للمدير وترحيل الفئات عن طريق تمرير appالمثيل إلى المُنشئين المعنيين.
  • في line 17، نقوم بتمرير المثيلات dbو MigrateCommandإلى add_commandواجهة the managerلفضح جميع أوامر ترحيل قاعدة البيانات من خلال Flask-Script.
  • line 20و 25علامات الوظيفتين أنه قابل للتنفيذ من سطر الأوامر.
يعرض Flask-Migrate فئتين ، Migrateو MigrateCommand. و Migrateالطبقة تحتوي على كافة وظائف للتمديد. و MigrateCommandيستخدم الدرجة فقط عندما يتم المطلوب لفضح الأوامر الهجرة قاعدة البيانات من خلال تمديد قارورة سكريبت.

في هذه المرحلة ، يمكننا اختبار التطبيق عن طريق تشغيل الأمر أدناه في الدليل الجذر للمشروع.

python manage.py run

إذا كان كل شيء على ما يرام ، يجب أن ترى شيئًا مثل هذا:

نماذج قواعد البيانات والهجرة

لنقم الآن بإنشاء نماذجنا. سنكون باستخدام dbمثيل sqlalchemy لخلق نماذج لدينا.

على 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 "".format(self.username)

الكود أعلاه user.pyيقوم بما يلي:

  • line 3:و userيرث فئة من db.Modelالطبقة التي تعلن الطبقة كنموذج لsqlalchemy.
  • line 7من خلال 13إنشاء الأعمدة المطلوبة لجدول المستخدم.
  • line 21هو أداة ضبط للحقل password_hashويستخدم flask-bcryptلإنشاء تجزئة باستخدام كلمة المرور المقدمة.
  • line 24يقارن كلمة مرور معينة مع كلمة مرور محفوظة بالفعل password_hash.

الآن لإنشاء جدول قاعدة البيانات من userالنموذج الذي أنشأناه للتو ، سنستخدمه من migrateCommandخلال managerالواجهة. ل managerللكشف عن النماذج لدينا، وسوف تضطر إلى استيراد userالنموذج بإضافة رمز أدناه إلى manage.pyملف:

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

يمكننا الآن المتابعة لإجراء الترحيل عن طريق تشغيل الأوامر التالية على الدليل الجذر للمشروع:

  1. ابدأ مجلد ترحيل باستخدام initالأمر لـ alembic لإجراء عمليات الترحيل.
python manage.py db init

2. قم بإنشاء برنامج نصي للترحيل من التغييرات المكتشفة في النموذج باستخدام migrateالأمر. هذا لا يؤثر على قاعدة البيانات حتى الآن.

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

3. تطبيق البرنامج النصي الترحيل على قاعدة البيانات باستخدام upgradeالأمر

python manage.py db upgrade

إذا كان كل شيء يعمل بنجاح ، يجب أن يكون لديك قاعدة بيانات sqlLite جديدة

flask_boilerplate_main.db ملف تم إنشاؤه داخل الحزمة الرئيسية.

في كل مرة يتغير فيها نموذج قاعدة البيانات ، كرر الأوامر migrateو upgrade

اختبارات

ترتيب

للتأكد من عمل الإعداد الخاص بتهيئة بيئتنا ، دعنا نكتب اختبارين له.

Create a file called test_config.py in the test package with the content below:

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()

Run the test using the command below:

python manage.py test

You should get the following output:

User Operations

Now let’s work on the following user related operations:

  • creating a new user
  • getting a registered user with his public_id
  • getting all registered users.

User Service class: This class handles all the logic relating to the user model.

In the service package, create a new file user_service.py with the following content:

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()

The above code within user_service.py does the following:

  • line 8 through 29 creates a new user by first checking if the user already exists; it returns a success response_object if the user doesn’t exist else it returns an error code 409 and a failure response_object.
  • line 33و 37إرجاع قائمة جميع المستخدمين المسجلين وكائن المستخدم من خلال توفير public_idالتوالي.
  • line 40إلى 42تلزم تغييرات في قاعدة البيانات.
لا حاجة لاستخدام jsonify لتنسيق كائن إلى JSON ، يقوم 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يقوم بما يلي:

  • line 5 creates a new namespace for user related operations. Flask-RESTPlus provides a way to use almost the same pattern as Blueprint. The main idea is to split your app into reusable namespaces. A namespace module will contain models and resources declaration.
  • line 6 creates a new user dto through the model interface provided by the api namespace in line 5.

User Controller: The user controller class handles all the incoming HTTP requests relating to the user .

Under the controller package, create a new file called user_controller.py with the following content:

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

line 1 through 8 imports all the required resources for the user controller.

We defined two concrete classes in our user controller which are

userList and user. These two classes extends the abstract flask-restplus resource.

Concrete resources should extend from this class and expose methods for each supported HTTP method. If a resource is invoked with an unsupported HTTP method, the API will return a response with status 405 Method Not Allowed. Otherwise the appropriate method is called and passed all arguments from the URL rule used when adding the resource to an API instance.

The api namespace in line 7 above provides the controller with several decorators which includes but is not limited to the following:

  • api.route: A decorator to route resources
  • api.marshal_with: A decorator specifying the fields to use for serialization (This is where we use the userDto we created earlier)
  • api.marshal_list_with: A shortcut decorator for marshal_with above withas_list = True
  • api.doc: A decorator to add some api documentation to the decorated object
  • api.response: A decorator to specify one of the expected responses
  • api.expect: A decorator to Specify the expected input model ( we still use the userDto for the expected input)
  • api.param: A decorator to specify one of the expected parameters

We have now defined our namespace with the user controller. Now its time to add it to the application entry point.

In the __init__.py file of app package, enter the following:

# 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,, version="1.0", description="a boilerplate for flask restplus web service" ) api.add_namespace(user_ns, path="/user")

The above code within blueprint.py does the following:

  • In line 8, we create a blueprint instance by passing name and import_name.API is the main entry point for the application resources and hence needs to be initialized with the blueprint in line 10.
  • In line 16 , we add the user namespace user_ns to the list of namespaces in the API instance.

We have now defined our blueprint. It’s time to register it on our Flask app.

Update manage.py by importing blueprint and registering it with the Flask application instance.

from app import blueprint ... app = create_app(os.getenv('BOILERPLATE_ENV') or 'dev') app.register_blueprint(blueprint) app.app_context().push() ...

We can now test our application to see that everything is working fine.

python manage.py run

Now open the URL //127.0.0.1:5000 in your browser. You should see the swagger documentation.

Let’s test the create new user endpoint using the swagger testing functionality.

You should get the following response

Security and Authentication

Let’s create a model blacklistToken for storing blacklisted tokens. In the models package, create a blacklist.py file with the following content:

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 '

Lets not forget to migrate the changes to take effect on our database.

Import the blacklist class in manage.py.

from app.main.model import blacklist

Run the migrate and upgrade commands

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

Next create blacklist_service.py in the service package with the following content for blacklisting a token:

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

Update the user model with two static methods for encoding and decoding tokens. Add the following imports:

import datetime import jwt from app.main.model.blacklist import BlacklistToken from ..config import key
  • Encoding
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
  • Decoding: Blacklisted token, expired token and invalid token are taken into consideration while decoding the authentication token.
 @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.'

Now let’s write a test for the user model to ensure that our encode and decode functions are working properly.

In the test package, create base.py file with the following content:

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()

The BaseTestCase sets up our test environment ready before and after every test case that extends it.

Create test_user_medol.py with the following test cases:

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="[email protected]", 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="[email protected]", 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()

Run the test with python manage.py test. All the tests should pass.

Let’s create the authentication endpoints for login and logout.

  • First we need a dto for the login payload. We will use the auth dto for the @expect annotation in login endpoint. Add the code below to the 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"), })
  • Next, we create an authentication helper class for handling all authentication related operations. This auth_helper.py will be in the service package and will contain two static methods which are login_user and logout_user
When a user is logged out, the user’s token is blacklisted ie the user can’t log in again with that same token.
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
  • Let us now create endpoints for login and logout operations.

    In the controller package, create

    auth_controller.py with the following contents:

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)
  • At this point the only thing left is to register the auth api namespace with the application Blueprint

Update __init__.py file of app package with the following

# 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,, version="1.0", description="a boilerplate for flask restplus web service" ) api.add_namespace(user_ns, path="/user") api.add_namespace(auth_ns)

Run the application with python manage.py run and open the url //127.0.0.1:5000 in your browser.

The swagger documentation should now reflect the newly created auth namespace with the login and logout endpoints.

Before we write some tests to ensure our authentication is working as expected, let’s modify our registration endpoint to automatically login a user once the registration is successful.

Add the method generate_token below to 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

The generate_token method generates an authentication token by encoding the user id. This token isthe returned as a response.

Next, replace the return block in save_new_user method below

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

with

return generate_token(new_user)

Now its time to test the login and logout functionalities. Create a new test file test_auth.py in the test package with the following content:

import unittest import json from app.test.base import BaseTestCase def register_user(self): return self.client.post( '/user/', data=json.dumps(dict( email="[email protected]", username="username", password="123456" )), content_type="application/json" ) def login_user(self): return self.client.post( '/auth/login', data=json.dumps(dict( email="[email protected]", 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()

Visit the github repo for a more exhaustive test cases.

Route protection and Authorization

So far, we have successfully created our endpoints, implemented login and logout functionalities but our endpoints remains unprotected.

We need a way to define rules that determines which of our endpoint is open or requires authentication or even an admin privilege.

We can achieve this by creating custom decorators for our endpoints.

Before we can protect or authorize any of our endpoints, we need to know the currently logged in user. We can do this by pulling the Authorization token from the header of the current request by using the flask library request.We then decode the user details from the Authorization token.

In the Auth class of auth_helper.py file, add the following static method:

@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

Now that we can retrieve the logged in user from the request, let’s go ahead and create the decorators.

Create a file decorator.py in the util package with the following content:

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

For more information about decorators and how to create them, take a look at this link.

Now that we have created the decorators token_required and admin_token_required for valid token and for an admin token respectively, all that is left is to annotate the endpoints which we wish to protect with the freecodecamp orgappropriate decorator.

Extra tips

Currently to perform some tasks in our application, we are required to run different commands for starting the app, running tests, installing dependencies etc. We can automate those processes by arranging all the commands in one file using Makefile.

On the root directory of the application, create a Makefile with no file extension. The file should contain the following:

.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

Here are the options of the make file.

  1. make install : installs both system-packages and python-packages
  2. make clean : cleans up the app
  3. make tests : runs the all the tests
  4. make run : starts the application
  5. make all : performs clean-up,installation , run tests , and starts the app.

Extending the App & Conclusion

It’s pretty easy to copy the current application structure and extend it to add more functionalities/endpoints to the App. Just view any of the previous routes that have been implemented.

Feel free to leave a comment have you any question, observations or recommendations. Also, if this post was helpful to you, click on the clap icon so others will see this here and benefit as well.

Visit the github repository for the complete project.

Thanks for reading and good luck!