In this article, you will learn how to build an sms reminder backend system using Vonage SMS API.
FastAPI is a modern, fast (high-performance), a web framework for building APIs with Python 3.6+ based on standard Python type hints. It is built on top of Starlette for the web parts and Pydantic for the data parts. Its advantages are the ability to create endpoints quickly, having built-in support for WebSockets and being asynchronous.
I would use Python for the backend system because of its simplicity and scalability as well as its vast ecosystem and libraries that can be used for a wide range of tasks.
Disclaimer: I would recommend you go through the FastAPI official documentation or a video tutorial to get started with the framework; as doing this would help speed things up.
Vonage's SMS API enables you to send and receive text messages to and from users worldwide, using our REST APIs. Get started here.
Vonage Account Setup
To follow along with this tutorial, I would strongly advise that you set up a Vonage account. You can start by signing up here and start building with free credit. Once you have, kindly grab your API and Secret key at the top of the Vonage api dashboard.
Project Setup and Installation
1). Create and Activate The Virtual Environment
To begin, start by creating a directory, a virtual environment (preferably pipenv) and activate it:
mkdir sms_reminder && cd sms_reminder
python3.9 -m pipenv shell
2). Install Required Dependencies
Create a requirements.txt file in the directory and continue by copying and pasting the code snippet below in your requirements.txt file:
aiosqlite==0.18.0
alembic==1.9.2
anyio==3.6.2
apscheduler==3.9.1.post1
backoff==2.2.1
certifi==2022.12.7
click==8.1.3
databases==0.7.0
fastapi==0.89.1
greenlet==2.0.1; python_version >= '3' and (platform_machine == 'aarch64' or (platform_machine == 'ppc64le' or (platform_machine == 'x86_64' or (platform_machine == 'amd64' or (platform_machine == 'AMD64' or (platform_machine == 'win32' or platform_machine == 'WIN32'))))))
h11==0.14.0
httpcore==0.16.3
httptools==0.5.0
httpx==0.23.3
idna==3.4
mako==1.2.4
markupsafe==2.1.2
pydantic==1.10.4
python-decouple==3.7
python-dotenv==0.21.1
pytz-deprecation-shim==0.1.0.post0
pytz==2022.7.1
pyyaml==6.0
rfc3986==1.5.0
six==1.16.0
sniffio==1.3.0
sqlalchemy==1.4.46
starlette==0.22.0
typing-extensions==4.4.0; python_version < '3.10'
tzdata==2022.7; python_version >= '3.6'
tzlocal==4.2
uvicorn==0.20.0
uvloop==0.17.0
watchfiles==0.18.1
websockets==10.4
Proceed to install the required packages at once using the command below:
pipenv install -r requirements.txt
3). Create a main Module
Now that you have installed the required dependencies, let's proceed by creating a main.py file and pasting the below code into it.
This is the entry point of the backend system that will be responsible for constructing the FastAPI application, registering API routers, events to connect and disconnect the database, starting job schedulers and running the application.
# Uvicorn Imports
import uvicorn
# FastAPI Imports
from fastapi import FastAPI
# construct application
app = FastAPI(
title="SMS Reminder System",
description="An SMS reminder system that allows users to text a specific number to set reminders for themselves.",
verison=1.0,
)
@app.on_event("startup")
async def startup():
pass # connect to database will come here
@app.on_event("shutdown")
async def disconnect():
pass # disconnect from the database will come here
if __name__ == "__main__":
uvicorn.run(
"sms_reminder.main:app", host="0.0.0.0", port=3030, reload=True
)
4). Create models Directory
We created main.py, the entry point of our application. Next would be creating the models' directory, and basically, what will be there is going to be our database tables defined as class(es).
Let's begin, create an __init__.py file to mark this directory as a Python module.
“Clean code is not written by following a set of rules. You don't become a software craftsman by learning a list of heuristics. Professionalism and craftsmanship come from values that drive disciplines.”
Create a base.py file containing a class named ObjectTracker, which serves as a base class for all tracked objects in the database.
# Stdlib Imports
from datetime import datetime
# SQLAlchemy Imports
from sqlalchemy import Column, Integer, DateTime
class ObjectTracker(object):
id = Column(Integer, primary_key=True, index=True)
date_created = Column(DateTime, default=datetime.now)
date_updated = Column(DateTime, onupdate=datetime.now)
To create the base class for object tracking in the database, we import the Column, Integer, and DateTime classes from the sqlalchemy library, and also import datetime from the Python standard library to record object creation and update times.
Next would be creating a sms.py file, copy and paste the below code into it:
# SQLAlchemy Imports
from sqlalchemy import Column, String, DateTime
# Own Imports
from sms_reminder.models.base import ObjectTracker
class Reminder(ObjectTracker):
phone_number = Column(String)
message = Column(String)
remind_when = Column(DateTime)
The Reminder class represents the reminders table in our database, but we must configure our database to translate this object into a table.
5). Create a config Directory
To ensure Python treats this directory as a module, we create an __init__.py file. Next, we create a database.py file, which contains information for configuring the database, constructing a session maker, a database connector (to connect and shutdown our database), and creating a declarative base class that converts our models into database tables.
Let's proceed; continue by copying the below code and pasting it into the database.py file:
# SQLAlchemy Imports
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from sqlalchemy.ext.declarative import declarative_base
# Third Imports
from databases import Database
# create a database and engine
DATABASE_URL = "sqlite:///./sms_reminder.sqlite"
DATABASE_ENGINE = create_engine(
DATABASE_URL, connect_args={"check_same_thread": False}
)
# construct a session maker
session_factory = sessionmaker(
autocommit=False, autoflush=False, bind=DATABASE_ENGINE
)
# construct a scoped session
SessionLocal = scoped_session(session_factory)
# Construct a base class for declarative class definitions
Base = declarative_base()
# Construct a db connector to connect, and shutdown the database
db_connect = Database(DATABASE_URL)
You might be thinking: "Things might seem pretty complex right now”. But don't worry, I'll explain the magic you're witnessing.
create_engine: this method takes in the connection (database) URL and returns a sqlalchemy engine that references both a Dialect and a Pool, which together interpret the DBAPI's module functions as well as the behaviour of the database (Source).session_maker: to explain this, we need to start by understanding what aSessionis. A Session in sqlalchemy establishes all conversations with the database, it is a regular python class which can directly be instantiated. Thesession_makeris used to create a top-levelSessionconfiguration which can then be used throughout an application without the need to repeat certain configurational arguments. Read more here and here.scoped_session: according to sqlalchemy official documentation, thescoped_sessionobject produces a thread-managed registry of Session objects. It is commonly used in web applications so that a single global variable can be used to safely represent transactional sessions with sets of objects, localized to a single thread. In short, what there are trying to say is, to usescoped_sessionin your application for thread safety. Because it opens and closes a session object for each request, this is a safe and recommended way of doing things. Read more here.declarative_base: this function is used to define classes mapped to relational database tables.Database: this class is constructing a database connector that will help us connect and shut down our database.
Now that you understand what's happening, let's add Base to the Reminder class in the sms models:
# SQLAlchemy Imports
from sqlalchemy import Column, String, DateTime
# Own Imports
from sms_reminder.config.database import Base # new line
from sms_reminder.models.base import ObjectTracker
class Reminder(ObjectTracker, Base): # added Base
__tablename__ = "reminders" # new line
phone_number = Column(String)
message = Column(String)
remind_when = Column(DateTime)
By doing the above, sqlalchemy will map the defined class to a relational database table. Next, we update the main.py entry point to connect and shutdown our database whenever our backend server emits a startup and shutdown event.
# Uvicorn Imports
import uvicorn
# FastAPI Imports
from fastapi import FastAPI
# Own Imports
from sms_reminder.config.database import db_connect # new line
# construct application
app = FastAPI(
title="SMS Reminder System",
description="An SMS reminder system that allows users to text a specific number to set reminders for themselves.",
verison=1.0,
)
@app.on_event("startup")
async def startup():
scheduler.start()
await db_connect.connect() # new line
@app.on_event("shutdown")
async def disconnect():
await db_connect.disconnect() # new line
if __name__ == "__main__":
uvicorn.run(
"sms_reminder.main:app", host="0.0.0.0", port=3030, reload=True
)
Proceed by creating a settings.py file to hold and store our environmental variables. Copy and paste the following code into the file:
# Stdlib Imports
from functools import lru_cache
from pydantic import BaseSettings
# Third Party Imports
from decouple import config as environ
class Settings(BaseSettings):
"""
Settings to hold environmental variables.
"""
VOYAGE_API_KEY: str = environ("VOYAGE_API_KEY", cast=str)
VOYAGE_SECRET_KEY: str = environ("VOYAGE_SECRET_KEY", cast=str)
@lru_cache
def get_setting_values() -> Settings:
env_var = Settings()
return env_var
Let's go over what we have done:
lru_cache: this decorator will create theSettingsobject once. After that, it will be cached and reused, instead of having to create theSettingsobject every time we want to access the values.BaseSettings: allows values to be overridden by environment variables. This is useful in production for secrets you do not wish to save in code, it plays nicely with docker(-compose), Heroku and any 12-factor app design.
The Settings object is responsible for getting our environment variables, we need to create a .env file that will be loaded by environ. Copy and paste the below code into the file:
VOYAGE_API_KEY=value
VOYAGE_SECRET_KEY=value
Replace the value with the correct keys from your Vonage dashboard.
6). Initialize alembic to handle database migrations
Alembic is a lightweight database migration tool for usage with the SQLAlchemy Database Toolkit for Python. It is widely used for database migrations. Let's proceed to use it.
We installed alembic at the start of our project, so now we need to initialize it for our working project directory. Run the following command in your terminal:
alembic init migrations
I used migrations to store all database migrations in that folder. Running the above command will create a folder named migrations, which contains the following files/folders:
- env.py
- README
- script.py.mako
- versions (directory)
In the project directory, a file alembic.ini will also be created. There will be no version files in your versions directory because we haven’t made any migrations yet. Now to use alembic, we need to do certain changes.
First, change the sqlalchemy.url in your alembic.ini file:
sqlalchemy.url = sqlite:///./sms_reminder.sqlite
Next, we need to provide our database model to alembic and access the metadata from the model.
Edit your env.py file inside your migrations folder:
... # this means there are other imports here
# Own Imports
from sms_reminder.models.sms import Reminder # new line
... # this means there are other codes above
target_metadata = Reminder.metadata # update
As shown above, we have to provide our database model to alembic. Run the below command in your terminal:
alembic revision --autogenerate -m "Create reminder table"
Running the above command will tell alembic to generate our migration file and store it in the versions folder. Once this file is generated, run the below command:
alembic upgrade head
Your tables will be generated in your database. That's all to the magic. Read more about alembic here.
7). Create schemas Directory
We successfully created our models classes that are represented as a database table(s), and made database migrations. Next would be creating our pydantic models. Pydantic is basically used for data parsing and validation using type annotation.
An interesting thing to note is, pydantic forces type hints at runtime, and provides user-friendly errors when data is invalid. To begin, create an __init__.py file to tell Python to treat the directory as a module, create another file named crud.py and paste the below codes inside it:
# Stdlib Imports
import pytz
from datetime import datetime
from pydantic import BaseModel, Field
class BaseReminderSchema(BaseModel):
phone_number: str = Field(
description="What's your phone number? Ensure you include your country code and is valid. E.g 234xxxxxxxxxx"
)
message: str = Field(
description="What message do you want to remind yourself with? E.g Time to go to the gym!"
)
remind_when: datetime = Field(
description="When should I send this message to you?",
default=datetime.now(tz=pytz.timezone("Africa/Lagos")),
)
class CreateReminderSchema(BaseReminderSchema):
pass
class ReminderSchema(BaseReminderSchema):
id: int
date_created: datetime
class Config:
orm_mode = True
Let's go over what's happening in the above code. We are importing the python timezone (pytz) library, datetime from the datetime library and importing BaseModel and Field from pydantic.
pytz: this library allows accurate and cross-platform timezone calculations using Python 2.4 or higher. It also solves the issue of ambiguous times at the end of daylight saving time. Read more here.BaseModel: this is a class used in defining objects in pydantic. You can think of models as similar to types in strictly typed languages, or as the requirements of a single endpoint in an API. Read more here.Field: this is a class used to provide extra information about a field, either for the model schema or complex validation.BaseReminderSchema: this is a class that inherits fromBaseModelto create fields that will be used across others’ schemas.CreateReminderSchema: this is a class that inherits fromBaseReminderSchemathat will be used as a requirement to create a reminder.ReminderSchema: this is a class that inherits fromBaseReminderSchemathat will be used as a requirement to list reminders created. Theorm_mode(aka arbitrary class instances) support models that map to ORM objects.
8). Create interface Directory
We want an interface that sits between our data access layer (database) and the business logic layer (services) of an application. The best way to go about doing that is by using the repository design pattern. You see, the repository pattern abstracts the manner in which your data is queried or created for you.
Moving on, create a __init__.py file to tell Python to treat the directory as a module, and then- create a file named sms.py. Copy and paste the below code into the file:
# Stdlib Imports
from typing import List
from datetime import datetime
# SQLAlchemy Imports
from sqlalchemy.orm import Session
# Own Imports
from sms_reminder.models.sms import Reminder
from sms_reminder.config.database import SessionLocal
class ReminderORMInterface:
"""Reminder ORM that interface with the database."""
def __init__(self) -> None:
self.db: Session = self.get_db_session().__next__()
def get_db_session(self):
"""
This method creates a database session,
yields it to the caller, rollback the session
if an exception occurs; otherwise, close the session.
"""
session = SessionLocal()
try:
yield session
except Exception:
session.rollback()
finally:
session.close()
async def get(self) -> List[Reminder]:
"""This method gets a list of reminders."""
reminders = self.db.query(Reminder).all()
return reminders
async def create(
self, phone_number: str, message: str, remind_when: datetime
) -> Reminder:
"""This method creates a new reminder to the database."""
reminder = Reminder(
phone_number=phone_number,
message=message,
remind_when=remind_when,
)
self.db.add(reminder)
self.db.commit()
self.db.refresh(reminder)
return reminder
reminder_orm = ReminderORMInterface()
Let's break down what's happening in the above code, we have four methods:
get_db_session: this method creates a database session for us, yields it, and rollback the session if an exception occurs; otherwise, close the session.__init__: this method initializes the database session.get: this method queries the Reminder model (table), getting us a list of all the objects in it.create: this method creates a new reminder instance, persists it into a state (that will either rollback or move forward), flush pending changes and commit the current transaction and refresh the newly created instance in the table.
To begin using our interface, we need to initialize our class to a variable.
9). Create services Directory
In our previous section, we went over what an interface (a repository pattern) is; and how it sits between our data access layer and the business logic layer of our application. In this section, we will be creating the business logic layer. Moving on, create an __init__.py file to tell Python to treat the directory as a module.
i). Create a file named vonage.py file, copy and paste the below code into it:
# Stdlib Imports
from typing import Dict
# FastAPI Imports
from fastapi import HTTPException
# Own Imports
from sms_reminder.config.settings import get_setting_values as settings
# Third Party Imports
import httpx
import backoff
class VonageSMS:
"""SMS API Service provider to handle sending text messages."""
def __init__(self) -> None:
"""
The method is used to initialize the class
and set the base_url, settings, secret_key and api_key.
"""
self.base_url = "https://rest.nexmo.com/sms/json"
self.settings = settings()
self.secret_key = self.settings.VOYAGE_SECRET_KEY
self.api_key = self.settings.VOYAGE_API_KEY
def headers(self) -> Dict[str, str]:
"""
This method returns a header with a key of "Content-Type" and a value of
"application/x-www-form-urlencoded".
"""
return {"Content-Type": "application/x-www-form-urlencoded"}
@backoff.on_exception(backoff.expo, httpx.ConnectTimeout, max_time=100)
async def send(self, phone_number: str, message: str) -> True:
"""
This method sends a message to the provided phone number using the VonageSMS API.
:param phone_number: The phone number to send the message to
:type phone_number: str
:param message: The message you want to send
:type message: str
:return: True
"""
async with httpx.AsyncClient() as client:
request_data = f"from=Send Reminder!&text={message}&to={phone_number}&api_key={self.api_key}&api_secret={self.secret_key}"
response = await client.post(
url=self.base_url, headers=self.headers(), data=request_data
)
response_data = response.json()["messages"][0]
if response_data["status"] == "0":
return True
raise HTTPException(
500,
{
"source": "VonageSMS",
"message": response_data["error-text"],
},
)
vonage_sms = VonageSMS()
"Woahhh. That's a lot of code, Abram!" Don't worry about it, let's go over it together. One import, method at a time. I won't be explaining generic stuff I assume you should already know, such as Dict from
typing.
HTTPException: this is a base class for all http exceptions. You can override it to create custom exceptions like ServerException, NotFoundException, etc.settings: this is a function from theconfig/settings.pymodule that when called caches our environment variables and also gives us access to the value.httpx: this is a fully featured http client library for python3. It has support for both http/1.1 and http/2 and provides both sync and async APIs.backoff: this library provides function decorators which can be used to wrap a function such that it will be retried until some condition is met.
Do you see it now? There was no need to panic, I am here to explain everything just right. Next is, what each method is doing:
-
__init__: this method is used to initialize ourVonageSMSclass and set the base_url and settings; secret_key and api_key of vonage sms service. -
headers: this method is used to set the content type we want in our headers, when we make a request to Vonage sms service via api. -
send: this method sends a message to the phone number provided using Vonage sms api. To be able to send an sms, we require a phone number and the message we want to tell send.Note the use of
httpx.AsyncClient()rather thanhttpx.Client, in the method. This is becauseAsyncClientis asynchronous and its context manager usesasync withnot justwith. -
Moving forward, we are encoding the form in an URL and making a post request to our
base_urlthat we have set in our__init__method. Next, we are checking t see if we have a status that equals to"0"returnTrue, otherwise, raise a 500 exception that tells the client what went wrong.
ii). Create a file named tasks.py, copy and paste the below code into it:
# Stdlib Imports
from random import randint
from datetime import datetime
# APScheduler Imports
from apscheduler.schedulers.asyncio import AsyncIOScheduler
# Own Imports
from sms_reminder.services.vonage import vonage_sms
# initialize ayncio scheduler
scheduler = AsyncIOScheduler()
async def create_reminder_job(
phone_number: str, message: str, remind_when: datetime
) -> dict:
"""
This function creates a job that will send a message to the provided
phone number at a specific time.
:param phone_number: The phone number to send the message to
:type phone_number: str
:param message: The message to be sent
:type message: str
:param remind_when: The datetime object that represents the time when the reminder should be sent
:type remind_when: datetime
:return: A dictionary with the keys "scheduled" and "job_id".
"""
job_uid = randint(0, 9999)
reminder_job = scheduler.add_job(
func=vonage_sms.send,
trigger="date",
args=(phone_number, message),
name=f"reminder_set_{phone_number}_{job_uid}",
id=f"{phone_number}_{job_uid}",
next_run_time=remind_when,
)
return {"scheduled": True, "job_id": reminder_job.id}
We are importing AsyncIOScheduler from the apscheduler library and importing vonage_sms from our services/vonage module. The create_reminder_job function is responsible for creating a job that will send a message to the provided phone number at a specific number.
iii). Create a file named sms.py, copy and paste the below code into it:
# Stdlib Imports
from datetime import datetime
from typing import List
# FastAPI Imports
from fastapi import HTTPException
# Own Imports
from sms_reminder.services.tasks import create_reminder_job
from sms_reminder.interface.sms import reminder_orm, Reminder
async def create_user_reminder(
phone_number: str, message: str, remind_when: datetime
) -> Reminder:
"""
This function creates a new reminder in the database.
:param phone_number: The phone number of the user who will receive the reminder
:type phone_number: str
:param message: The message that will be sent to the user
:type message: str
:param reminder_when: datetime
:type reminder_when: datetime
:return: A reminder object
"""
reminder = await reminder_orm.create(phone_number, message, remind_when)
reminder_job = await create_reminder_job(
phone_number, message, remind_when
)
if reminder_job["scheduled"]:
return reminder
raise HTTPException(400, {"message": "Was not able to set reminder!"})
async def get_user_reminders() -> List[Reminder]:
"""
This function gets the list of user reminders.
:return: A list of reminder objects.
"""
reminders = await reminder_orm.get()
return reminders
We are importing create_reminder_job from services/tasks module, reminder_orm and Reminder model class from interface/sms module. The function create_user_reminder is responsible for creating a new reminder in the database. When we create a new reminder, we are also creating a job that would send the reminder message to the phone number provided. A 400 exception is raised if we were not able to set a job.
Next is the get_user_reminders function, we are basically getting a list of reminders in the database.
10). Create an api Directory
In the previous section, we created our business logic layer. Next would be creating our API endpoints, to begin, create an __init__.py file to tell Python to treat the directory as a module.
i). Create a file named index.py, and copy and paste the below code into it:
# FastAPI Imports
from fastapi import APIRouter
# initialize api router
router = APIRouter(tags=["Root"])
@router.get("/")
async def root() -> dict:
"""
SMS Reminder System root
Returns: dict: version and description
"""
return {
"version": 1.0,
"description": "An SMS reminder system that allows users to text a specific number to set reminders for themselves.",
}
APIRouter is a class imported from fastapi, we use it to declare our path operations. The root is an api view that will return the version and description of our backend application.
ii). Create a file named crud.py, and copy and paste the below code into it:
# Stdlib Imports
from typing import List
from datetime import datetime, timezone
# FastAPI Imports
from fastapi import APIRouter, status, HTTPException
# Own Imports
from sms_reminder.services.sms import create_user_reminder, get_user_reminders
from sms_reminder.schemas.crud import CreateReminderSchema, ReminderSchema
# initialize the api router
router = APIRouter(tags=["Reminder"], prefix="/reminders")
@router.post("/", status_code=status.HTTP_201_CREATED)
async def create_reminder(payload: CreateReminderSchema):
"""
This API view creates and sets a reminder.
:param payload: CreateReminderSchema\n
:type payload: CreateReminderSchema
:return: The reminder is serialized to a JSON.
"""
if payload.remind_when <= datetime.now(timezone.utc):
raise HTTPException(
400,
{
"message": "Kindly set a date and time that exceeds the past and now."
},
)
reminder = await create_user_reminder(
payload.phone_number, payload.message, payload.remind_when
)
return {"message": "Reminder set!", "data": reminder}
@router.get("/", response_model=List[ReminderSchema])
async def get_reminders():
"""
This API view gets the total available reminders.
:return: A list of reminders.
"""
reminders = await get_user_reminders()
return reminders
We have gone over what APIRouter and HTTPException is; moving on, we declared the HTTP status code we would want to receive in our response.
We are also importing our business logic, create_user_remeinder and get_user_reminders functions from the services/sms module, and finally; importing CreateReminderSchema and ReminderSchema from schemas/crud module.
The api view create_reminder is responsible for creating and setting a reminder. To declare the HTTP method that we want to use, we use the router as a decorator to access the post method.
The async keyword in the function’s definition tells FastAPI that it is to be run asynchronously i.e. without blocking the current thread of execution. We use CreateReminderSchema as a type annotation for the payload parameter.
The if statement is a validation to ensure that we can not create a reminder with a past date and time. Moving forward, the create_user_reminder is called to handle the creation and setting of the reminder. If no exception is raised, the response of the api would be a dictionary showing a message and the reminder.
12). Start Server Start the uvicorn server with the below command in your terminal:
python sms_reminder/main.py
NOTE: If you get the following error:
Traceback (most recent call last):
File "/.../sms-based-reminder-system/sms_reminder/main.py", line 8, in from sms_reminder.services.tasks import scheduler
ModuleNotFoundError: No module named 'sms_reminder'
Run the below code in your terminal and start your uvicorn server again:
export PYTHONPATH=$PWD
You would get something close to the below message in your terminal console:
INFO: Will watch for changes in these directories: [‘/…/sms-based-reminder-system’]
INFO: Uvicorn running on [http://0.0.0.0:3030](http://0.0.0.0:3030/) (Press CTRL+C to quit)
INFO: Started reloader process [1203267] using WatchFiles
INFO: Started server process [1203280]
INFO: Waiting for application startup.
INFO: Application startup complete.
Conclusion
Congratulations! You have learned how to build an SMS reminder backend system using FastAPI and Vonage SMS API. There are many other things you can do, such as throttling the create-reminder API to handle 10 requests per minute to reduce spam bots. You can find the source code for this tutorial project on my Github.