Build a Lightning-fast SMS Reminder System With Vonage SMS API and FastAPI Backend!

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 or a video to get started with the framework; as doing this would help speed things up. documentation tutorial 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 and start building with free credit. Once you have, kindly grab your API and Secret key at the top of the Vonage api . here 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 file in the directory and continue by copying and pasting the code snippet below in your file: requirements.txt requirements.txt 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.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 Module main Now that you have installed the required dependencies, let's proceed by creating a file and pasting the below code into it. main.py 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 Directory models We created , 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). main.py Let's begin, create an file to mark this directory as a Python module. __init__.py “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 , , and classes from the sqlalchemy library, and also import from the Python standard library to record object creation and update times. Column Integer DateTime datetime Next would be creating a file, copy and paste the below code into it: sms.py # 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 Directory config To ensure Python treats this directory as a module, we create an file. Next, we create a 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. __init__.py database.py Let's proceed; continue by copying the below code and pasting it into the file: database.py # 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. : 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 ( ). create_engine Source : to explain this, we need to start by understanding what a is. A Session in sqlalchemy establishes all conversations with the database, it is a regular python class which can directly be instantiated. The is used to create a top-level configuration which can then be used throughout an application without the need to repeat certain configurational arguments. Read more and . session_maker Session session_maker Session here here : according to sqlalchemy official documentation, the object 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 use in 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 . scoped_session scoped_session scoped_session here : this function is used to define classes mapped to relational database tables. declarative_base : this class is constructing a database connector that will help us connect and shut down our database. Database Now that you understand what's happening, let's add to the class in the sms models: Base Reminder # 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 entry point to connect and shutdown our database whenever our backend server emits a startup and shutdown event. main.py # 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 file to hold and store our environmental variables. Copy and paste the following code into the file: settings.py # 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: : this decorator will create the object once. After that, it will be cached and reused, instead of having to create the object every time we want to access the values. lru_cache Settings Settings : 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. BaseSettings The Settings object is responsible for getting our environment variables, we need to create a file that will be loaded by . Copy and paste the below code into the file: .env environ VOYAGE_API_KEY=value VOYAGE_SECRET_KEY=value Replace the with the correct keys from your Vonage . value dashboard 6). Initialize 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. Alembic 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 will also be created. There will be no version files in your directory because we haven’t made any migrations yet. Now to use alembic, we need to do certain changes. alembic.ini versions First, change the in your file: sqlalchemy.url alembic.ini 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 file inside your folder: env.py migrations ... # 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 folder. Once this file is generated, run the below command: versions alembic upgrade head Your tables will be generated in your database. That's all to the magic. Read more about alembic . here 7). Create Directory schemas We successfully created our 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. models An interesting thing to note is, To begin, create an file to tell Python to treat the directory as a module, create another file named and paste the below codes inside it: pydantic forces type hints at runtime, and provides user-friendly errors when data is invalid. __init__.py crud.py # 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 ( ) library, datetime from the library and importing BaseModel and Field from . pytz datetime pydantic : 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 . pytz here : 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 . BaseModel here : this is a class used to provide extra information about a field, either for the model schema or complex validation. Field : this is a class that inherits from to create fields that will be used across others’ schemas. BaseReminderSchema BaseModel : this is a class that inherits from that will be used as a requirement to create a reminder. CreateReminderSchema BaseReminderSchema : this is a class that inherits from that will be used as a requirement to list reminders created. The (aka arbitrary class instances) support models that map to ORM objects. ReminderSchema BaseReminderSchema orm_mode 8). Create Directory interface 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 file to tell Python to treat the directory as a module, and then- create a file named . Copy and paste the below code into the file: __init__.py sms.py # 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: : this method creates a database session for us, yields it, and rollback the session if an exception occurs; otherwise, close the session. get_db_session : this method initializes the database session. __init__ : this method queries the Reminder model (table), getting us a list of all the objects in it. get : 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. create To begin using our interface, we need to initialize our class to a variable. 9). Create Directory services 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 file to tell Python to treat the directory as a module. __init__.py i). Create a file named file, copy and paste the below code into it: vonage.py # 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 : this is a base class for all http exceptions. You can override it to create custom exceptions like ServerException, NotFoundException, etc. HTTPException : this is a function from the module that when called caches our environment variables and also gives us access to the value. settings config/settings.py : 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. httpx : this library provides function decorators which can be used to wrap a function such that it will be retried until some condition is met. backoff 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: : this method is used to initialize our class and set the base_url and settings; secret_key and api_key of vonage sms service. __init__ VonageSMS : 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. headers : 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. send Note the use of rather than , in the method. This is because is asynchronous and its context manager uses not just . httpx.AsyncClient() httpx.Client AsyncClient async with with Moving forward, we are encoding the form in an URL and making a post request to our that we have set in our method. Next, we are checking t see if we have a status that equals to return , otherwise, raise a 500 exception that tells the client what went wrong. base_url __init__ "0" True ii). Create a file named , copy and paste the below code into it: tasks.py # 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 from the library and importing from our module. The function is responsible for creating a job that will send a message to the provided phone number at a specific number. AsyncIOScheduler apscheduler vonage_sms services/vonage create_reminder_job iii). Create a file named , copy and paste the below code into it: sms.py # 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 from module, and model class from module. The function 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. create_reminder_job services/tasks reminder_orm Reminder interface/sms create_user_reminder Next is the function, we are basically getting a list of reminders in the database. get_user_reminders 10). Create an Directory api In the previous section, we created our business logic layer. Next would be creating our API endpoints, to begin, create an file to tell Python to treat the directory as a module. __init__.py i). Create a file named , and copy and paste the below code into it: index.py # 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.", } is a class imported from , we use it to declare our path operations. The is an api view that will return the version and description of our backend application. APIRouter fastapi root ii). Create a file named , and copy and paste the below code into it: crud.py # 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 and is; moving on, we declared the HTTP code we would want to receive in our response. APIRouter HTTPException status We are also importing our business logic, and functions from the module, and finally; importing and from module. create_user_remeinder get_user_reminders services/sms CreateReminderSchema ReminderSchema schemas/crud The api view 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 method. create_reminder post The 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 as a type annotation for the parameter. async CreateReminderSchema payload The statement is a validation to ensure that we can not create a reminder with a past date and time. Moving forward, the 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. if create_user_reminder 12). Start Server Start the uvicorn server with the below command in your terminal: python sms_reminder/main.py : If you get the following error: NOTE 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

Build a Lightning-fast SMS Reminder System With Vonage SMS API and FastAPI Backend!

Too Long; Didn't Read

@abram

Credibility

RELATED STORIES