API Development using FastAPI
Introduction
Nowadays, API become a necessary component of software development. Stands for Application Programming Interface, API acts like the bridge that communicates between different applications or software using a set of rules and protocols that govern how the communication takes place. API can be used to extract or share data seamlessly, perform actions, and access services provided by a particular software.
API concept of working can easily be understood with this example. Imagine I’m a customer visiting a restaurant, and then the waiter presents me with the menu. I decide what to eat, and the waiter takes note of the order. After that, the waiter takes the order to the kitchen. Once the food is ready, the waiter serves the food to me.
In this example, the waiter acts like the API that links between me and the kitchen as the two applications. All I have to do, after placing the food order, is wait for the waiter to bring my food from the kitchen. I do not have to worry about how the food will be prepared or any other thing that happens inside the kitchen. It is the same as API. When an application or in this case me sends a request to the API or the waiter, the application only expects the response from the API, without caring how the other application or the kitchen processes the request.
Now, to a more relevant example. In a marketplace application or website, one of the activities I do is add some products to my cart. After I click the “Add to Cart” button, what happens next is the application will respond with either a successful or failed notification, depending on the product stocks. What happens from the developer’s perspective is when I click the button, I hit the API with the information about the product, and then the API requests data to the server. The server or back-end retrieves that data, then interprets it, and performs stock checking into the database. The server then sends a response back to this request, whether the product is in stock or out of stock, to the application through the API. The application then turns this response into a visible message for me through the user interface.
When talking about software accessibility to a broad range of users, it can’t be separated from API cause it’s everywhere. The example above is just one of many implementations of API. There are many more examples, such as logging in to social media accounts, checking the weather application, and most of the activities with web or mobile applications.
API Implementation Use Case
It’s clear from the previous section that API is used to solve many software connectivity use cases. It also helps me personally with one of my project’s use cases. In summary, the use case can be seen in the following picture:
There is an Insurance Web Application created by the software developers that will do some sort of calculation when the user clicks the “calculate” button. However, the calculation is not as simple as taking the parameters and using them on a formula. The calculation needs to take data from many tables and after the calculation, the result needs to be inserted into another table so the application can display the data table into the application. This is where API will help to solve the problem.
Based on the picture, in step 1, the Insurance Web Application will send a request to the on-the-fly calculation API. In steps 2 & 3, the request will trigger the API function and run an SQL file that will do the calculation by taking columns from different tables into a calculation algorithm and then inserting the result into a different table. After the calculation and data insertion are finished, in step 4, the API will send a response JSON to the application so the application is notified that the process is done, and the application can show the data to the user. That is the summary of the use case, but how do I manage to realize this concept?
API Development
To create an API, first I have to choose how to build my API. In this case, I choose FastAPI, a high-performance web framework for building API with Python 3.8+ based on standard Python-type hints. FastAPI is chosen because it offers the following benefits:
· Fast to run: With the help of Starlette and pydantic library, it offers very high performance
· Fast to code: It allows for significant increases in development speed.
· Reduced number of bugs: It reduces the possibility of human-induced errors.
· Intuitive: It offers great editor support, with completion everywhere and less time debugging.
· Straightforward: It’s designed to be uncomplicated to use and learn.
· Short: It minimizes code duplication.
· Robust: It provides production-ready code with automatic interactive documentation.
· Standards-based: It’s based on the open standards for APIs, OpenAPI and JSON Schema.
The framework is designed to optimize the developer experience so that I can write simple code to build production-ready APIs with best practices by default. It is also easy to install, just run the following command on my bash and I can get all the functionality.
pip install "fastapi[all]"The next step is to create the API script. For a more structured script, the main API script is divided into two, routers.py and backend.py. The routers.py contains the API endpoint and operation. Endpoint refers to the last part of the URL starting from the first slash character (/). For example, if I have a URL called https://www.example.com/foo/bar, then the endpoint is /foo/bar. With an endpoint, for one API instance, I can separate different resources or APIs. Operation refers to HTTP request methods. These methods define the type of operation to be performed on a resource and are important for enabling various interactions between users and servers. Following are some examples of HTTP requests that are commonly used:
· GET: used to retrieve information and resources from the server.
· POST: for creating new resources on the server.
· PUT: used for updating or replacing an existing resource.
· DELETE: deletes a resource from the server.
And others like OPTIONS, HEAD, PATCH, and TRACE. The example of the routers.py script is as follows.
# routers.py
from fastapi import FastAPI, APIRouter, Depends
from sqlalchemy.orm import Session
from api.shared.database import get_db
from api.shared.schemas import TriggerParams, TriggerResponse
from api.services.v1.productions.backend import (
prc_insurance_calculation
)
app = FastAPI()
router = APIRouter(
prefix = "/v1/productions",
tags = ["v1", "productions"]
)
@router.post("/insurance_calculation", response_model=TriggerResponse)
def route_insurance_calculation(trigger_params: TriggerParams, db: Session = Depends(get_db)):
elapsed_time = prc_insurance_calculation(db, params=trigger_params.model_dump())
response = TriggerResponse(
api_version = "v1",
api_group = "production",
api_function = "insurance_calculation",
elapsed_time = elapsed_time,
params = trigger_params
)
return response
app.include_router(router)This script will do on-the-fly calculations for many insurance metrics, but in this example, I call it insurance_calculation. First, importing the necessary library such as FastAPI for the API and SQLAlchemy for the SQL functionality. Then, there is a variable called app to define the FastAPI instance that will be the main entry point and a variable called router to group the endpoints. After that, I declare the router decorator that consists of the endpoint and the operation that will run the route_insurance_calculation function if the request is sent to the “/v1/productions/insurance_calculation” endpoint with the POST operation. It also includes the response_model parameter to declare the type of data that will be returned.
What the route_insurance_calculation function does is run another function called the prc_insurance_calculation function that is imported from the backend.py script and will be explained later. The route_insurance_calculation function also returns a variable called the response variable that consists of the TriggerResponse class, which has to be the same as the type described in the response_model parameter, that will act as the response body of the API. The response body, like its name, is the response data from the API after the function finishes processing the request from the application, and this request is called a request body. In the route_insurance_calculation function, the request body type is declared through the trigger_params parameter. So, in short, the response_model and the trigger_params parameter act as validation points to make sure the response body and the request body type are the same as the TriggerResponse class and TriggerParams class. These classes are inherited from the pydantic BaseModel which looks like the script below.
from pydantic import BaseModel
# Trigger
class TriggerParams(BaseModel):
version_number_id: int
class TriggerResponse(BaseModel):
api_version: str
api_group: str
api_function: str
elapsed_time: float
params: TriggerParamsFinished with the routers.py, the next is to create the backend.py script as follows.
# backend.py
import os
from sqlalchemy.orm import Session
from api.shared.utils import time_it_prc
from api.shared.database import run_query, upsert_data
from api.shared.models import (
FactInsuranceCalculation
)
QUERY_DIR = os.path.join(os.path.dirname(__file__), "query")
@time_it_prc
def prc_insurance_calculation (db: Session, params: dict):
data = run_query(db, QUERY_DIR, "insurance_calculation_query", params)
upsert_data(db, data, FactInsuranceCalculation)Same as before, I’m importing the necessary library such as SQLAlchemy and the decorator named time_it_prc to calculate how long the function is running. Then, it will run the prc_insurance_calculation function. This function needs the db parameter to connect with the database and the params parameter which is already known from the routers.py, is passed from the trigger_params parameter, specifically the TriggerParams class that contains the version_number_id variable.
After that, it will run another function named run_query function, which simply would be running a query file for calculation named insurance_calculation_query.sql that is located in the place described by the QUERY_DIR variable. The run_query function still needs the db parameter to connect with the database and the params parameter as the binding parameter for the query.
The query result is then stored in the data variable and triggers the upsert_data function for data insertion to the table. The function requires the FactInsuranceCalculation parameter that contains the specification of the destination table, which in this case is named “fact_insurance_calculation”, then it will check the list of columns and their respective type for validation. What’s inside the FactInsuranceCalculation parameter can be seen in the following script.
from sqlalchemy import Column, Integer, Float, String, DateTime, Date
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class FactInsuranceCalculation(Base):
__tablename__ = "fact_insurance_calculation"
version_number_id = Column(Integer, primary_key=True)
insurance_metrics = Column(String(length=100), primary_key=True)
periode = Column(DateTime, primary_key=True)
calculation_result_1 = Column(Float)
calculation_result_2 = Column(Float)
calculation_result_3 = Column(Float)If the columns, data type, and the primary key described in this class do not match with the existing table, then the data insertion will fail.
Back to the prc_insurance_calculation function, note that the function did not return anything because it already covered from the time_it_prc decorator. All I need to know is the function finishes querying and inserting the data, so the decorator that counts the time is enough. That’s why in the routers.py script, this function is defined in a variable named elapsed_time because the result of the function is the elapsed time of the function.
In summary, the script works with the same flow as described in the picture in the API Implementation Use Case section. Starting from step 1, the user of the Insurance Web Application clicks the “calculate” button that will send a request body to the API endpoint which is “/v1/productions/insurance_calculation” with the POST method. In steps 2 & 3, the request body which carries the version_number_id parameter, will trigger the route_insurance_calculation function that contains the prc_insurance_calculation function. Then this function will trigger other functions that will run a calculation query using the version_number_id parameter and the result of the query is inserted into the “fact_insurance_calculation” table. The prc_insurance_calculation will return the elapsed time of the function and be inserted into the response body JSON. In step 4, the response body will be sent by the API back to the application to notify the application that the calculation is successfully done.
The script is now fully finished. But, it’s not all done until I test the script to know if it works as I want. To test the API, testing tools like Postman or cURL in the command line can be used. In this case, I’m using Postman. Simply put the URL with the endpoint, then change the method to POST. Then, in the body tab, type the request body based on the TriggerParams class, and click the “Send” button. The response body can be seen at the bottom part.
It can be seen from the response body that the query and the data insertion process are successfully done within 0.66 seconds.
Conclusion
In this article, I learned the process of creating an API using FastAPI in Python to solve software connectivity use cases. I started by understanding the definition of API, and how it works. Next, I defined the use case and designed ways to solve it using API, developed the API, and also tested my API. From this use case, this concept can be expanded to create more complex and innovative APIs that are production-ready and can be used on other projects. The API can also be integrated with various services on-premises as well as on the cloud services.
