[python]Deploy flask api with gunicorn and nginx

NGINX is a load balancer and a reverse proxy, the purpose is to interface between user and WSGI server. Web server cannot directly interact with Flask like php, to use the Flask web application a WSGI server is required, in this case Gunicorn is a WSGI server. Gunicorn will run the python flask application and interact with Nginx.

Lots of documentation suggested to use virtualenv, but this may not be necessary if the ubuntu server is meant to serve just this flask api.

Also lots of sites and training suggest to deploy to Digital Ocean and Heroku, almost I cannot find any documentation to deploy for an on premise server, so this post is a reminder on how it can be done quickly.

Nginx configuration

server {
listen 80;
real_ip_header X-Forwarded-For;
server_name localhost;

location / {
proxy_pass "";

The nginx config is to tell that it is listening as a http, and will act as a proxy for the WSGI server which is defined in the proxy_pass. So to the web API user it only needs to go to http://nginx_address/contacts to get the list of contacts, the /contacts is an API endpoint which nginx interacts with gunicorn. Create the soft link of the nginx.conf to /etc/nginx/sites-enabled/, there is a default symbolic link file, you will need to delete the default file and replaced with the actual nginx.conf.

Gunicorn command to run flask app gunicorn -b localhost:5000 app:app & gunicorn binds itself to and listens on port 5000 and runs the app:app.

Screenshot 2019-08-27 at 5.14.43 PM
The web browser interacts with nginx, nginx interacts with gunicorn, gunicorn returns the result from the flask app back to nginx, and nginx presents back to the browser.
Screenshot 2019-08-27 at 5.16.10 PM.png
Postman adds a new entry to the web server – nginx.

Codes for the python flask app

import sqlite3
from flask import Flask
from flask_restful import Api
from contacts import Register, ContactList, Contact

Create contacts table for contacts database (contacts.db).
If the file contacts.db does not exist create one.
There are 7 columns:
1. contact_id, this is primary key hence is unique, will be incremented automatically.
2. fullname, this is the concatenation of last and first name.
3. last_name, this column stores the last name.
4. first_name, this column stores the first name.
5. email_address, this column stores the email address.
6. location_address, this column stores the physical address.
7. contact_number, this column stores the contact number.
def create_table():
    query = "CREATE TABLE IF NOT EXISTS contacts (contact_id INTEGER PRIMARY KEY," \
            "fullname TEXT," \
            "last_name TEXT," \
            "first_name TEXT," \
            "email_address TEXT," \
            "location_address TEXT," \
    db = sqlite3.connect('contacts.db')
    cursor = db.cursor()

# create the web app.
app = Flask(__name__)
# create the api object.
api = Api(app)

There are three endpoints based on Postman design.
{{url}}:5000/register - Post new contact
{{url}}:5000/contacts - Get all available contacts
{{url}}:5000/contact/ - Get specific single contact based on contact_id.
 corresponds with the contact_id of the get method of class Contact.
api.add_resource(Register, '/register')
api.add_resource(ContactList, '/contacts')
api.add_resource(Contact, '/contact/')

# gunicorn will look for this.
if __name__ == '__main__':

The resources python codes

import sqlite3
from flask_restful import reqparse, Resource
import re

# Register contact resource, this resource api endpoint is /register uri
class Register(Resource):
    # Create RequestParser object
    parser = reqparse.RequestParser()
    # last_name field is mandatory, and is string.
                        help='This field cannot be left blank.'
    # first_name field is mandatory, and is string.
                        help='This field cannot be left blank.')
    # email_address field is a string
                        help="This field is a string.")
    # location_address field is a string
                        help='This field is a string')
    # contact_number field is a string
                        help='This field is a string of numbers.')
    # database table name
    TABLE_NAME = 'contacts'
    # POST method
    def post(self):
        # match email pattern
        email_pattern = "\S+@[a-zA-Z0-9]+\.[a-zA-Z]{2,5}"
        regex = re.compile(email_pattern)
        query = "INSERT INTO {} VALUES (NULL, ?, ?, ?, ?, ?, ?)".format(Register.TABLE_NAME)
        # JSON data is parsed and store to data.
        data = Register.parser.parse_args()
        # combine first and last name and store in fullname column in contacts table.
        fullname = data['first_name'] + " " + data['last_name']
        # check valid email address format, if valid then keep the data else make it to null.
        data['email_address'] = data['email_address'] if regex.match(data['email_address']) else None
        # prepare the list to be passed into the query.
        # the use of get method can pass None if the data is not in the json field.
        # get method is for dictionary only, since json is a dictionary as well.
        data_list = [
            data.get('last_name', None),
            data.get('first_name', None),
            data.get('email_address', None),
            data.get('location_address', None),
            data.get('contact_number', None)
        All sqlite3 db codes use the below 5 steps
        # Step 1: Connect db file.
        db = sqlite3.connect('contacts.db')
        # Step 2: create cursor object
        cursor = db.cursor()
        # Step 3: execute query
        cursor.execute(query, (fullname, *data_list))
        # Step 4: commit the changes.
        # Step 5: close the db.
        # Message back to requester, 201 is accepted.
        return {"message": "data registered."}, 201

# Get all contacts in a list
class ContactList(Resource):
    # GET method
    def get(self):
        # initialize the contact list.
        contact_list = []
        # Get all data from contacts table.
        query = "SELECT * FROM {}".format(Register.TABLE_NAME)
        No changes to database, hence no commit is needed.
        The database code only require 4 steps.
        Step 1: Connect to database.
        Step 2: Create cursor object.
        Step 3: Execute the SELECT query
        Step 4: Close the database.
        db = sqlite3.connect('contacts.db')
        cursor = db.cursor()
        # The returned result is a list of rows.
        result = cursor.execute(query)
        # Iterate through the list, for each row.
        for row in result:
            # Convert the row into dictionary
            # contact_list collects dictionary, each dictionary is a contact's information.
            contact_list.append({"id": row[0],
                                 "full_name": row[1],
                                 "last_name": row[2],
                                 "first_name": row[3],
                                 "email_address": row[4],
                                 "location_address": row[5],
                                 "contact_number": row[6]})
        return {"contacts": contact_list}, 200

# Search single contact by id.
class Contact(Resource):
    def find_by_id(cls, contact_id):
        query = "SELECT * FROM {} WHERE contact_id=?".format(Register.TABLE_NAME)
        db = sqlite3.connect('contacts.db')
        cursor = db.cursor()
        # query variable is in tuple, (contact_id,) is to tell python this is a tuple.
        cursor.execute(query, (contact_id,))
        # contact_id has a PRIMARY KEY constraint in the database,
        # PRIMARY KEY must be unique hence use fetchone()
        result = cursor.fetchone()
        # If result is not None i.e. if result exists, else this method returns "None"
        # The result omits contact_id, since it is already searched by contact_id.
        if result:
            return {"fullname": result[1],
                    "last_name": result[2],
                    "first_name": result[3],
                    "email_address": result[4],
                    "location_address": result[5],
                    "contact_number": result[6]}
    # GET method
    def get(self, contact_id):
        # if contact_id is found in contacts return the result with 200 ok.
        found = self.find_by_id(contact_id)
        if found:
            return found, 200
        # if contact_id is Nonetype, returns not found 404, if this is not treated 500 internal server error is raised.
            return {"message": "contact information not found"}, 404

    # delete a contact base on contact_id.
    def delete(self, contact_id):
        query = "DELETE FROM {} WHERE contact_id=?".format(Register.TABLE_NAME)
        db = sqlite3.connect('contacts.db')
        cursor = db.cursor()
        # if the contact_id is found in database delete the contact.
        if self.find_by_id(contact_id):
            cursor.execute(query, (contact_id,))
            return {"message": "Contact deleted."}, 201
        # if contact_id does not exist in database, raise a bad request 400.
            return {"message": "contact id not found"}, 400

Leave a Reply

Fill in your details below or click an icon to log in:

WordPress.com Logo

You are commenting using your WordPress.com account. Log Out /  Change )

Facebook photo

You are commenting using your Facebook account. Log Out /  Change )

Connecting to %s