One place for hosting & domains

      Flask

      Build a RESTful API with Flask – The TDD Way: Part 2

      Introduction

      In Part 1 of this series, we learned how to create a RESTful API the TDD way. We covered writing tests and learned a lot about Flask. If you haven’t read Part 1, please do because this tutorial will build upon it.

      In this part of the series, we’ll learn how to authenticate and authorize users in our API.

      In this tutorial, we’ll talk about securing our API with token-based authentication and user authorization. We will integrate users into the API we built in Part 1.

      In order to get started, ensure your virtual environment is activated.

      We intend to allow bucketlists to be owned by users. For now, anyone can manipulate a bucketlist even if they did not create it. We’ve got to fix this security hole.

      How do we keep track of users, you ask? We define a model.

      
      
      from app import db
      from flask_bcrypt import Bcrypt
      
      class User(db.Model):
          """This class defines the users table """
      
          __tablename__ = 'users'
      
          
          id = db.Column(db.Integer, primary_key=True)
          email = db.Column(db.String(256), nullable=False, unique=True)
          password = db.Column(db.String(256), nullable=False)
          bucketlists = db.relationship(
              'Bucketlist', order_by='Bucketlist.id', cascade="all, delete-orphan")
      
          def __init__(self, email, password):
              """Initialize the user with an email and a password."""
              self.email = email
              self.password = Bcrypt().generate_password_hash(password).decode()
      
          def password_is_valid(self, password):
              """
              Checks the password against it's hash to validates the user's password
              """
              return Bcrypt().check_password_hash(self.password, password)
      
          def save(self):
              """Save a user to the database.
              This includes creating a new user and editing one.
              """
              db.session.add(self)
              db.session.commit()
      
      class Bucketlist(db.Model):
          """This class defines the bucketlist table."""
      
          __tablename__ = 'bucketlists'
      
          
          id = db.Column(db.Integer, primary_key=True)
          name = db.Column(db.String(255))
          date_created = db.Column(db.DateTime, default=db.func.current_timestamp())
          date_modified = db.Column(
              db.DateTime, default=db.func.current_timestamp(),
              onupdate=db.func.current_timestamp())
          created_by = db.Column(db.Integer, db.ForeignKey(User.id))
      
          def __init__(self, name, created_by):
              """Initialize the bucketlist with a name and its creator."""
              self.name = name
              self.created_by = created_by
      
          def save(self):
              """Save a bucketlist.
              This applies for both creating a new bucketlist
              and updating an existing onupdate
              """
              db.session.add(self)
              db.session.commit()
      
          @staticmethod
          def get_all(user_id):
              """This method gets all the bucketlists for a given user."""
              return Bucketlist.query.filter_by(created_by=user_id)
      
          def delete(self):
              """Deletes a given bucketlist."""
              db.session.delete(self)
              db.session.commit()
      
          def __repr__(self):
              """Return a representation of a bucketlist instance."""
              return "<Bucketlist: {}>".format(self.name)
      

      Here’s what we’ve done:

      • We imported Flask-Bcrypt extension to help us in hashing our passwords. You should never store passwords in plaintext.
      • We created a User model that represents the users table. It contains the email and password fields to capture the user’s credentials.
      • Since a user can own many bucketlists, we defined a One-to-Many relationship between the two tables. We defined this relationship by adding the db.relationship() function on the User table (parent table)
      • We added a foreign key on the child table (Bucketlist) referencing the User table. The foreign key has some arguments. The cascade="all, delete-orphan" will delete all bucketlists when a referenced user is deleted.
      • We hash the password by using generate_password_hash(password). This will make our users’ passwords be secure from dictionary and brute force attacks.
      • We refactored the get_all() method to get all the bucketlists for a given user.

      Don’t forget to install Flask-Bcrypt

      1. pip install flask-bcrypt

      Migrate them

      Migrate the changes we’ve just made to the database we initially created in Part 1 of the series.

      1. python manage.py db migrate
      2. python manage.py db upgrade

      Now we have a user table to keep track of registered users.

      Our app will have many tests from now on. It’s best practice to have a test folder that will house all our tests. We’ll create a folder called tests. Inside this folder, we’ll move our test_bucketlists.py file into it.

      Our directory structure should now look like this:

      1. ├── bucketlist
      2. ├── app
      3. │ ├── __init__.py
      4. │ └── models.py
      5. ├── instance
      6. │ ├── __init__.py
      7. │ └── config.py
      8. ├── manage.py
      9. ├── requirements.txt
      10. ├── run.py
      11. ├── tests
      12. │ └── test_bucketlist.py

      Also, we’ll edit the manage.py as follows:

      import os
      import unittest
      
      from flask_script import Manager
      from flask_migrate import Migrate, MigrateCommand
      from app import db, create_app
      
      
      app = create_app(config_name=os.getenv('APP_SETTINGS'))
      migrate = Migrate(app, db)
      
      manager = Manager(app)
      
      
      
      manager.add_command('db', MigrateCommand)
      
      
      
      @manager.command
      def test():
          """Runs the unit tests without test coverage."""
          tests = unittest.TestLoader().discover('./tests', pattern='test*.py')
          result = unittest.TextTestRunner(verbosity=2).run(tests)
          if result.wasSuccessful():
              return 0
          return 1
      
      
      if __name__ == '__main__':
          manager.run()
      

      The decorator on top of test() allows us to define a command called test. Inside the function, we load the tests from the tests folder using the TestLoader() class and then run them with TextTestRunner.run(). If it’s successful, we exit gracefully with a return 0.

      Let’s test it out on our terminal.

      1. python manage.py test

      The tests should fail. This is because we’ve not modified our code to work with the new changes in the model.

      From now on, we’ll use this command to run our tests.

      Token-based authentication is a security technique that authenticates users who attempt to login to a server using a security token provided by the server. Without the token, a user won’t be granted access to restricted resources. You can find more intricate details about token-based authentication here

      For us to implement this authentication, we’ll use a Python package called PyJWT. PyJWT allows us to encode and decode JSON Web Tokens (JWT).

      That being said, let’s install it:

      1. pip install PyJWT

      For our users to authenticate, the access token is going to be placed in the Authorization HTTP header in all our bucketlist requests.

      Here’s how the header looks like:

      Authorization:  "Bearer <The-access-token-is-here>"
      

      We’ll put the word Bearer before the token and separate them with a space character.

      Don’t forget the space in between the Bearer and the token.

      Encode and Decode the Token

      We need to create a way to encode the token before it’s sent to the user. We also need to have a way to decode the token when the user sends it via the Authorization header.

      In our model.py we’ll create a function inside our User model to generate the token and another one to decode it. Let’s add the following code:

      
      
      
      import jwt
      from datetime import datetime, timedelta
      
      class User(db.Model):
          """Maps to users table """
      
          __tablename__ = 'users'
      
          
          
          
      
          def __init__(self, email, password):
              
              
      
          def password_is_valid(self, password):
              
              
      
          def save(self):
              
              
      
          def generate_token(self, user_id):
              """ Generates the access token"""
      
              try:
                  
                  payload = {
                      'exp': datetime.utcnow() + timedelta(minutes=5),
                      'iat': datetime.utcnow(),
                      'sub': user_id
                  }
                  
                  jwt_string = jwt.encode(
                      payload,
                      current_app.config.get('SECRET'),
                      algorithm='HS256'
                  )
                  return jwt_string
      
              except Exception as e:
                  
                  return str(e)
      
          @staticmethod
          def decode_token(token):
              """Decodes the access token from the Authorization header."""
              try:
                  
                  payload = jwt.decode(token, current_app.config.get('SECRET'))
                  return payload['sub']
              except jwt.ExpiredSignatureError:
                  
                  return "Expired token. Please login to get a new token"
              except jwt.InvalidTokenError:
                  
                  return "Invalid token. Please register or login"
      

      The generate_token() takes in a user ID as an argument, uses jwt to create a token using the secret key, and makes it time-based by defining its expiration time. The token is valid for 5 minutes as specified in the timedelta. You can set it to your liking.

      The decode_token() takes in a token as an argument and checks whether the token is valid. If it is, it returns the user ID as the payload. It returns an error message if the token is expired or invalid.

      Don’t forget to import jwt and the datetime above.

      Our app is growing bigger. We’ll have to organize it into components. Flask uses the concept of Blueprints to make application components.

      Blueprints are simply a set of operations that can be registered on a given app. Think of it as an extension of the app that can address a specific functionality.

      We’ll create an authentication blueprint.
      This blueprint will focus on handling user registration and logins.

      Inside our /app directory create a folder and call it auth.

      Our auth folder should contain:

      • __init__.py file
      • views.py file

      In our auth/__init__.py file, initialize a blueprint.

      
      
      from flask import Blueprint
      
      
      auth_blueprint = Blueprint('auth', __name__)
      
      from . import views
      

      Then import the blueprint and register it at the bottom of the app/__init__.py, just before the return app line.

      
      
      
      
      def create_app(config_name):
        
        
        
      
        @app.route('/bucketlists/<int:id>', methods=['GET', 'PUT', 'DELETE'])
        def bucketlist_manipulation(id, **kwargs):
               
          
          
          ...
      
        
        from .auth import auth_blueprint
        app.register_blueprint(auth_blueprint)
      
        return app
      

      Testing should never be an afterthought. It should always come first.

      We’re going to add a new test file that will house all our tests for the authentication blueprint.
      It’ll test whether our API can handle user registration, user log in, and access-token generation.

      In our tests directory, create a file naming it test_auth.py. Write the following code in it:

      
      
      import unittest
      import json
      from app import create_app, db
      
      class AuthTestCase(unittest.TestCase):
          """Test case for the authentication blueprint."""
      
          def setUp(self):
              """Set up test variables."""
              self.app = create_app(config_name="testing")
              
              self.client = self.app.test_client
              
              self.user_data = {
                  'email': '[email protected]',
                  'password': 'test_password'
              }
      
              with self.app.app_context():
                  
                  db.session.close()
                  db.drop_all()
                  db.create_all()
      
          def test_registration(self):
              """Test user registration works correcty."""
              res = self.client().post('/auth/register', data=self.user_data)
              
              result = json.loads(res.data.decode())
              
              self.assertEqual(result['message'], "You registered successfully.")
              self.assertEqual(res.status_code, 201)
      
          def test_already_registered_user(self):
              """Test that a user cannot be registered twice."""
              res = self.client().post('/auth/register', data=self.user_data)
              self.assertEqual(res.status_code, 201)
              second_res = self.client().post('/auth/register', data=self.user_data)
              self.assertEqual(second_res.status_code, 202)
              
              result = json.loads(second_res.data.decode())
              self.assertEqual(
                  result['message'], "User already exists. Please login.")
      

      We’ve initialized our test with a test client for making requests to our API and some test data.
      The first test function test_registration() sends a post request to /auth/register and tests the response it gets. It ensures that the status code is 201, meaning we’ve successfully created a user.
      The second test function tests whether the API can only register a user once. Having duplicates in the database is bad for business.

      Now let’s run the tests using python manage.py test. The tests should fail.

      1. ----------------------------------------------------------------------
      2. raise JSONDecodeError("Expecting value", s, err.value) from None
      3. json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0)

      The reason our tests fail is simply because we lack the functionality they need to test. Let’s implement something that’ll make these two tests pass.

      Open up the views.py file and add the following code:

      
      
      from . import auth_blueprint
      
      from flask.views import MethodView
      from flask import make_response, request, jsonify
      from app.models import User
      
      class RegistrationView(MethodView):
          """This class registers a new user."""
      
          def post(self):
              """Handle POST request for this view. Url ---> /auth/register"""
      
              
              user = User.query.filter_by(email=request.data['email']).first()
      
              if not user:
                  
                  try:
                      post_data = request.data
                      
                      email = post_data['email']
                      password = post_data['password']
                      user = User(email=email, password=password)
                      user.save()
      
                      response = {
                          'message': 'You registered successfully. Please log in.'
                      }
                      
                      return make_response(jsonify(response)), 201
                  except Exception as e:
                      
                      response = {
                          'message': str(e)
                      }
                      return make_response(jsonify(response)), 401
              else:
                  
                  
                  response = {
                      'message': 'User already exists. Please login.'
                  }
      
                  return make_response(jsonify(response)), 202
      
      registration_view = RegistrationView.as_view('register_view')
      
      
      auth_blueprint.add_url_rule(
          '/auth/register',
          view_func=registration_view,
          methods=['POST'])
      

      Here’s what we have added:

      • We imported our blueprint together with Flask’s make_response (for returning our response) and jsonify (for encoding our data in JSON and adding an application/json header to the response)
      • We’ve defined a class-based view to handle registration by dispatching a POST request to our post() method.
      • Inside our post() method, we check if the user exists in our database. If they don’t, we create a new user and return a message to them notifying their successful registration.
        If the user exists they are reminded to log in.
      • Lastly, we used as_view() method to make our class-based view callable so that it can take a request and return a response. We then defined the URL for registering a user as /auth/register.

      Let’s run our tests once more. Only the AuthTestCase tests should pass. The bucketlist tests still fail because we haven’t modified the __init__.py code.

      1. test_already_registered_user (test_auth.AuthTestCase)
      2. Test that a user cannot be registered twice. ... ok
      3. test_registration (test_auth.AuthTestCase)
      4. Test user registration works correcty. ... ok
      5. Bucketlist failed tests fall here
      6. ----------------------------------------------------------------------

      Using Postman for auth/register

      We’ll test our registration functionality by making a request using Postman.

      But before we make the requests, ensure the API is up and running.

      1. python run.py development
       * Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)
       * Restarting with stat
       * Debugger is active!
       * Debugger PIN: 225-021-817
      

      Now you can make a POST request to localhost:5000/auth/register. Specify an email and a password of your choice to represent the user we are registering. Click send.

      A user will have to log in to gain access to our API. Currently, we are lacking this login functionality. Let’s start with some tests. We’ll add two more tests at the bottom of our test_auth.py as follows:

      
      class AuthTestCase(unittest.TestCase):
          """Test case for the authentication blueprint."""
      
          def setUp(self):
              
      
          def test_registration(self):
              
      
          def test_already_registered_user(self):
              
      
          def test_user_login(self):
              """Test registered user can login."""
              res = self.client().post('/auth/register', data=self.user_data)
              self.assertEqual(res.status_code, 201)
              login_res = self.client().post('/auth/login', data=self.user_data)
      
              
              result = json.loads(login_res.data.decode())
              
              self.assertEqual(result['message'], "You logged in successfully.")
              
              self.assertEqual(login_res.status_code, 200)
              self.assertTrue(result['access_token'])
      
          def test_non_registered_user_login(self):
              """Test non registered users cannot login."""
              
              not_a_user = {
                  'email': '[email protected]',
                  'password': 'nope'
              }
              
              res = self.client().post('/auth/login', data=not_a_user)
              
              result = json.loads(res.data.decode())
      
              
              
              self.assertEqual(res.status_code, 401)
              self.assertEqual(
                  result['message'], "Invalid email or password, Please try again")
      

      The test_user_login() function tests whether our API can successfully log in as a registered user. It also tests for the access token.

      The other test function test_non_registered_user_login() tests whether our API can restrict signing in to only registered users.

      Again, we’ll make the tests pass by implementing its functionality. Let’s create the login view.

      from . import auth_blueprint
      
      from flask.views import MethodView
      from flask import Blueprint, make_response, request, jsonify
      from app.models import User
      
      class RegistrationView(MethodView):
          """This class-based view registers a new user."""
          
          
      
      class LoginView(MethodView):
          """This class-based view handles user login and access token generation."""
      
          def post(self):
              """Handle POST request for this view. Url ---> /auth/login"""
              try:
                  
                  user = User.query.filter_by(email=request.data['email']).first()
      
                  
                  if user and user.password_is_valid(request.data['password']):
                      
                      access_token = user.generate_token(user.id)
                      if access_token:
                          response = {
                              'message': 'You logged in successfully.',
                              'access_token': access_token.decode()
                          }
                          return make_response(jsonify(response)), 200
                  else:
                      
                      response = {
                          'message': 'Invalid email or password, Please try again'
                      }
                      return make_response(jsonify(response)), 401
      
              except Exception as e:
                  
                  response = {
                      'message': str(e)
                  }
                  
                  return make_response(jsonify(response)), 500
      
      
      registration_view = RegistrationView.as_view('registration_view')
      login_view = LoginView.as_view('login_view')
      
      
      
      auth_blueprint.add_url_rule(
          '/auth/register',
          view_func=registration_view,
          methods=['POST'])
      
      
      
      auth_blueprint.add_url_rule(
          '/auth/login',
          view_func=login_view,
          methods=['POST']
      )
      

      Here, we’ve defined a class-based view just like we did in the registration section.

      It dispatches the POST request to the post() method as well. This is to capture the user credentials (email, password) when they log in. It checks whether the password given is valid, generates an access token for the user, and returns a response containing the token.

      We’ve also handled exceptions gracefully so that if one occurs, our API will continue running and won’t crush.

      Finally, we defined a URL for the login route.

      Logging in on Postman with auth/login

      Make a POST request. Input the email and password we specified for the user during registration. Click send. You should get an access token in the JSON response.

      Running the tests

      If you run the tests, you will notice that the login tests pass, but the bucketlist one still fails. It’s time to refactor these tests.

      First, we’ll create two helper functions for registering and signing in to our test user.

      
      class BucketlistTestCase(unittest.TestCase):
          """This class represents the bucketlist test case"""
      
          def setUp(self):
              """Set up test variables."""
         
         
      
          def register_user(self, email="[email protected]", password="test1234"):
              """This helper method helps register a test user."""
              user_data = {
                  'email': email,
                  'password': password
              }
              return self.client().post('/auth/register', data=user_data)
      
          def login_user(self, email="[email protected]", password="test1234"):
              """This helper method helps log in a test user."""
              user_data = {
                  'email': email,
                  'password': password
              }
              return self.client().post('/auth/login', data=user_data)
      
          
          
      
      
      if __name__ == "__main__":
          unittest.main()
      

      We do this so that when we want to register or log in as a test user (which is in all the tests), we don’t have to repeat ourselves. We’ll simply call the function and we are set.

      Next, we’ll define a way to get the access token and add it to the Authorization header in all our client requests. Here’s a code snippet of how we’re going to do it.

          def test_bucketlist_creation(self):
              """Test the API can create a bucketlist (POST request)"""
              
              self.register_user():
              result = self.login_user()
              
              access_token = json.loads(result.data.decode())['access_token']
      
              
              res = self.client().post(
                  '/bucketlists/',
                  headers=dict(Authorization="Bearer " + access_token),
                  data=self.bucketlist)
      

      We can now go ahead and refactor the whole test_bucketlist.py file. After refactoring all our requests, we should have something like this:

      import unittest
      import os
      import json
      from app import create_app, db
      
      class BucketlistTestCase(unittest.TestCase):
          """This class represents the bucketlist test case"""
      
          def setUp(self):
              """Define test variables and initialize app."""
              self.app = create_app(config_name="testing")
              self.client = self.app.test_client
              self.bucketlist = {'name': 'Go to Borabora for vacay'}
      
              
              with self.app.app_context():
                  
                  db.session.close()
                  db.drop_all()
                  db.create_all()
      
          def register_user(self, email="[email protected]", password="test1234"):
              user_data = {
                  'email': email,
                  'password': password
              }
              return self.client().post('/auth/register', data=user_data)
      
          def login_user(self, email="[email protected]", password="test1234"):
              user_data = {
                  'email': email,
                  'password': password
              }
              return self.client().post('/auth/login', data=user_data)
      
          def test_bucketlist_creation(self):
              """Test API can create a bucketlist (POST request)"""
              self.register_user()
              result = self.login_user()
              access_token = json.loads(result.data.decode())['access_token']
      
              
              res = self.client().post(
                  '/bucketlists/',
                  headers=dict(Authorization="Bearer " + access_token),
                  data=self.bucketlist)
              self.assertEqual(res.status_code, 201)
              self.assertIn('Go to Borabora', str(res.data))
      
          def test_api_can_get_all_bucketlists(self):
              """Test API can get a bucketlist (GET request)."""
              self.register_user()
              result = self.login_user()
              access_token = json.loads(result.data.decode())['access_token']
      
              
              res = self.client().post(
                  '/bucketlists/',
                  headers=dict(Authorization="Bearer " + access_token),
                  data=self.bucketlist)
              self.assertEqual(res.status_code, 201)
      
              
              res = self.client().get(
                  '/bucketlists/',
                  headers=dict(Authorization="Bearer " + access_token),
              )
              self.assertEqual(res.status_code, 200)
              self.assertIn('Go to Borabora', str(res.data))
      
          def test_api_can_get_bucketlist_by_id(self):
              """Test API can get a single bucketlist by using it's id."""
              self.register_user()
              result = self.login_user()
              access_token = json.loads(result.data.decode())['access_token']
      
              rv = self.client().post(
                  '/bucketlists/',
                  headers=dict(Authorization="Bearer " + access_token),
                  data=self.bucketlist)
      
              
              self.assertEqual(rv.status_code, 201)
              
              results = json.loads(rv.data.decode())
      
              result = self.client().get(
                  '/bucketlists/{}'.format(results['id']),
                  headers=dict(Authorization="Bearer " + access_token))
              
              self.assertEqual(result.status_code, 200)
              self.assertIn('Go to Borabora', str(result.data))
      
          def test_bucketlist_can_be_edited(self):
              """Test API can edit an existing bucketlist. (PUT request)"""
              self.register_user()
              result = self.login_user()
              access_token = json.loads(result.data.decode())['access_token']
      
              
              rv = self.client().post(
                  '/bucketlists/',
                  headers=dict(Authorization="Bearer " + access_token),
                  data={'name': 'Eat, pray and love'})
              self.assertEqual(rv.status_code, 201)
              
              results = json.loads(rv.data.decode())
      
              
              rv = self.client().put(
                  '/bucketlists/{}'.format(results['id']),
                  headers=dict(Authorization="Bearer " + access_token),
                  data={
                      "name": "Dont just eat, but also pray and love :-)"
                  })
              self.assertEqual(rv.status_code, 200)
      
              
              results = self.client().get(
                  '/bucketlists/{}'.format(results['id']),
                  headers=dict(Authorization="Bearer " + access_token))
              self.assertIn('Dont just eat', str(results.data))
      
          def test_bucketlist_deletion(self):
              """Test API can delete an existing bucketlist. (DELETE request)."""
              self.register_user()
              result = self.login_user()
              access_token = json.loads(result.data.decode())['access_token']
      
              rv = self.client().post(
                  '/bucketlists/',
                  headers=dict(Authorization="Bearer " + access_token),
                  data={'name': 'Eat, pray and love'})
              self.assertEqual(rv.status_code, 201)
              
              results = json.loads(rv.data.decode())
      
              
              res = self.client().delete(
                  '/bucketlists/{}'.format(results['id']),
                  headers=dict(Authorization="Bearer " + access_token),)
              self.assertEqual(res.status_code, 200)
      
              
              result = self.client().get(
                  '/bucketlists/1',
                  headers=dict(Authorization="Bearer " + access_token))
              self.assertEqual(result.status_code, 404)
      
      
      if __name__ == "__main__":
          unittest.main()
      

      We’ll refactor the methods that handle the HTTP requests for bucketlist creation and getting all the bucketlists. Open up /app/__init__.py file and edit as follows:

      
      
      
      from flask import request, jsonify, abort, make_response
      
      def create_app(config_name):
          from models import Bucketlist, User
      
          
          
          
      
          @app.route('/bucketlists/', methods=['POST', 'GET'])
          def bucketlists():
              
              auth_header = request.headers.get('Authorization')
              access_token = auth_header.split(" ")[1]
      
              if access_token:
             
                  user_id = User.decode_token(access_token)
                  if not isinstance(user_id, str):
                      
      
                      if request.method == "POST":
                          name = str(request.data.get('name', ''))
                          if name:
                              bucketlist = Bucketlist(name=name, created_by=user_id)
                              bucketlist.save()
                              response = jsonify({
                                  'id': bucketlist.id,
                                  'name': bucketlist.name,
                                  'date_created': bucketlist.date_created,
                                  'date_modified': bucketlist.date_modified,
                                  'created_by': user_id
                              })
      
                              return make_response(response), 201
      
                      else:
                          
                          bucketlists = Bucketlist.query.filter_by(created_by=user_id)
                          results = []
      
                          for bucketlist in bucketlists:
                              obj = {
                                  'id': bucketlist.id,
                                  'name': bucketlist.name,
                                  'date_created': bucketlist.date_created,
                                  'date_modified': bucketlist.date_modified,
                                  'created_by': bucketlist.created_by
                              }
                              results.append(obj)
      
                          return make_response(jsonify(results)), 200
                  else:
                      
                      message = user_id
                      response = {
                          'message': message
                      }
                      return make_response(jsonify(response)), 401
      

      We first added two imports: the User model and the make_response from Flask.

      In the bucketlist function, we check for the authorization header from the request and extract the access token. Then, we decoded the token using User.decode_token(token) to give us the payload. The payload is expected to be a user ID if the token is valid and not expired. If the token is not valid or expired, the payload will be an error message as a string.

      Create a bucketlist or two

      Copy the token and paste it to the header section, creating an Authorization header. Don’t forget to put the word Bearer before the token with a space separating them like this:

      Authorization:   "Bearer dfg32r22349r40eiwoijr232394029wfopi23r2.2342..."
      

      Make a POST request to localhost:5000/bucketlists/, specifying the name of the bucketlist. Click send.

      Get all bucketlists for a given user

      Ensure you’ve set the Authorization header just as we did for the POST request.

      Make a GET request to localhost:5000/bucketlists/ and retrieve all the bucketlists our user just created.

      We’ll refactor the PUT and DELETE functionality the same way we tackled the GET and POST.

      
      
      
      
      from flask import request, jsonify, abort, make_response
      
      def create_app(config_name):
          from models import Bucketlist, User
      
          
          
          
      
          @app.route('/bucketlists/', methods=['POST', 'GET'])
          def bucketlists():
              
              
      
          @app.route('/bucketlists/<int:id>', methods=['GET', 'PUT', 'DELETE'])
          def bucketlist_manipulation(id, **kwargs):
              
              auth_header = request.headers.get('Authorization')
              access_token = auth_header.split(" ")[1]
      
              if access_token:
                  
                  user_id = User.decode_token(access_token)
      
                  if not isinstance(user_id, str):
                      
                      
                      bucketlist = Bucketlist.query.filter_by(id=id).first()
                      if not bucketlist:
                          
                          
                          abort(404)
      
                      if request.method == "DELETE":
                          
                          bucketlist.delete()
                          return {
                              "message": "bucketlist {} deleted".format(bucketlist.id)
                          }, 200
      
                      elif request.method == 'PUT':
                          
                          name = str(request.data.get('name', ''))
      
                          bucketlist.name = name
                          bucketlist.save()
      
                          response = {
                              'id': bucketlist.id,
                              'name': bucketlist.name,
                              'date_created': bucketlist.date_created,
                              'date_modified': bucketlist.date_modified,
                              'created_by': bucketlist.created_by
                          }
                          return make_response(jsonify(response)), 200
                      else:
                          
                          response = {
                              'id': bucketlist.id,
                              'name': bucketlist.name,
                              'date_created': bucketlist.date_created,
                              'date_modified': bucketlist.date_modified,
                              'created_by': bucketlist.created_by
                          }
                          return make_response(jsonify(response)), 200
                  else:
                      
                      message = user_id
                      response = {
                          'message': message
                      }
                      
                      return make_response(jsonify(response)), 401
      
          
          from .auth import auth_blueprint
          app.register_blueprint(auth_blueprint)
      
          return app
      

      Running python manage.py test should now yield passing tests.

      1. test_already_registered_user (test_auth.AuthTestCase)
      2. Test that a user cannot be registered twice. ... ok
      3. test_non_registered_user_login (test_auth.AuthTestCase)
      4. Test non registered users cannot login. ... ok
      5. test_registration (test_auth.AuthTestCase)
      6. Test user registration works correcty. ... ok
      7. test_user_login (test_auth.AuthTestCase)
      8. Test registered user can login. ... ok
      9. test_api_can_get_all_bucketlists (test_bucketlist.BucketlistTestCase)
      10. Test API can get a bucketlist (GET request). ... ok
      11. test_api_can_get_bucketlist_by_id (test_bucketlist.BucketlistTestCase)
      12. Test API can get a single bucketlist by using it's id. ... ok
      13. test_bucketlist_can_be_edited (test_bucketlist.BucketlistTestCase)
      14. Test API can edit an existing bucketlist. (PUT request) ... ok
      15. test_bucketlist_creation (test_bucketlist.BucketlistTestCase)
      16. Test API can create a bucketlist (POST request) ... ok
      17. test_bucketlist_deletion (test_bucketlist.BucketlistTestCase)
      18. Test API can delete an existing bucketlist. (DELETE request). ... ok
      19. ----------------------------------------------------------------------
      20. Ran 9 tests in 1.579s
      21. OK

      Now let’s test to see if it works on Postman.

      Fire up the API using python run.py development

      Make a GET request for a single bucketlist to localhost:5000/bucketlists/2

      Feel free to play around with the PUT and DELETE functionality.

      We’ve covered quite a lot on securing our API. We went through defining a user model and integrating users into our API. We also covered token-based authentication and used an authentication blueprint to implement it.

      Even though our main focus is to write the code, we should not let testing be an afterthought.
      For us to improve on code quality, there have to be tests. Testing is the secret to increasing the agility of your product development. In everything project you do, put TTD first.

      If you’ve coded this to the end, you are awesome!

      Feel free to recommend this to friends and colleagues.

      How To Use a PostgreSQL Database in a Flask Application


      The author selected the Free and Open Source Fund to receive a donation as part of the Write for DOnations program.

      Introduction

      In web applications, you usually need a database, which is an organized collection of data. You use a database to store and maintain persistent data that can be retrieved and manipulated efficiently. For example, in a social media application, you have a database where user data (personal information, posts, comments, followers) is stored in a way that can be efficiently manipulated. You can add data to a database, retrieve it, modify it, or delete it, depending on different requirements and conditions. In a web application, these requirements might be a user adding a new post, deleting a post, or deleting their account, which might or might not delete their posts. The actions you perform to manipulate data will depend on specific features in your application. For example, you might not want users to add posts with no titles.

      Flask is a lightweight Python web framework that provides useful tools and features for creating web applications in the Python Language. PostgreSQL, or Postgres, is a relational database management system that provides an implementation of the SQL querying language. It’s standards-compliant and has many advanced features such as reliable transactions and concurrency without read locks.

      In this tutorial, you’ll build a small book review web application that demonstrates how to use the psycopg2 library, a PostgreSQL database adapter that allows you to interact with your PostgreSQL database in Python. You’ll use it with Flask to perform basic tasks, such as connecting to a database server, creating tables, inserting data to a table, and retrieving data from a table.

      Prerequisites

      Step 1 — Creating the PostgreSQL Database and User

      In this step, you’ll create a database called flask_db and a database user called sammy for your Flask application.

      During the Postgres installation, an operating system user named postgres was created to correspond to the postgres PostgreSQL administrative user. You need to use this user to perform administrative tasks. You can use sudo and pass in the username with the -iu option.

      Log in to an interactive Postgres session using the following command:

      You will be given a PostgreSQL prompt where you can set up your requirements.

      First, create a database for your project:

      • CREATE DATABASE flask_db;

      Note: Every Postgres statement must end with a semi-colon, so make sure that your command ends with one if you are experiencing issues.

      Next, create a database user for our project. Make sure to select a secure password:

      • CREATE USER sammy WITH PASSWORD "https://www.digitalocean.com/community/tutorials/password';

      Then give this new user access to administer your new database:

      • GRANT ALL PRIVILEGES ON DATABASE flask_db TO sammy;

      To confirm the database was created, get the list of databases by typing the following command:

      You’ll see flask_db in the list of databases.

      When you are finished, exit out of the PostgreSQL prompt by typing:

      Postgres is now set up so that you can connect to and manage its database information via Python using the psycopg2 library. Next, you’ll install this library alongside the Flask package.

      Step 2 — Installing Flask and psycopg2

      In this step, you will install Flask and the psycopg2 library so that you can interact with your database using Python.

      With your virtual environment activated, use pip to install Flask and the psycopg2 library:

      • pip install Flask psycopg2-binary

      Once the installation is successfully finished, you’ll see a line similar to the following at the end of the output:

      Output

      Successfully installed Flask-2.0.2 Jinja2-3.0.3 MarkupSafe-2.0.1 Werkzeug-2.0.2 click-8.0.3 itsdangerous-2.0.1 psycopg2-binary-2.9.2

      You now have the required packages installed on your virtual environment. Next, you’ll connect to and set up your database.

      Step 3 — Setting up a Database

      In this step, you’ll create a Python file in your flask_app project directory to connect to the flask_db database, create a table for storing books, and insert some books with reviews into it.

      First with your programming environment activated, open a new file called init_db.py in your flask_app directory.

      This file will open a connection to the flask_db database, create a table called books, and populate the table using sample data. Add the following code to it:

      flask_app/init_db.py

      import os
      import psycopg2
      
      conn = psycopg2.connect(
              host="localhost",
              database="flask_db",
              user=os.environ['DB_USERNAME'],
              password=os.environ['DB_PASSWORD'])
      
      # Open a cursor to perform database operations
      cur = conn.cursor()
      
      # Execute a command: this creates a new table
      cur.execute('DROP TABLE IF EXISTS books;')
      cur.execute('CREATE TABLE books (id serial PRIMARY KEY,'
                                       'title varchar (150) NOT NULL,'
                                       'author varchar (50) NOT NULL,'
                                       'pages_num integer NOT NULL,'
                                       'review text,'
                                       'date_added date DEFAULT CURRENT_TIMESTAMP);'
                                       )
      
      # Insert data into the table
      
      cur.execute('INSERT INTO books (title, author, pages_num, review)'
                  'VALUES (%s, %s, %s, %s)',
                  ('A Tale of Two Cities',
                   'Charles Dickens',
                   489,
                   'A great classic!')
                  )
      
      
      cur.execute('INSERT INTO books (title, author, pages_num, review)'
                  'VALUES (%s, %s, %s, %s)',
                  ('Anna Karenina',
                   'Leo Tolstoy',
                   864,
                   'Another great classic!')
                  )
      
      conn.commit()
      
      cur.close()
      conn.close()
      

      Save and close the file.

      In this file, you first import the os module you’ll use to access environment variables where you’ll store your database username and password so that they are not visible in your source code.

      You import the psycopg2 library. Then you open a connection to the flask_db database using the psycopg2.connect() function. You specify the host, which is the localhost in this case. You pass the database name to the database parameter.

      You provide your username and password via the os.environ object, which gives you access to environment variables you set in your programming environment. You will store the database username in an environment variable called DB_USERNAME and the password in an environment variable called DB_PASSWORD. This allows you to store your username and password outside your source code, so that your sensitive information is not leaked when the source code is saved in source control or uploaded to a server on the internet. Even if an attacker gains access to your source code, they will not gain access to the database.

      You create a cursor called cur using the connection.cursor() method, which allows Python code to execute PostgreSQL commands in a database session.

      You use the cursor’s execute() method to delete the books table if it already exists. This avoids the possibility of another table named books existing, which might result in confusing behavior (for example, if it has different columns). This isn’t the case here, because you haven’t created the table yet, so the SQL command won’t be executed. Note that this will delete all of the existing data whenever you execute this init_db.py file. For our purposes, you will only execute this file once to initiate the database, but you might want to execute it again to delete whatever data you inserted and start with the initial sample data again.

      Then you use CREATE TABLE books to create a table named books with the following columns:

      • id: An ID of the serial type, which is an autoincrementing integer. This column represents a primary key you specify using the PRIMARY KEY keywords. The database will assign a unique value to this key for each entry.
      • title: The book’s title of the varchar type, which is a character type of variable length with a limit. varchar (150) means that the title can be up to 150 characters long. NOT NULL signifies that this column can’t be empty.
      • author: The book’s author, with a limit of 50 characters. NOT NULL signifies that this column can’t be empty.
      • pages_num: An integer representing the number of pages the book has. NOT NULL signifies that this column can’t be empty.
      • review: The book review. The text type signifies that the review can be text of any length.
      • date_added: The date the book was added to the table. DEFAULT sets the default value of the column to CURRENT_TIMESTAMP, which is the time at which the book was added to the database. Just like id, you don’t need to specify a value for this column, as it will be automatically filled in.

      After creating the table, you use the cursor’s execute() method to insert two books into the table, A Tale of Two Cities by Charles Dickens, and Anna Karenina by Leo Tolstoy. You use the %s placeholder to pass the values to the SQL statement. psycopg2 handles the insertion in the background in a way that prevents SQL Injection attacks.

      Once you finish inserting book data into your table, you use the connection.commit() method to commit the transaction and apply the changes to the database. Then you clean things up by closing the cursor with cur.close(), and the connection with conn.close().

      For the database connection to be established, set the DB_USERNAME and DB_PASSWORD environment variables by running the following commands. Remember to use your own username and password:

      • export DB_USERNAME="https://www.digitalocean.com/community/tutorials/sammy"
      • export DB_PASSWORD="https://www.digitalocean.com/community/tutorials/password"

      Now, run your init_db.py file in the terminal using the python command:

      Once the file finishes execution with no errors, a new books table will be added to your flask_db database.

      Log in to an interactive Postgres session to check out the new books table.

      Connect to the flask_db database using the c command:

      Then use a SELECT statement to get the titles and authors of books from the books table:

      • SELECT title, author FROM books;

      You’ll see an output like the following:

              title         |      author
      ----------------------+------------------
       A Tale of Two Cities | Charles Dickens
       Anna Karenina        | Leo Tolstoy
      

      Quit the interactive session with q.

      Next, you’ll create a small Flask application, connect to the database, retrieve the two book reviews you inserted into the database, and display them on the index page.

      Step 4 — Displaying Books

      In this step, you’ll create a Flask application with an index page that retrieves the books that are in the database, and display them.

      With your programming environment activated and Flask installed, open a file called app.py for editing inside your flask_app directory:

      This file will set up your database connection and create a single Flask route to use that connection. Add the following code to the file:

      flask_app/app.py

      import os
      import psycopg2
      from flask import Flask, render_template
      
      app = Flask(__name__)
      
      def get_db_connection():
          conn = psycopg2.connect(host="localhost",
                                  database="https://www.digitalocean.com/community/tutorials/flask_db",
                                  user=os.environ['DB_USERNAME'],
                                  password=os.environ['DB_PASSWORD'])
          return conn
      
      
      @app.route('/')
      def index():
          conn = get_db_connection()
          cur = conn.cursor()
          cur.execute('SELECT * FROM books;')
          books = cur.fetchall()
          cur.close()
          conn.close()
          return render_template('index.html', books=books)
      

      Save and close the file.

      Here, you import the os module, the psycopg2 library, and the Flask class and the render_template() from the flask package. You make a Flask application instance called app.

      You define a function called get_db_connection(), which opens a connection to the flask_db database using the user and password you store in your DB_USERNAME and DB_PASSWORD environment variables. The function returns the conn connection object you’ll be using to access the database.

      Then you create a main / route and an index() view function using the app.route() decorator. In the index() view function, you open a database connection using the get_db_connection() function, you create a cursor, and execute the SELECT * FROM books; SQL statement to get all the books that are in the database. You use the fetchall() method to save the data in a variable called books. Then you close the cursor and the connection. Lastly, you return a call to the render_template() function to render a template file called index.html passing it the list of books you fetched from the database in the books variable.

      To display the books you have in your database on the index page, you will first create a base template, which will have all the basic HTML code other templates will also use to avoid code repetition. Then you’ll create the index.html template file you rendered in your index() function. To learn more about templates, see How to Use Templates in a Flask Application.

      Create a templates directory, then open a new template called base.html:

      • mkdir templates
      • nano templates/base.html

      Add the following code inside the base.html file:

      flask_app/templates/base.html

      <!DOCTYPE html>
      <html lang="en">
      <head>
          <meta charset="UTF-8">
          <title>{% block title %} {% endblock %}- FlaskApp</title>
          <style>
              nav a {
                  color: #d64161;
                  font-size: 3em;
                  margin-left: 50px;
                  text-decoration: none;
              }
      
              .book {
                  padding: 20px;
                  margin: 10px;
                  background-color: #f7f4f4;
              }
      
              .review {
                      margin-left: 50px;
                      font-size: 20px;
              }
      
          </style>
      </head>
      <body>
          <nav>
              <a href="https://www.digitalocean.com/community/tutorials/{{ url_for('index') }}">FlaskApp</a>
              <a href="#">About</a>
          </nav>
          <hr>
          <div class="content">
              {% block content %} {% endblock %}
          </div>
      </body>
      </html>
      

      Save and close the file.

      This base template has all the HTML boilerplate you’ll need to reuse in your other templates. The title block will be replaced to set a title for each page, and the content block will be replaced with the content of each page. The navigation bar has two links, one for the index page where you use the url_for() helper function to link to the index() view function, and the other for an About page if you choose to include one in your application.

      Next, open a template called index.html. This is the template you referenced in the app.py file:

      • nano templates/index.html

      Add the following code to it:

      flask_app/templates/index.html

      
      {% extends 'base.html' %}
      
      {% block content %}
          <h1>{% block title %} Books {% endblock %}</h1>
          {% for book in books %}
              <div class="book">
                  <h3>#{{ book[0] }} - {{ book[1] }} BY {{ book[2] }}</h3>
                  <i><p>({{ book[3] }} pages)</p></i>
                  <p class="review">{{ book[4] }}</p>
                  <i><p>Added {{ book[5] }}</p></i>
              </div>
          {% endfor %}
      {% endblock %}
      

      Save and close the file.

      In this file, you extend the base template, and replace the contents of the content block. You use an <h1> heading that also serves as a title.

      You use a Jinja for loop in the line {% for book in books %} to go through each book in the books list. You display the book ID, which is the first item using book[0]. You then display the book title, author, number of pages, review, and the date the book was added.

      While in your flask_app directory with your virtual environment activated, tell Flask about the application (app.py in this case) using the FLASK_APP environment variable. Then set the FLASK_ENV environment variable to development to run the application in development mode and get access to the debugger. For more information about the Flask debugger, see How To Handle Errors in a Flask Application. Use the following commands to do this:

      • export FLASK_APP=app
      • export FLASK_ENV=development

      Make sure you set the DB_USERNAME and DB_PASSWORD environment variables if you haven’t already:

      • export DB_USERNAME="https://www.digitalocean.com/community/tutorials/sammy"
      • export DB_PASSWORD="https://www.digitalocean.com/community/tutorials/password"

      Next, run the application:

      With the development server running, visit the following URL using your browser:

      http://127.0.0.1:5000/
      

      You’ll see the books you added to the database on the first initiation.

      Index Page

      You’ve displayed the books in your database on the index page. You now need to allow users to add new books. You’ll add a new route for adding books in the next step.

      Step 5 — Adding New Books

      In this step, you’ll create a new route for adding new books and reviews to the database.

      You’ll add a page with a web form where users enter the book title, book author, the number of pages, and the book review.

      Leave the development server running and open a new terminal window.

      First, open your app.py file:

      For handling the web form, you’ll need to import a few things from the flask package:

      • The global request object to access submitted data.
      • The url_for() function to generate URLs.
      • The redirect() function to redirect users to the index page after adding a book to the database.

      Add these imports to the first line in the file:

      flask_app/app.py

      
      from flask import Flask, render_template, request, url_for, redirect
      
      # ...
      

      Then add the following route at the end of the app.py file:

      flask_app/app.py

      
      # ...
      
      
      @app.route('/create/', methods=('GET', 'POST'))
      def create():
          return render_template('create.html')
      

      Save and close the file.

      In this route, you pass the tuple ('GET', 'POST') to the methods parameter to allow both GET and POST requests. GET requests are used to retrieve data from the server. POST requests are used to post data to a specific route. By default, only GET requests are allowed. When the user first requests the /create route using a GET request, a template file called create.html will be rendered. You will later edit this route to handle POST requests for when users fill and submit the web form for adding new books.

      Open the new create.html template:

      • nano templates/create.html

      Add the following code to it:

      flask_app/templates/create.html

      {% extends 'base.html' %}
      
      {% block content %}
          <h1>{% block title %} Add a New Book {% endblock %}</h1>
          <form method="post">
              <p>
                  <label for="title">Title</label>
                  <input type="text" name="title"
                         placeholder="Book title">
                  </input>
              </p>
      
              <p>
                  <label for="author">Author</label>
                  <input type="text" name="author"
                         placeholder="Book author">
                  </input>
              </p>
      
              <p>
                  <label for="pages_num">Number of pages</label>
                  <input type="number" name="pages_num"
                         placeholder="Number of pages">
                  </input>
              </p>
              <p>
              <label for="review">Review</label>
              <br>
              <textarea name="review"
                        placeholder="Review"
                        rows="15"
                        cols="60"
                        ></textarea>
              </p>
              <p>
                  <button type="submit">Submit</button>
              </p>
          </form>
      {% endblock %}
      

      Save and close the file.

      You extend the base template, set a heading as a title, and use a <form> tag with the attribute method set to post to indicate that the form will submit a POST request.

      You have a text field with the name title, which you’ll use to access the title data in your /create route.

      You have a text field for the author, a number field for the number of pages, and a text area for the book review.

      Last, you have a Submit button at the end of the form.

      Now, with the development server running, use your browser to navigate to the /create route:

      http://127.0.0.1:5000/create
      

      You will see an Add a New Book page with an input field for a book title, one for its author, and one for the number of pages the book has, a text area for the book’s review, and a Submit button.

      Add a New Book

      If you fill in the form and submit it, sending a POST request to the server, nothing happens because you did not handle POST requests on the /create route.

      Open app.py to handle the POST request the user submits:

      Edit the /create route to look as follows:

      flask_app/app.py

      
      # ...
      
      @app.route('/create/', methods=('GET', 'POST'))
      def create():
          if request.method == 'POST':
              title = request.form['title']
              author = request.form['author']
              pages_num = int(request.form['pages_num'])
              review = request.form['review']
      
              conn = get_db_connection()
              cur = conn.cursor()
              cur.execute('INSERT INTO books (title, author, pages_num, review)'
                          'VALUES (%s, %s, %s, %s)',
                          (title, author, pages_num, review))
              conn.commit()
              cur.close()
              conn.close()
              return redirect(url_for('index'))
      
          return render_template('create.html')
      

      Save and close the file.

      You handle POST requests inside the if request.method == 'POST' condition. You extract the title, author, number of pages, and the review the user submits from the request.form object.

      You open a database using the get_db_connection() function, and create a cursor. Then you execute an INSERT INTO SQL statement to insert the title, author, number of pages, and review the user submitted into the books table.

      You commit the transaction and close the cursor and connection.

      Lastly, you redirect the user to the index page where they can see the newly added book below the existing books.

      With the development server running, use your browser to navigate to the /create route:

      http://127.0.0.1:5000/create
      

      Fill in the form with some data and submit it.

      You’ll be redirected to the index page where you’ll see your new book review.

      Next, you’ll add a link to the Create page in the navigation bar. Open base.html:

      Edit the file to look as follows:

      flask_app/templates/base.html

      
      <!DOCTYPE html>
      <html lang="en">
      <head>
          <meta charset="UTF-8">
          <title>{% block title %} {% endblock %} - FlaskApp</title>
          <style>
              nav a {
                  color: #d64161;
                  font-size: 3em;
                  margin-left: 50px;
                  text-decoration: none;
              }
      
              .book {
                  padding: 20px;
                  margin: 10px;
                  background-color: #f7f4f4;
              }
      
              .review {
                      margin-left: 50px;
                      font-size: 20px;
              }
      
          </style>
      </head>
      <body>
          <nav>
              <a href="https://www.digitalocean.com/community/tutorials/{{ url_for("index') }}">FlaskApp</a>
              <a href="https://www.digitalocean.com/community/tutorials/{{ url_for("create') }}">Create</a>
              <a href="#">About</a>
          </nav>
          <hr>
          <div class="content">
              {% block content %} {% endblock %}
          </div>
      </body>
      </html>
      

      Save and close the file.

      Here, you add a new <a> link to the navigation bar that points to the Create page.

      Refresh your index page and you’ll see the new link in the navigation bar.

      You now have a page with a web form for adding new book reviews. For more on web forms, see How To Use Web Forms in a Flask Application. For a more advanced and more secure method of managing web forms, see How To Use and Validate Web Forms with Flask-WTF.

      Conclusion

      You built a small web application for book reviews that communicates with a PostgreSQL database. You have basic database functionality in your Flask application, such as adding new data to the database, retrieving data, and displaying it on a page.

      If you would like to read more about Flask, check out the other tutorials in the Flask series.



      Source link

      How To Build and Deploy a Flask Application Using Docker on Ubuntu 20.04


      The author selected the Tech Education Fund to receive a donation as part of the Write for DOnations program.

      Introduction

      Docker is an open-source application that allows administrators to create, manage, deploy, and replicate applications using containers. Containers can be thought of as a package that houses dependencies that an application requires to run at an operating system level. This means that each application deployed using Docker lives in an environment of its own and its requirements are handled separately.

      Flask is a web micro-framework that is built on Python. It is called a micro-framework because it does not require specific tools or plug-ins to run. The Flask framework is lightweight and flexible, yet highly structured, making it especially popular for small web apps written in Python.

      Deploying a Flask application with Docker will allow you to replicate the application across different servers with minimal reconfiguration.

      In this tutorial, you will create a Flask application and deploy it with Docker. This tutorial will also cover how to update an application after deployment.

      Prerequisites

      To follow this tutorial, you will need the following:

      Step 1 — Setting Up the Flask Application

      To get started, you will create a directory structure that will hold your Flask application. This tutorial will create a directory called TestApp in /var/www, but you can modify the command to name it whatever you’d like.

      • sudo mkdir /var/www/TestApp

      Move in to the newly created TestApp directory:

      Next, create the base folder structure for the Flask application:

      • sudo mkdir -p app/static app/templates

      The -p flag indicates that mkdir will create a directory and all parent directories that don’t exist. In this case, mkdir will create the app parent directory in the process of making the static and templates directories.

      The app directory will contain all files related to the Flask application such as its views and blueprints. Views are the code you write to respond to requests to your application. Blueprints create application components and support common patterns within an application or across multiple applications.

      The static directory is where assets such as images, CSS, and JavaScript files live. The templates directory is where you will put the HTML templates for your project.

      Now that the base folder structure is complete, you need to create the files needed to run the Flask application. First, create an __init__.py file inside the app directory using nano or a text editor of your choice. This file tells the Python interpreter that the app directory is a package and should be treated as such.

      Run the following command to create the file:

      • sudo nano app/__init__.py

      Packages in Python allow you to group modules into logical namespaces or hierarchies. This approach enables the code to be broken down into individual and manageable blocks that perform specific functions.

      Next, you will add code to the __init__.py that will create a Flask instance and import the logic from the views.py file, which you will create after saving this file. Add the following code to your new file:

      /var/www/TestApp/app/__init__.py

      from flask import Flask
      app = Flask(__name__)
      from app import views
      

      Once you’ve added that code, save and close the file. You can save and close the file by pressing Ctrl+X, then when prompted, Y and Enter.

      With the __init__.py file created, you’re ready to create the views.py file in your app directory. This file will contain most of your application logic.

      Next, add the code to your views.py file. This code will return the hello world! string to users who visit your web page:

      /var/www/TestApp/app/views.py

      from app import app
      
      @app.route('/')
      def home():
         return "hello world!"
      

      The @app.route line above the function is called a decorator. Decorators are a Python language convention that are widely used by Flask; their purpose is to modify the functions immediately following them. In this case, the decorator tells Flask which URL will trigger the home() function. The hello world text returned by the home function will be displayed to the user on the browser.

      With the views.py file in place, you’re ready to create the uwsgi.ini file. This file will contain the uWSGI configurations for our application. uWSGI is a deployment option for Nginx that is both a protocol and an application server; the application server can serve uWSGI, FastCGI, and HTTP protocols.

      To create this file, run the following command:

      Next, add the following content to your file to configure the uWSGI server:

      /var/www/TestApp/uwsgi.ini

      [uwsgi]
      module = main
      callable = app
      master = true
      

      This code defines the module that the Flask application will be served from. In this case, this is the main.py file, referenced here as main. The callable option instructs uWSGI to use the app instance exported by the main application. The master option allows your application to keep running, so there is little downtime even when reloading the entire application.

      Next, create the main.py file, which is the entry point to the application. The entry point instructs uWSGI on how to interact with the application.

      Next, copy and paste the following into the file. This imports the Flask instance named app from the application package that was previously created.

      /var/www/TestApp/main.py

      from app import app
      

      Finally, create a requirements.txt file to specify the dependencies that the pip package manager will install to your Docker deployment:

      • sudo nano requirements.txt

      Add the following line to add Flask as a dependency:

      /var/www/TestApp/requirements.txt

      Flask>=2.0.2
      

      This specifies the version of Flask to be installed. At the time of writing this tutorial, 2.0.2 is the latest Flask version, and specifying >=2.0.2 will ensure you get version 2.0.2 or newer. Because you’re making a basic test app in this tutorial, the syntax is unlikely to go out of date due to future updates to Flask, but if you wanted to be safe and still receive minor updates, you could specify that you don’t want to install a future major version by specifying something like Flask>=2.0.2,<3.0. You can check for updates at the official website for Flask, or on the Python Package Index’s landing page for the Flask library.

      Save and close the file. You have successfully set up your Flask application and are ready to set up Docker.

      Step 2 — Setting Up Docker

      In this step you will create two files, Dockerfile and start.sh, to create your Docker deployment. The Dockerfile is a text document that contains the commands used to assemble the image. The start.sh file is a shell script that will build an image and create a container from the Dockerfile.

      First, create the Dockerfile.

      Next, add your desired configuration to the Dockerfile. These commands specify how the image will be built, and what extra requirements will be included.

      /var/www/TestApp/Dockerfile

      FROM tiangolo/uwsgi-nginx-flask:python3.8-alpine
      RUN apk --update add bash nano
      ENV STATIC_URL /static
      ENV STATIC_PATH /var/www/app/static
      COPY ./requirements.txt /var/www/requirements.txt
      RUN pip install -r /var/www/requirements.txt
      

      In this example, the Docker image will be built off an existing image, tiangolo/uwsgi-nginx-flask, which you can find on DockerHub. This particular Docker image is a good choice over others because it supports a wide range of Python versions and OS images.

      The first two lines specify the parent image that you’ll use to run the application and install the bash command processor and the nano text editor. It also installs the git client for pulling and pushing to version control hosting services such as GitHub, GitLab, and Bitbucket. ENV STATIC_URL /static is an environment variable specific to this Docker image. It defines the static folder where all assets such as images, CSS files, and JavaScript files are served from.

      The last two lines will copy the requirements.txt file into the container so that it can be executed, and then parses the requirements.txt file to install the specified dependencies.

      Save and close the file after adding your configuration.

      With your Dockerfile in place, you’re almost ready to write your start.sh script that will build the Docker container. Before writing the start.sh script, first make sure that you have an open port to use in the configuration. To check if a port is free, run the following command:

      • sudo nc localhost 56733 < /dev/null; echo $?

      If the output of the command above is 1, then the port is free and usable. Otherwise, you will need to select a different port to use in your start.sh configuration file.

      Once you’ve found an open port to use, create the start.sh script:

      The start.sh script is a shell script that will build an image from the Dockerfile and create a container from the resulting Docker image. Add your configuration to the new file:

      /var/www/TestApp/start.sh

      #!/bin/bash
      app="docker.test"
      docker build -t ${app} .
      docker run -d -p 56733:80 
        --name=${app} 
        -v $PWD:/app ${app}
      

      The first line is called a shebang. It specifies that this is a bash file and will be executed as commands. The next line specifies the name you want to give the image and container and saves as a variable named app. The next line instructs Docker to build an image from your Dockerfile located in the current directory. This will create an image called docker.test in this example.

      The last three lines create a new container named docker.test that is exposed at port 56733. Finally, it links the present directory to the /var/www directory of the container.

      You use the -d flag to start a container in daemon mode, or as a background process. You include the -p flag to bind a port on the server to a particular port on the Docker container. In this case, you are binding port 56733 to port 80 on the Docker container. The -v flag specifies a Docker volume to mount on the container, and in this case, you are mounting the entire project directory to the /var/www folder on the Docker container.

      Save and close the file after adding your configuration.

      Execute the start.sh script to create the Docker image and build a container from the resulting image:

      Once the script finishes running, use the following command to list all running containers:

      You will receive output that shows the containers:

      Output

      CONTAINER ID IMAGE COMMAND CREATED STATUS PORTS NAMES 58b05508f4dd docker.test "/entrypoint.sh /sta…" 12 seconds ago Up 3 seconds 443/tcp, 0.0.0.0:56733->80/tcp docker.test

      You will find that the docker.test container is running. Now that it is running, visit the IP address at the specified port in your browser: http://ip-address:56733

      You’ll see a page similar to the following:

      the home page

      In this step you have successfully deployed your Flask application on Docker. Next, you will use templates to display content to users.

      Step 3 — Serving Template Files

      Templates are files that display static and dynamic content to users who visit your application. In this step, you will create a HTML template to create a homepage for the application.

      Start by creating a home.html file in the app/templates directory:

      • sudo nano app/templates/home.html

      Add the code for your template. This code will create an HTML5 page that contains a title and some text.

      /var/www/TestApp/app/templates/home.html

      
      <!doctype html>
      
      <html lang="en-us">   
        <head>
          <meta charset="utf-8">
          <meta http-equiv="x-ua-compatible" content="ie=edge">
          <title>Welcome home</title>
        </head>
      
        <body>
          <h1>Home Page</h1>
          <p>This is the home page of our application.</p>
        </body> 
      </html>
      

      Save and close the file once you’ve added your template.

      Next, modify the app/views.py file to serve the newly created file:

      First, add the following line at the beginning of your file to import the render_template method from Flask. This method parses an HTML file to render a web page to the user.

      /var/www/TestApp/app/views.py

      from flask import render_template
      ...
      

      At the end of the file, you will also add a new route to render the template file. This code specifies that users are served the contents of the home.html file whenever they visit the /template route on your application.

      /var/www/TestApp/app/views.py

      ...
      
      @app.route('/template')
      def template():
          return render_template('home.html')
      

      The updated app/views.py file will look like this:

      /var/www/TestApp/app/views.py

      from flask import render_template
      from app import app 
      
      @app.route('/')
      def home():
          return "Hello world!"
      
      @app.route('/template')
      def template():
          return render_template('home.html')
      

      Save and close the file when done.

      In order for these changes to take effect, you will need to stop and restart the Docker containers. Run the following command to rebuild the container:

      • sudo docker stop docker.test && sudo docker start docker.test

      Visit your application at http://your-ip-address:56733/template to see the new template being served.

      homepage

      In this you’ve created a Docker template file to serve visitors on your application. In the next step you will see how the changes you make to your application can take effect without having to restart the Docker container.

      Step 4 — Updating the Application

      Sometimes you will need to make changes to the application, whether it is installing new requirements, updating the Docker container, or HTML and logic changes. In this section, you will configure touch-reload to make these changes without needing to restart the Docker container.

      Python autoreloading watches the entire file system for changes and refreshes the application when it detects a change. Autoreloading is discouraged in production because it can become resource intensive very quickly. In this step, you will use touch-reload to watch for changes to a particular file and reload when the file is updated or replaced.

      To implement this, start by opening your uwsgi.ini file:

      Next, add the highlighted line to the end of the file:

      /var/www/TestApp/uwsgi.ini

      module = main
      callable = app
      master = true
      touch-reload = /app/uwsgi.ini
      

      This specifies a file that will be modified to trigger an entire application reload. Once you’ve made the changes, save and close the file.

      To demonstrate this, make a small change to your application. Start by opening your app/views.py file:

      Replace the string returned by the home function:

      /var/www/TestApp/app/views.py

      from flask import render_template
      from app import app
      
      @app.route('/')
      def home():
          return "<b>There has been a change</b>"
      
      @app.route('/template')
      def template():
          return render_template('home.html')
      

      Save and close the file after you’ve made a change.

      Next, if you open your application’s homepage at http://ip-address:56733, you will notice that the changes are not reflected. This is because the condition for reload is a change to the uwsgi.ini file. To reload the application, use touch to activate the condition:

      Reload the application homepage in your browser again. You will find that the application has incorporated the changes:

      Homepage Updated

      In this step, you set up a touch-reload condition to update your application after making changes.

      Conclusion

      In this tutorial, you created and deployed a Flask application to a Docker container. You also configured touch-reload to refresh your application without needing to restart the container.

      With your new application on Docker, you can now scale with ease. To learn more about using Docker, check out their official documentation.



      Source link