DEV Community

Okedele Favour
Okedele Favour

Posted on


Sending app notifications over Whatsapp using Twilio


In this article, we are going to build an application that sends appointment notifications to users over whatsapp via twilio. According to statistics, 90.4% of the young generation are active on social media so this approach of sending notifications will be more effective in setting remiders without being intrusive.

User Journey

An API request is sent to the application containing the appointment details (title, description, phone and time). Once the appointment time is reached, the user gets notified on whatsapp about the appointment.

Tutorial Requirements

To follow this tutorial, you are expected to:

  • Have sufficient understanding of Python and Flask
  • Have Python 3 installed on your machine
  • Have MongoDB installed on your machine
  • Have a smartphone with an active phone number and Whatsapp installed
  • Have a Twilio account. You can create a free account if you are new to Twilio.

Setting up the Twilio Whatsapp Sandbox

First of all, you need to create a Twilio account if you don’t have one or sign in if you have one. You will need to activate Twilio Whatsapp sandbox since we are going to be working with Whatsapp. The sandbox allows you to immediately test with Whatsapp using a phone number without waiting for you Twilio number to be approved by Whatsapp. You can request production access for your Twilio phone number here.
To connect to the sandbox, log into your Twilio Console and select Programmable SMS on the side menu. After that, click on Whatsapp. This will open the Whatsapp sandbox page which contains your sandbox phone number and a join code. Send the join code starting with the word “join” to your sandbox phone number to enable the Whatsapp sandbox for your phone. You will then receive a message confirming the activation of the phone number to use the sandbox.

Application Setup

We are going to be using the Flask framework to build the application and MongoDB as our preferred database.

Creating the Application Directory and Virtual Environment
Run the following commands in your terminal to create the project folder and virtual environment for this project.

  • Create project directory named whatsapp_appointments mkdir whatsapp_appointments
  • Enter into the project directory cd whatsapp_appointments
  • Create a virtual environment for the project python3 -m venv venv
  • Activate the virtual environment

For macOS and Linux users:

source venv/bin/activate
Enter fullscreen mode Exit fullscreen mode

For windows users:

Enter fullscreen mode Exit fullscreen mode

The virtual environment helps create an isolated environment separate from the global python environment to run the project.

Structuring the Project
In the application directory, run the commands below to set up the folder structure:

mkdir controllers jobs services utils 
Enter fullscreen mode Exit fullscreen mode

The above commands create 4 folders and a file called in the application directory.
Your application directory should look like this:

Installing Project Dependencies
Finally, let’s install all the dependencies used in this project.

  • Flask: This library will be used for running our web server.
  • Pymongo: This library will be used to interface with the MongoDB database on your computer.
  • Twilio Python Helper Library: This library will be used to send Whatsapp messages to users.

Run the command below to install these dependencies:

pip install Flask pymongo twilio
Enter fullscreen mode Exit fullscreen mode

Building appointment reminders

In this section, we are going to write the logic for appointment reminders. We are going to use the mutiprocessing module in python to build the reminders logic. The main reason we are using processes over threads in this section is because in python, there is no safe way to terminate threads and we will need to edit and cancel appointments 🤷🏿‍♂️.
First of all, create 2 files called and in the jobs/ directory. Also, create a file called in the utils/ directory.
Copy the code below into the file:

import os
from import Client

AUTH_TOKEN = os.getenv('AUTH_TOKEN')
client = Client(ACCOUNT_SID, AUTH_TOKEN)

def format_message(appointment):
    message = "You have an appointment\n"
    message += f"Title: {appointment['title']}\n"
    message += f"Description: {appointment['description']}\n"
    message += f"Time: {appointment['time']}"
    return message

def send_message(appointment):
    message = client.messages.create(
    return message.sid
Enter fullscreen mode Exit fullscreen mode

In the above code, we have the logic for sending whatsapp messages. Note that the ACCOUNT_SID, AUTH_TOKEN, TWILIO_NUMBER are set as environment variables and we use os.getenv to access them. To send whatsapp messages, you must prefix the phone numbers with whatsapp: else it will be sent as sms. The format_message function accepts an appointment object and represents it in the following format:

You have an appointment
Title: <Appointment Title>
Description: <Appointment Description>
Time: <Appointment Time>
Enter fullscreen mode Exit fullscreen mode

Next, we are going to write the reminder’s logic. Open the file jobs/ and copy the following code inside:

import time
from datetime import datetime

from utils.whatsapp import send_message

def start_appointment_job(appointment):
    print(f"=====> Scheduled Appointment {appointment['_id']} for :> {appointment['time']}")
    if > appointment['time']:
    diff = appointment['time'] -
Enter fullscreen mode Exit fullscreen mode

The start_appointment_job function first checks if the appointment is in the past and if so, it returns. We use the time.sleep function to tell the application to pause execution for the number of seconds it takes for the appointment’s time to be reached. Once it finishes sleeping, it calls the send_message function written earlier to notify the user of the appointment via whatsapp.
Next, we are going to write the service functions for creating, getting, updating and deleting appointments. Create a file called in the services/ folder and copy the following inside:

from bson.objectid import ObjectId
from pymongo import MongoClient

client = MongoClient('mongodb://localhost/')

appointments = client['twilio']['appointments']

def create_appointment(data):

def get_appointment(appointment_id):
    result = appointments.find_one({'_id': ObjectId(appointment_id)})
    return dict(result) if result else None

def get_appointments(conditions):
    if '_id' in conditions:
        conditions['_id'] = ObjectId(conditions['_id'])
    results = appointments.find(conditions)
    data = []
    for result in results:
    return data

def update_appointment(appointment_id, data):
    appointment = appointments.find_one_and_update({'_id': ObjectId(appointment_id)}, {'$set': data},
    return appointment

def delete_appointment(appointment_id):
    appointments.find_one_and_delete({'_id': ObjectId(appointment_id)})
Enter fullscreen mode Exit fullscreen mode

In the above code, we write functions to perform CRUD operations on the appointments collections in mongodb. We use the pymongo library to connect to our local mongodb instance. Note that the _id field cannot be sent as a string else no data will be returned, we have to cast it to an ObjectId instance which mongodb uses.

Finally, we are going to write the logic for creating processes for an appointment and modifying them based on the change in appointments. Open jobs/ and copy the following code inside:

from multiprocessing import Process
from datetime import datetime

from .appointments import start_appointment_job

from services.appointments import get_appointments


def terminate_worker(worker):
    except Exception as err:
        print('====> Error occurred terminating process', err)

def schedule_appointment(appointment):
    appointment_id = str(appointment['_id'])
    worker = Process(target=start_appointment_job, args=(appointment,))
    WORKERS[appointment_id] = worker

def update_scheduled_appointment(appointment_id, updated_appt):
    worker = WORKERS[appointment_id]
    new_worker = Process(target=start_appointment_job, args=(updated_appt,))
    WORKERS[appointment_id] = new_worker

def delete_scheduled_appointment(appointment_id):
    worker = WORKERS[appointment_id]
    del WORKERS[appointment_id]

def init_workers():
    print('=====> Initializing workers')
    appts = get_appointments({})
    for appt in appts:
        if > appt['time']:

def close_workers():
    for appointment_id, worker in WORKERS.items():
Enter fullscreen mode Exit fullscreen mode

In the above code, when we want to schedule an appointment, we create a Process targeting the start_appointment_job function with the appointment object as argument. Note that we have a dictionary objection named WORKERS where we store the process object with the appointment_id as key, this is what we use to get the Process object in case of an update.
To update an appointment, we get the process object via the appointment_id and terminate it. We then create a new Process object with the updated details and update the reference in the WORKERS dictionary.
To delete an appointment, we get the Process object and terminate it. We then delete the key from the WORKERS dictionary so that it doesn’t have a reference there again.
Note that, to terminate a Process object, we first call the terminate function on it when sends a SIGTERM signal to the process. The join function waits for the Process object to terminate and the close function tells the process to release the resources that it holds.
Lets say our application restarts, we want to be able to recreate our Process objects and that’s where the init_workers function comes into play. It gets all the appointments and calls the schedule_appointment function on appointments that are still pending.

Building the API

In this section, we are going to be building the api routes which we are going use to create, update, get and delete appointments. First of all, we have to create utility functions for handling both our requests and our api responses. Create 2 files called and in the utils/ folder. Copy the code below into the file:

from datetime import datetime

def validate_body(data, required_fields):
    for field in required_fields:
        if field not in data:
            return False, field
    return True, None

def parse_appointment(body):
        if body.get('time'):
            time_obj = datetime.strptime(body['time'], '%Y-%m-%d %H:%M')
            body['time'] = time_obj
        return True, None
    except Exception as err:
        return False, str(err)
Enter fullscreen mode Exit fullscreen mode

From the above code, we can see that the validate_body function takes in a data dictionary and a list of required fields. It checks the required fields against the data dictionary and returns a status of False if a field is absent. The parse_appointment function is used to convert the time field in an appointment object into a datetime object by the use of datetime.strptime function which accepts the time string and the time format. The function returns false if it’s not able to convert the string into a datetime object.
Next, copy the code below into the file:

import json
import time

from flask import jsonify

def stringify_objectid(data):
    str_data = json.dumps(data, default=str)
    return json.loads(str_data)

def response(status, message, data, status_code=200):
    :param status : Boolean Status of the request
    :param status_code: Status Code of response
    :param message : String message to be sent out as description of the message
    :param data : dictionary representing extra data
    if data:
        data = stringify_objectid(data)
    res = {'status': status, 'message': message, 'data': data, 'timestamp': timestamp()}
    return jsonify(res), status_code

def error_response(message, status='error', code='R0', status_code=400):
    res = {'message': message, 'status': status, 'code': code}
    return jsonify(res), status_code

def timestamp():
    Helper Function to generate the current time
    return time.time()
Enter fullscreen mode Exit fullscreen mode

In the above code, we have helper functions to return api responses and error messages. The stringify_objectid function is used to convert the mongodb ObjectId object into a string.

Next, we are going to write our controller functions. Create a file called in the controllers/ folder and copy the following inside:

from flask import Blueprint, request

from services import appointments as apt_service
from utils.request import validate_body, parse_appointment
from utils.response import response, error_response
from utils.whatsapp import send_sms

import jobs

appointments = Blueprint('appointments', __name__)

@appointments.route('/', methods=['POST'])
def create():
    body = request.get_json()
    status, missing_field = validate_body(body, ['title', 'phone', 'description', 'time'])
    if not status:
        return error_response(f'{missing_field} is required')
    status, error = parse_appointment(body)
    if not status:
        return error_response(error)
        return response(True, 'Appointment created successfully', body)
    except Exception as err:
        print('=====> Error', err)
        return error_response(str(err))

def view():
    conditions = dict(request.args)
        data = apt_service.get_appointments(conditions)
        return response(True, 'Appointments', data)
    except Exception as err:
        print('=====> Error', err)
        return error_response(str(err))

def view_one(appointment_id):
        data = apt_service.get_appointment(appointment_id)
        return response(True, 'Appointment', data)
    except Exception as err:
        print('=====> Error', err)
        return error_response(str(err))

@appointments.route('/<appointment_id>', methods=['PUT'])
def update(appointment_id):
    body = request.get_json()
        appointment = apt_service.update_appointment(appointment_id, body)
        jobs.update_scheduled_appointment(appointment_id, appointment)
        return response(True, 'Updated Appointment', appointment)
    except Exception as err:
        print('=====> Error', err)
        return error_response(str(err))

@appointments.route('/<appointment_id>', methods=['DELETE'])
def delete(appointment_id):
        return response(True, 'Appointment deleted successfully', None)
    except Exception as err:
        print('====> Error', err)
        return error_response(str(err))
Enter fullscreen mode Exit fullscreen mode

In the above code, we created a new flask blueprint and functions to create appointments, get appointments, update appointments and delete appointments. As you can see from the above code, we call the schedule appointment function to start the scheduling process when an appointment is created. The update function calls the update_scheduled_appointment function to reshedule the appointment and the delete function calls the delete_scheduled_appointment function to delete the scheduling process.

Finally, open the file and copy the following inside:

import atexit

from flask import Flask

from controllers.appointments import appointments
from jobs import close_workers, init_workers

def create_app():
    return Flask(__name__)

app = create_app()

app.register_blueprint(appointments, url_prefix='/api/appointments')

if __name__ == '__main__':
Enter fullscreen mode Exit fullscreen mode

In the above code, the use the register_blueprint function to make sure that urls starting with /api/appointments are routed through the appointments blueprint and our controller functions are executed for matching urls. We use the atexit module to perform a task when the application is closing. In the create_app function, we are telling the system to first schedule pending appointments and then close all running processes when the application is closing.

Testing the Application

To run the application, open your terminal and execute python in your application’s directory. You should see something like the image below:

Now, you can send a POST request to to create an appointment and you will get a whatsapp message when the appointment time has been reached. Below is a sample request body:

        "title": "Data Structures & Algorithms",
        "phone": "+2349094739283",
        "description": "Watch Youtube videos on Data Structures and Algorithms",
        "time":"2020-02-23 18:10"
Enter fullscreen mode Exit fullscreen mode

You should receive a whatsapp message when the appointment time is reached as shown in the image below


We have now come to the end of the tutorial. We have successfully built an application that sends appointment notification over Whatsapp using Twilio, Python and Flask.
You will find the source code for this tutorial here.

Top comments (0)

Timeless DEV post...

Git Concepts I Wish I Knew Years Ago

The most used technology by developers is not Javascript.

It's not Python or HTML.

It hardly even gets mentioned in interviews or listed as a pre-requisite for jobs.

I'm talking about Git and version control of course.