We have data from 201 municipalities during the period between 1990-2022. The data presents the mean wealth per municipality and year.
FastAPI
API Development
FastAPI
Python
SQL
CRUD Operations
SQLAlchemy
Introduction
Did you ever dream to build your own fast API written in Python?
FastAPI is a modern, high-performance web framework for building APIs with Python. It blends the ease of use of Python with exceptional speed, making it an ideal choice for developing robust and efficient web applications. Whether you’re crafting a simple REST API or a complex microservices architecture, FastAPI will help you with that.
The installation of FastAPI is very simple and it involves the installation of two packages: (1) FastAPI and (2) Uvicorn. The first one is the framework itself and the second one is the server that will run the API.
In this tutorial, we will build a simple API that will allow us to perform basic CRUD (Create, Read, Update, Delete) operations on a database. The access to the database will be done by SQLAlchemy, which is a Python SQL toolkit for SQL databases. By the means of fastAPI, we will also provide a documentation of the API, which will be accessible through a web browser.
Data
In this post we are going to talk about taxes. More specifically, we will focus on the tax system of Switzerland. Switzerland’s unique tax system is characterized by its federal structure, granting significant fiscal autonomy to its municipalities (cities) and cantons (states). This system encourages competition among cantons and municipalities while also fostering a sense of local ownership and control over financial matters. As a result, Switzerland’s tax landscape is complex, diverse, and reflective of the nation’s commitment to decentralized governance.
::: {#cell-fig-income_tax_zurich .cell include=‘false’ execution_count=4}
:::
Figure 1 shows a positive trend of average wealth in the canton of Zurich per year. The mean value constantly increased from 1990 to 2022. It is quite impressive that the average value is more than doubled in 32 years (considering very low inflation in Switzerland).
Database
The data is stored in a database, a crucial endeavor that lays the foundation for simulating a dynamic production environment. We are going to use sqlalchemy: a remarkable tool empowers us to seamlessly alter the connection string enabling effortlessly establishment of connections across a diverse database types.
First, the connection is defined by sqlalchemy.create_engine() and subsequently used in order to save the data to the database using pandas method to_sql().
::: {#lst-database .cell output=‘false’ execution_count=7}
import statsmodels.api as sm
from sqlalchemy import create_engine
# Create a connection to the database
engine = create_engine(
## Path to database
'sqlite:///../../data/FastAPI.db', # <1>
)
## Create a table in new database
(
df
[['Municipality','Year','Tax']]
.to_sql(
name = 'fastapi',
con = engine,
if_exists = 'replace',
index = False,
)
);:::
After the creation of the fastapi table within the FastAPI database, we can continue with the creation of the API.
Building our API
The building blocks of our API-service consists of basic CRUD (Create, Read, Update, Delete) operations as follows:
| Function (api endpoints) | Description |
|---|---|
| Get Data for a Given Year | Returns wealth tax data for a given year. |
| Get Data for a Given Municipality | Returns wealth tax data for a given municipality. |
| Get Data for a Given District | Returns wealth tax data for a given district. |
| Get Data for the Canton | Returns wealth tax data for the entire canton. |
| Create New Entry | Creates a new entry in the wealth tax table. |
| Update Tax Entry | Updates the wealth tax value for a given municipality and year. |
| Update Year Entry | Updates the year value for a given municipality and year. |
| Delete Entry | Deletes an entry from the wealth tax table. |
Our api-service is going to use no authentication for its endpoints. This is not recommended for production environments. However, it is a good starting point for learning purposes.
Next, we are going to discuss and review portions of the code that are relevant for the creation of the API. The complete code can be found in the end of the article.
Note: An API endpoint serves as a specific URL or URI through which an application can interact with a particular functionality or resource on a server. It acts as a gateway, allowing clients to send requests for data or perform actions, while the server processes these requests and sends back the appropriate response. Here’s an example of an API endpoint for a hypothetical online bookstore:
API Base URL: https://api.bookstore.com
Example Endpoints:
Action: Retrieve Book Information Endpoint: /books/{book_id}
Description: Retrieves detailed information about a specific book.
Example URL: https://api.bookstore.com/books/123
Setting Up the Project
To begin, we import the necessary libraries including pandas, pydantic, and fastapi. We’re also using sqlalchemy for database interactions. Our code starts by creating a database connection using the SQLAlchemy library. We’ve chosen an SQLite database for this demonstration. However, you can use any database you like.
import pandas as pd
from pydantic import BaseModel, Field
from fastapi import FastAPI, HTTPException,
from sqlalchemy import create_engine, MetaData, Table, select, insert, update, delete
## Create a metadata object
metadata = MetaData()
## Create a connection to the database
engine = create_engine(
## Path to database
'sqlite:///../../data/FastAPI.db', # <1>
)
## Reflect census table from the engine: census
fastapi = Table(
"fastapi",
metadata,
autoload_with = engine,
)Moreover, we create a utility function to connect to the database and return a json-style format. This function will be used in the endpoints to fetch data from the database and return the dictionary, which will be converted to json format by FastAPI.
## Helper function to fetch data
def fetch_return_dict(
stmt,
engine = engine,
) -> dict:
"""
Utility function to convert sql query results to dict via pandas dataframe
"""
## Create a connection to the database
connection = engine.connection()
data = connection.execute(stmt).fetchall()
connection.close()
return (
pd.DataFrame(
data = data
)
.to_dict(
orient = "records",
)
)FastAPI Object
We instantiate a FastAPI app object with a title, description, and version. This app will serve as the backbone of our API. The description section contains details about the API and apear in the documentation.
Defining the Data Model
FastAPI encourages the use of Pydantic models for data validation and serialization. Pydantic is a Python library that simplifies the process of data validation and settings management in applications. It allows you to define data schemas using Python data classes with type annotations, and automatically validates and parses incoming data according to these schemas.
We’ve defined IncomeTaxModel, MunicipalityDataOut, IncomeTaxDataOut, and YearDataOut as Pydantic models (BaseModel and type annotation) to structure the data that will be sent and received by the API endpoints.
class TaxModel(BaseModel):
Municipality : str = Field("Name of item")
Tax : int = Field("Average tax for a given year")
Year : int = Field("Year of meassurement")
class MunicipalityDataOut(BaseModel):
Tax : int = Field("Average tax for a given year")
Year : int = Field("Year of meassurement")
class TaxDataOut(BaseModel):
Municipality : str = Field("Name of item")
Year : int = Field("Year of meassurement")
class YearDataOut(BaseModel):
Municipality : str = Field("Name of item")
Tax : int = Field("Average tax for a given year")Building API Endpoints
We proceed to define various API endpoints using FastAPI’s decorators. These endpoints cover different scenarios, from fetching data for a specific year to creating and updating entries in the database. Each endpoint corresponds to a specific HTTP method (GET, POST, PUT, DELETE) and handles specific types of requests.
For instance, the index route handles a basic GET request and returns all data from the database, ordered by the year. The get_data_of_year route fetches data for a given year, while get_municipality_data and get_district_data retrieve data for specific municipalities or districts, respectively. Additionally, the create_new_entry, update_tax_entry, update_year_entry, and delete_tax_entry routes allow for CRUD (Create, Read, Update, Delete) operations on the data.
Next, I will show you the incomplete code for the endpoints (the complete code can be found in the end of the article):
@app.get("/")
def index():
return something
## Create a route to return data for a given year
@app.get("/year/{year}")
def get_data_of_year():
return something
## Create a route to return data for a given city
@app.get("/municipality/{municipality}")
def get_municipality_data():
return something
## Create a route to return data for a given district
@app.get("/district/{district}")
def get_district_data():
return something
## Create a route to return data from the canton
@app.get("/canton/")
def get_canton_data():
return something
## Create a new entry
@app.post('/entry/{municipality}/{year}/{tax}')
def create_new_entry():
return something
## Update an income tax entry
@app.put("/update_tax/{municipality}/{year}/{tax}")
def update_tax_entry():
return something
## update year entry
@app.put("/update_year/{municipality}/{year_old}/{year_new}")
def update_year_entry():
return something
@app.delete("/delete/{municipality}/{year}/{tax}")
def delete_tax_entry():
return somethingThe components of the the fastapi endpoints are as follows:
- Decorators are used to define the route of each endpoints and the HTTP-method.
- Type annotation of the parameters allows the definition and validation of the data type of the parameters.
Error Handling
The code also incorporates error handling. For instance, if a user attempts to create a new entry that already exists in the database, an HTTPException with an appropriate status code and detail message is raised. Similarly, error handling is employed for updating and deleting entries that don’t exist in the database. The following codes are use: 400 for Bad Request, 404 for Not Found, and 500 for Internal Server Error. More information can be found here.
The implementation of error handling is done by raise HTTPException() as follows:
API-Documentation
FastAPI provides a built-in documentation system that is automatically generated based on the API’s metadata and doc-strings provided. The documentation is accessible at the /docs and /redoc endpoint (we focus in this tutorial on /docs page only). The documentation is interactive and allows users to test the API directly from the browser. The documentation is also available in JSON format at the /openapi.json endpoint.
Conclusion
In this article, we’ve walked through the creation of a FastAPI web service for accessing income tax data. We’ve covered the setup process, defining data models, creating API endpoints, and handling database interactions. FastAPI’s intuitive syntax and powerful features make it a fantastic choice for building efficient and robust APIs. This project serves as a foundation, demonstrating how to construct APIs that facilitate data retrieval and manipulation, crucial tasks in today’s data-centric world.
Overall, this FastAPI-based web service exemplifies the elegance and utility of modern Python frameworks in building APIs that bridge the gap between data and applications.
Appendix: Complete code
::: {#cell-31 .cell code-fold=‘false’ execution_count=8}
import pandas as pd
from pydantic import BaseModel, Field
from fastapi import FastAPI, HTTPException
from sqlalchemy import create_engine, MetaData, Table, select, insert, update, delete
## Create a metadata object
metadata = MetaData()
## Create a connection to the database
engine = create_engine(
## Path to database
'sqlite:///../../data/FastAPI.db', # <1>
)
## Reflect census table from the engine: census
fastapi = Table(
"fastapi",
metadata,
autoload_with = engine,
)
## Helper function to fetch data
def fetch_return_dict(
stmt,
engine = engine,
) -> dict:
"""
Utility function to convert sql query results to dict via pandas dataframe
"""
## Create a connection to the database
connection = engine.connect()
data = connection.execute(stmt).fetchall()
connection.close()
return (
pd.DataFrame(
data = data
)
.to_dict(
orient = "records",
)
)
## Instantiate a FastAPI object
app = FastAPI(
title = "My first API",
description = """
API for access of wealth tax data from the canton of Zurich, Switzerland.
Basic CRUD operations are supported (Create, Read, Update, Delete)
# To-Do:
- Add tag parameter to routes decorators see https://fastapi.tiangolo.com/tutorial/metadata/
- Check HTTPException status codes
# Future:
- Add more routes
- Create Authentication
- Create separate modules for routes, database connection, etc.
© 2023, Bernardo Freire Barboza da Cruz
""",
version = "0.0.1",
)
## Create a BaseModel Child class for data representation
class TaxModel(BaseModel):
Municipality : str = Field("Name of item")
Tax : int = Field("Average tax for a given year")
Year : int = Field("Year of meassurement")
class MunicipalityDataOut(BaseModel):
Tax : int = Field("Average tax for a given year")
Year : int = Field("Year of meassurement")
class TaxDataOut(BaseModel):
Municipality : str = Field("Name of item")
Year : int = Field("Year of meassurement")
class YearDataOut(BaseModel):
Municipality : str = Field("Name of item")
Tax : int = Field("Average tax for a given year")
## Create a index route
@app.get("/")
def index() -> list[TaxModel]:
"""
Returns all data from the fastapi table
"""
stmt = (
## Select ALL columns from the census table
select(
fastapi
)
## Order by the Year column
.order_by(
fastapi.columns.Year.desc()
)
)
result = fetch_return_dict(stmt)
if result:
return result
else:
raise HTTPException(
status_code = 404,
detail = f"Item not found",
)
## Create a route to return data for a given year
@app.get("/year/{year}")
def get_data_of_year(year: int) -> list[YearDataOut]:
"""
Returns all data from the fastapi table for a given year
"""
## Create an sql statement to select ALL columns from the census table
stmt = (
## Select ALL columns from the census table
select(
fastapi.columns.Municipality,
fastapi.columns.Tax
)
## Order by the Year column
.order_by(
fastapi.columns.Year.desc()
)
.where(
fastapi.columns.Year == year
)
)
## Execute the statement and fetch the results
result = fetch_return_dict(stmt)
## If the result is not empty return it, otherwise raise an exception
if result:
return result
else:
raise HTTPException(
status_code = 404,
detail = f"Item not found",
)
## Create a route to return data for a given city
@app.get("/municipality/{municipality}")
def get_municipality_data(municipality: str) -> list[MunicipalityDataOut]:
"""
Returns all data from the fastapi table for a given municipality
"""
## Create an sql statement to select ALL columns from the census table
stmt = (
## Select ALL columns from the census table
select(
fastapi.columns.Year,
fastapi.columns.Tax
)
## Order by the Year column
.order_by(
fastapi.columns.Year.desc()
)
.where(
(~fastapi.columns.Municipality.startswith('Bezirk')) &
(~fastapi.columns.Municipality.startswith('Region')) &
(fastapi.columns.Municipality == municipality)
)
)
## Execute the statement and fetch the results
result = fetch_return_dict(stmt)
## If the result is not empty return it, otherwise raise an exception
if result:
return result
else:
raise HTTPException(
status_code = 404,
detail = f"Item not found",
)
## Create a route to return data for a given district
@app.get("/district/{district}")
def get_district_data(district: str) -> list[MunicipalityDataOut]:
"""
Returns all data from the fastapi table for a given district
"""
## Create an sql statement to select ALL columns from the census table
stmt = (
## Select ALL columns from the census table
select(
fastapi.columns.Year,
fastapi.columns.Tax
)
## Order by the Year column
.order_by(
fastapi.columns.Year.desc()
)
.where(
(fastapi.columns.Municipality.startswith('Bezirk')) &
(fastapi.columns.Municipality.contains(district))
)
)
## Execute the statement and fetch the results
result = fetch_return_dict(stmt)
## If the result is not empty return it, otherwise raise an exception
if result:
return result
else:
raise HTTPException(
status_code = 404,
detail = f"Item not found",
)
## Create a route to return data from the canton
@app.get("/canton/")
def get_canton_data() -> list[MunicipalityDataOut]:
"""
Returns all data from the fastapi table for the canton
"""
## Create an sql statement to select ALL columns from the census table
stmt = (
## Select ALL columns from the census table
select(
fastapi.columns.Year,
fastapi.columns.Tax
)
## Order by the Year column
.order_by(
fastapi.columns.Year.desc()
)
.where(
fastapi.columns.Municipality.contains('Kanton')
)
)
## Execute the statement and fetch the results
result = fetch_return_dict(stmt)
## If the result is not empty return it, otherwise raise an exception
if result:
return result
else:
raise HTTPException(
status_code = 404,
detail = f"Item not found",
)
## Create a new entry
@app.post('/entry/{municipality}/{year}/{tax}')
def create_new_entry(
municipality : str = None,
year : int = None,
tax : int = None,
):
"""
Create a new entry in database
"""
in_stmt = (
select(
fastapi
)
.where(
fastapi.columns.Municipality == municipality,
fastapi.columns.Year == year,
)
)
connection = engine.connect()
result = connection.execute(in_stmt).fetchall()
if result:
raise HTTPException(
status_code=405,
detail = f"Item with name {municipality} and year {year} already exists. "
+ "Use update to change this value or delete this entry and retry",
)
else:
stmt = (
insert(
fastapi
)
.values(
Municipality = municipality,
Year = year,
Tax = tax,
)
)
connection.execute(stmt)
connection.commit()
connection.close()
return {"success": f"Item with name {municipality}; tax of {tax}; and year {year} added."}
## Update an income tax entry
@app.put("/update_tax/{municipality}/{year}/{tax}")
def update_tax_entry(
municipality : str = None,
year : int = None,
tax : int = None,
):
"""
Update income tax value for a given municipality and year
"""
in_stmt = (
select(
fastapi
)
.where(
fastapi.columns.Municipality == municipality,
fastapi.columns.Year == year,
)
)
connection = engine.connect()
result = connection.execute(in_stmt).fetchall()
if not result:
raise HTTPException(
status_code=404,
detail = f"Item with name {municipality} and year {year} not found. "
+ "Only values available in database can be updated"
)
else:
stmt = (
update(
fastapi
)
.where(
fastapi.columns.Municipality == municipality,
fastapi.columns.Year == year,
)
.values(
Tax = tax,
)
)
connection.execute(stmt)
connection.commit()
connection.close()
return {"success": f"Item with name {municipality}; and year {year} updated to tax of {tax};"}
## update year entry
@app.put("/update_year/{municipality}/{year_old}/{year_new}")
def update_year_entry(
municipality : str = None,
year_old : int = None,
year_new : int = None,
):
"""
Update year of a municipality entry
"""
in_stmt = (
select(
fastapi
)
.where(
fastapi.columns.Municipality == municipality,
fastapi.columns.Year == year_old,
)
)
connection = engine.connect()
result = connection.execute(in_stmt).fetchall()
if not result:
raise HTTPException(
status_code=404,
detail = f"Item with name {municipality} and year {year_old} not found. "
+ "Only values available in database can be updated"
)
else:
stmt = (
update(
fastapi
)
.where(
fastapi.columns.Year == year_old,
)
.values(
Year = year_new,
)
)
connection.execute(stmt)
connection.commit()
connection.close()
return {"success": f"Item with name {municipality}; and year {year_old} updated to year {year_new};"}
@app.delete("/delete/{municipality}/{year}/{tax}")
def delete_tax_entry(
municipality : str = None,
year : int = None,
tax : int = None,
):
"""
Delete entry given municipality, year and income tax value
"""
in_stmt = (
select(
fastapi
)
.where(
fastapi.columns.Municipality == municipality,
fastapi.columns.Year == year,
)
)
connection = engine.connect()
result = connection.execute(in_stmt).fetchall()
if not result:
raise HTTPException(
status_code=404,
detail = f"Item with name {municipality} and year {year} not found. "
+ "Only values in database can be deleted",
)
else:
stmt = (
delete(
fastapi
)
.where(
fastapi.columns.Municipality == municipality,
fastapi.columns.Year == year,
fastapi.columns.Tax == tax,
)
)
connection.execute(stmt)
connection.commit()
connection.close()
return {"success": f"Item with name {municipality}; tax of {tax}; and year {year} deleted."}:::