Building a Secure and Robust FastAPI Project for Task Management Part 1

Learn how to develop efficient APIs with FastAPI by building Taskee.

Let’s start with a story 😁, I grew up writing down everything I would do daily, weekly, monthly, and ideas that came by. I even write code on paper. I recently started playing around with FastAPI a cool Python web framework for building scalable and efficient APIs. I thought what a good idea to build an API for Task management, you can use it to create tasks, track them, and stay organized. That will help with maximum productivity.

In this tutorial, we will walk through the process of building a secure robust APIs for a Task Management project using the FastAPI framework. “Taskee” is a platform where you can create tasks, delete, update, and view tasks, it will help you stay organized. This project will cover developing the APIs for the platform, it will feature Project setup, Data Models/Validation and Error Handling, API Endpoints, Documentation, User Authentication/Authorization, Testing, Docker, and Deployment. We will cover Project setup, Data Models/Validation and Error Handling, API Endpoints, and Documentation in Part 1 and continue with the rest in Part 2.

Let’s get started!

1. Project Setup:

Create a new FastAPI project or set up a virtual environment within an existing one.
Install the required dependencies, including FastAPI, a database connector (motor).
Create A FastAPI App and Project Structure.

Prerequisites

Before starting the project some important tools and packages need to be available on your system:

Python 3. x
pip3 (Python Package Manager)
A Code Editor (VS Code, or any of your code, but I will be using VS Code)
Basic Knowledge of Python and Web Development Concepts

Note: If you don’t have any of the above tools, you download them from the links provided below VS Code, Python, You can choose your operating system of choice and install.

You can verify Python is installed correctly by typing:

$ python3

Setting Up the Development Environment

The first step after all the installation, is to create a dedicated folder for your project and set up a virtual environment. Well, the virtual environment is a way to keep your projects organized and solve dependency issues, as you develop many projects over time. It is just a way of isolating each project to be in a container with only its packages no overlaps or conflicts. This ensures consistency and efficiency as you only install packages you need.

Let's create our folder and virtual environment, shall we? 😉 If you using Windows open Command Prompt or terminal on Mac/Linux, then type:

$ mkdir taskee_api
$ cd taskee_api
$ python3 -m venv env
$ source env/bin/activate # On Windows, use venv\\Scripts\\activate

Project Structure

Here we will create our project folders and files like this below:

├── app
│   ├── __init__.py
│   └── server
│       ├── api
|       |    └── v1
|       |    |   └── endpoints
|       |    |   |    └── tasks.py
│       ├── config
|       |    └── config.py
│       ├── models
|       |    └── tasks.py
│       └── database
|       |    |── crud.py
|       |    └── database_connection.py
|                └── app.py
└── requirements.txt
└── main.py
└── .env

Pretty right!!

Installing FastAPI Dependencies

The next step of our development is to install the right dependencies for the project in our created virtual environment. Our project will need some libraries to work for example the web framework for building the APIs in this case FastAPI, a database connector (motor for Mongodb), we will be using MongoDB as our database.

MongoDB Database

MongoDB is a NoSQL database based on document structure in the form of dictionary objects, it is suited for non-transactional applications. It allows very fast search and retrieval of data with a complex structure.

We start by installing MongoDB, you can refer to this Installation guide. Once you are done you can continue to run the daemon mongod process. After everything you can verify whether MongoDB is running in the background by using the CLI command:

$ mongo
$ mongo --version

You are all set with MongoDB installed, now let's get other packages.

We will install the database interfacing library in Python called "motor", along with "fastapi" for building APIs, "uvicorn" as the web server, and "pydantic" as the data validation package for fastapi.

$ pip install fastapi motor uvicorn pydantic

To make programming easier, you should create a requirements field to hold all the versions of the packages you are using in the project. This helps development, avoids conflicts, and reduces package dependency. So in your root directory, you can run the following command and you will see a new created:

$ pip freeze > requirements.txt

Create First App

Now Let's start building our “Taskee” API, before doing anything let's write code inside our to define the entry point of our application and set up the app base server.

app/server/app.py → Main app server route

from fastapi import FastAPI, status
from fastapi.responses import JSONResponse
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

@app.get("/")
async def task_app():
   content = {"Message": "Your Task Manager for Daily Productivity...."}
   return JSONResponse(content=content, status_code=status.HTTP_200_OK)

In the above code, I have imported the main libraries required to run our FastAPI application server. I then declared the app instance using the FastAPI() method. Every application needs to have a filter for traffic and be secured, we have CORSMiddleware for that. You can leave the settings as they are for now.

Now, let's create our first route, we can do that by using the app instance as a decorator and invoking the get method on the (”/”) path. FastAPI is an Asynchronous framework, that allows multiple requests at a time, one request can do an operation while the other is also in operation.

async def task_app()

We want to return some text to the browser when you hit the endpoint, so we create the content dictionary object and return the value using the JSON response method to conform with the JSON format. That’s it our simple route

It is a good practice to have your app entry point configuration in a separate file.

app/main.py

import uvicorn

if __name__ == "__main__":
    uvicorn.run("app.server.app:app", host="0.0.0.0", port=8000, reload=True)

This configuration starts the server using Uvicorn using the file instance of our app, on port 8000 and localhost, the reload parameter allows the server to listen to changes in the entire project and restart the server.

We can now run our simple application and type the command from the terminal:

$ python main.py

You will get a similar output from your terminal:

INFO:     Will watch for changes in these directories: ['/Users/macbookpro/Documents/GitHub/Task-Manager-using-FastAPI']
INFO:     Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [75412] using StatReload
Connection to Mongo DB Server Successfull
INFO:     Started server process [75414]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

You can view the application from the browser.

2. Data Models and Configuration:

Define a data model for the "Task" entity. This model should include fields like task name, description, status (e.g., "completed," "in progress," "pending"), creation timestamp, and any other relevant properties.
Create a configuration file for our environment variables.
Set up a database connection to store and retrieve tasks.

Define Task Data Model

Before we start writing out API routes we need to define our data schema, and configure our database and our config files.

Our data schema/model will created using pydantic to ensure data validation and type annotation.

Navigate and open the file app/server/models/tasks.py

# Importing libraries
from pydantic import BaseModel, Field
from typing import Optional, Union
from datetime import datetime

# Defining the data schema for creating a task
class TasksSchema(BaseModel):
   name: str = Field(...)
   description: str = Field(...)
   task_status: str = Field(...)
   priority: str = Field(...)
   due_date: str = Field(...)
   created_at: Union[datetime, None] = None
   updated_at: Union[datetime, None] = None

   class Config:
      json_schema_extra = {
         "example": [
            {
               "name":"Read a Book",
               "description":"Read a book on entreprenurship for 2 hours",
               "task_status":"completed",
               "priority":"p1",
               "due_date": "2023-09-10",
               "created_at":"2023-09-10T23:23:35.403+00:00",
               "updated_at":"2023-09-10T23:23:35.403+00:00"
            }
         ]
      }

# Defining Data Schema for updating the task
class UpdateTasksSchema(BaseModel):
   name: str = Field(...)
   description: str = Field(...)
   task_status: str = Field(...)
   priority: str = Field(...)
   due_date: str = Field(...)
   updated_at: Union[datetime, None] = None

   class Config:
      json_schema_extra = {
         "example": [
            {
               "name":"Read a Book",
               "description":"Read a book on entreprenurship for 2 hours",
               "task_status":"completed",
               "priority":"p1",
               "due_date": "2023-09-10",
            }
         ]
      }

The TasksSchema class inherits from the BaseModel class to create data fields based on the type of data you are expecting from the user, Pydantic ensures the validation of the data that it is the right type. This schema represents how our data will be saved in the database, the ellipses (…) in the Field method show it is required. the UpdateTasksSchema is used when updating the data in the database.

For both schemas we have used, a Union field to create a ‘updated_at’ and ‘created_at’ object using the datetime function to include the current time.

The class Config is a schema structure example for the documentation FastAPI generates and uses for validation. This helps anyone using the APIs to understand how the data is structured.

Create Config File

To use MongoDB with our application or some other tools that require some key access, we will need to secretly save them on our server and provide access through our config file. Pydantic has a special package to help with that ‘pydantic-settings’ . Run the following command to install it and also create a new file ‘.env’. make sure you are in your main folder.

$ pip install pydantic-settings
$ touch .env

When the installation is done the run the second command. Now open the .env and write this:

MONGODB_URL="mongodb://localhost:27017"

This will allow you to connect the database we will create config for later. The value can be different if you are using Mongodb atlas, but for local it will suffice.

Let's create our config file, since our .env file is a secret file we will use pydantic-settings to access it.

app/server/config/config.py

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    MONGODB_URL: str

    class Config:
        env_file = '.env'

Here, we have created a Settings class that inherits from the BaseSettings class, under it we declared our MongoDB url as str type, and any other env variable from .env. The class Config points to the location of the .env file. Now with this, you can access your key from anywhere in the code without exposing it.

Setup Database Connection

Our database will be used to store tasks users will create, we will use ‘motor’ an asynchronous Mongodb driver, a package we already installed. Create a new file in app/server/database/database_connection.py


# Import Libraries
import asyncio
from motor.motor_asyncio import AsyncIOMotorClient
from app.server.config.config import Settings

# Importing enviroment Variables
settings = Settings()

# Database Configuration

client = AsyncIOMotorClient(settings.MONGODB_URL, serverSelectionTimeoutMS=5000)
client.get_io_loop = asyncio.get_event_loop

# Database Connection Test

try:
    db_connection = client.server_info()
    print("Connection to Mongo DB Server Successfull")
except Exception as e:
    print("Connection Unsuccessfull")
    print(str(e))

# Create Database
database = client.tasks_api

# Database Collections

tasks = database.get_collection("tasks_collection")

We have imported the env variables from my config file safely, next, we create a database client using the AsyncIOMotorClient method, and we pass the MongoDB URL as the database to create an async connection to.

Inside the try-except block, we test the connection to the database, next we head on to create the database and subsequently the collections. So the logic here is we create one database, and can create multiple collections, if the collection is not available it will be created else it will use the existing one.

Motor is an async driver so we need to use asyncio to continue the event loop for the database running.

CRUD Operations

To use the collections we have created, there have to be create, read, update, and delete functions available. In this section, we will go through that.

Create/Open the file app/server/database/crud.py

We first import our modules, from models and database configs. Each Mongodb object is created with a unique id, in our case ObjectID from Bson is what we will use.

from app.server.database.database_connection import tasks
from app.server.models.tasks import TasksSchema, UpdateTasksSchema
from bson.objectid import ObjectId

# Create the necessary endpoints to handle 
# CRUD operations for tasks (Create, Read, Update, Delete).

# Create task
async def create_task_db(task: TasksSchema):
    task = task.model_dump()
    try:
        new_task = await tasks.insert_one(task)
        get_new_task = await tasks.find_one({"_id": new_task.inserted_id})
        return get_new_task
    except Exception as e:
        return {"Error_message":str(e)}

# Get one task
async def get_single_task_db(id: str):
    try:
        task = await tasks.find_one({"_id":ObjectId(id)})
        if task:
            return task
    except Exception as e:
        return {"Error_message": str(e)}

# Get all tasks
async def get_all_tasks_db():
    task_data = []
    try:
        get_all_tasks = tasks.find()
        if get_all_tasks:
            async for task in get_all_tasks:
                task_data.append(task)
            return task_data
    except Exception as e:
        return {"Error_message": str(e)}

# Update one task
async def update_single_task_db(id: str, task_data: UpdateTasksSchema):
    task = task_data.model_dump()
    try:
        update_task = await tasks.update_one({"_id":ObjectId(id)}, {"$set":{
            "name":task["name"],
            "description": task["description"],
            "priority": task["priority"],
            "task_status": task["task_status"],
            "due_date": task["due_date"],
            "updated_at": task["updated_at"]
        }})
        if update_task:
            get_task = await tasks.find_one({"_id":ObjectId(id)})
            return get_task
    except Exception as e:
        return {"Error_message": str(e)}

# Delete one task
async def delete_single_task_db(id: str):
    try:
        result = await tasks.delete_one({"_id":ObjectId(id)})
        return {"message":"Successfully Deleted task", "result":result}
    except Exception as e:
        return {"Error_message": str(e)}

# Delete all tasks
async def delete_all_tasks_db():
    try:
        result = await tasks.delete_many({})
        return {"message":"All Database Deleted"}
    except Exception as e:
        return {"Error_message": str(e)}

In the code above, we have created the asynchronous operations to create, read, update, and delete task data from the database collection via motor.

I used UpdateTasksSchema and TasksSchema as parameters for the create_task and update_task functions, I am passing directly the data schema from my endpoint so I need to convert it to a dictionary to be saved properly in my database. That is why I call ‘model_dump()’ on the data to get the dictionary object.

For each section we have implemented a try-except for proper error handling, this allows us to see exactly what the error is.

3. API Endpoints:

Create the necessary endpoints to handle CRUD operations for tasks (Create, Read, Update, Delete).
Implement proper HTTP methods (GET, POST, PUT, DELETE) and status codes to ensure RESTful API design.
Our Endpoints include:
- GET /tasks: Retrieve a list of all tasks.
- GET /tasks/{task_id}: Retrieve a specific task by its ID.
- POST /tasks: Create a new task.
- PUT /tasks/{task_id}: Update an existing task.
- DELETE /tasks/{task_id}: Delete a task.
- DELETE /tasks: Delete all tasks.

In this section, we will implement our endpoints to carry out the CRUD operations we have implemented above. Let’s code…

Navigate to your app/server/api/v1/endpoints/tasks.py We need to import our modules like models, database, and other packages.

app/server/api/v1/endpoints/tasks.py

from fastapi import APIRouter, status, HTTPException, Form, Body
from fastapi.responses import JSONResponse
from fastapi.encoders import jsonable_encoder
from datetime import datetime, time, timedelta
from typing import Annotated, Any, List
from bson.objectid import ObjectId

from app.server.database.crud import (
    create_task_db,
    get_single_task_db,
    get_all_tasks_db,
    update_single_task_db,
    delete_single_task_db,
    delete_all_tasks_db
)
from app.server.database.database_connection import tasks
from app.server.models.tasks import (
    TasksSchema,
    UpdateTasksSchema
)

router = APIRouter()

In the above you will notice I have a new ‘router’ instance not ‘app’, the reason behind that is we can create a single app instance and route different endpoints to it. This way we can create many routes, and just configure the path and the APIRouter class.

You can keep adding the code to the file..

POST /tasks: Create a new task.

# Create task route
@router.post("/", response_model=TasksSchema, response_description="Creating A Task")
async def create_task(name: Annotated[str, Form()],
                      description: Annotated[str, Form()],
                      task_status: Annotated[str, Form()],
                      priority: Annotated[str, Form()],
                      due_date: Annotated[str, Form()]) -> Any:
    try:
        created_at = datetime.utcnow()
        # Create task function
        task = TasksSchema(
            name=name,
            description=description,
            task_status=task_status,
            priority=priority,
            due_date=due_date,
            created_at=created_at,
            updated_at=None
        )
        new_task = await create_task_db(task)

        response = {
            "id": str(new_task["_id"]),
            "message": "Task Created Successfully"
        }
        response = jsonable_encoder(response)

        return JSONResponse(content=response, status_code=status.HTTP_201_CREATED)
    except Exception as e:
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e))

We have created the Post Request endpoint to create a task on the platform. We have used if you noticed @router.post since we created our router instance to handle everything. The response model allows us to specify the schema we will use and how the data should be presented, it ensures data validation through Pydantic. This only accepts Post requests.

The function has a form field for accepting data as form data from the users. We used Annotated for data validation and type safety so that we are sure that what the user provided is valid.

The Try clause gets the current time to create a schema object and creates the task using the create_task_db functions. The created task is transformed into the response to be sent to when successful, the response has to be JSON serializable, so we use jsonable_encoder GET /tasks: Retrieve a list of all tasks.

# Get all Tasks
@router.get("/", response_model=List[TasksSchema], response_description="Get all Tasks")
async def get_all_tasks():
    all_tasks_list = []
    try:
        get_all_tasks = await get_all_tasks_db()
        if get_all_tasks:
            for task in get_all_tasks:
                task_data = TasksSchema(
                    name=task["name"],
                    description=task["description"],
                    task_status=task["task_status"],
                    priority=task["priority"],
                    due_date=task["due_date"],
                    created_at=task["created_at"],
                    updated_at=task["updated_at"]
                )
                all_tasks_list.append(task_data)
            response = jsonable_encoder(all_tasks_list)
            # print(response)
        return JSONResponse(content=response, status_code=status.HTTP_200_OK)
    except Exception as e:
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e))

In the code above we have created the Get request endpoint to retrieve all the tasks created on the platform. We return the result of the values as a list of objects based on the TasksSchema to ensure data validation is applied. With the try-except block, we have checked the database for tasks and returned appropriate messages using JSON response or http exception.

GET /tasks/{task_id}: Retrieve a specific task by its ID.

# Get Single Task
@router.get("/{id}", response_model=TasksSchema, response_description="Get Single Task")
async def get_single_task(id: str):
    try:
        get_task = await get_single_task_db(id)
        if get_task:
            response = TasksSchema(
                name=get_task["name"],
                description=get_task["description"],
                task_status=get_task["task_status"],
                priority=get_task["priority"],
                due_date=get_task["due_date"],
                created_at=get_task["created_at"],
                updated_at=get_task["updated_at"]
            )
            response = jsonable_encoder(response)
        return JSONResponse(content=response, status_code=status.HTTP_200_OK)
    except Exception as e:
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e))

To get a single task we use the ID as a parameter when calling the endpoint, using the ID we check the database to see if the data is available else return an error message. When the data is available we validate it through TaskSchema when creating a response return it to the user.

PUT /tasks/{task_id}: Update an existing task.

# Update Single Task
@router.put("/{id}", response_model=TasksSchema, response_description="Update Single Task")
async def update_single_task(id: str, task: UpdateTasksSchema = Body(...)):
    try:
        task = task.model_dump()
        updated_at = datetime.utcnow()
        updated_task = UpdateTasksSchema(
            name=task['name'],
            description=task['description'],
            task_status=task['task_status'],
            priority=task['priority'],
            due_date=task['due_date'],
            updated_at=str(updated_at)
        )
        update_task = await update_single_task_db(id, updated_task)
        if update_task:
            response = TasksSchema(
                name=update_task["name"],
                description=update_task["description"],
                task_status=update_task["task_status"],
                priority=update_task["priority"],
                due_date=update_task["due_date"],
                created_at=update_task["created_at"],
                updated_at=update_task["updated_at"]
            )
            response = jsonable_encoder(response)
        return JSONResponse(content=response, status_code=status.HTTP_200_OK) 
    except Exception as e:
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e))

To Update the data already created, we use the Put method. The put method allows us to update the whole data and return the updated data. We use the id as a query parameter, to get the task in view, the data we get is based on our Pydantic schema, we convert it to a dictionary object and ensure data validation. We get the updated_at using a DateTime object. After we successfully update the task we can use our TasksSchema to validate it and encode it for our response object.

DELETE /tasks/{task_id}: Delete a task.

@router.delete("/{id}", response_description="Delete a Single Task")
async def delete_single_task(id: str):
    try:
        delete_task = await delete_single_task_db(id)

        response = {"Message": f"Task with Id:{id} is Successfuly Deleted!!"}

        response = jsonable_encoder(response)

        return JSONResponse(content=response, status_code=status.HTTP_200_OK) 
    except Exception as e:
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e))

You can delete a task by just using the Delete method and id as the query parameter.

DELETE /tasks: Delete all tasks.

@router.delete("/", response_description="Delete all Tasks")
async def delete_all_tasks():
    try:
        delete = await delete_all_tasks_db()

        response = {"Message": f"All Tasks Successfuly Deleted!!"}

        response = jsonable_encoder(response)

        return JSONResponse(content=response, status_code=status.HTTP_200_OK) 
    except Exception as e:
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e))

To delete all you just need the Delete method, while we already have the functions for clearing the whole collection.

4. Documentation:

Generate API documentation using FastAPI's built-in support for OpenAPI and Swagger.
Ensure that the API documentation is clear and provides information on how to use each endpoint.

Finally, we can run our application test and utilize the online Swagger documentation. The models and configurations we wrote will be very helpful because they will serve as a guide for sending requests using either the terminal or front-end, as it provide the necessary structure.

Before then let's make some updates to our app/server/app.py file

from fastapi import FastAPI, status
from fastapi.responses import JSONResponse
from fastapi.middleware.cors import CORSMiddleware

# Task API
from app.server.api.v1.endpoints.tasks import router as TasksRouter

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Task Application routes
app.include_router(TasksRouter, tags=["Tasks Management"], prefix="/tasks")

@app.get("/")
async def task_app():
   content = {"Message": "Your Task Manager for Daily Productivity...."}
   return JSONResponse(content=content, status_code=status.HTTP_200_OK)

We have added the task router to the application, this allows the APIs to be accessible.

Now you can type the command:

$ python main.py

Visit localhost:8000/docs You will get an interface like this one with all the endpoints visible and ready to be tested.

Post: Create a task

This is how the post request looks like you can click on Try Out to create a new task.

You have successfully created a new task. Now let's check the database to get all data and a single one.

Get: Retrieve Tasks/Task

Get All Tasks

The above images show you how to get a single task using the ID.

Put: Update A Single Task

You provide the ID and input the update you want. You can write the values using a JSON format.

Successfully updated the task, now you can the updated_at field has a value.

Delete single and Delete All

You provide an ID of the task you want to delete.

Delete everything in the collection.

Wow You successfully created and tested your robust Taskee API, you can go and celebrate.

Conclusion

In conclusion, you have now completed the development of a robust and efficient API for task management. The API allows users to create, read, update, and delete tasks, providing a seamless experience for managing tasks. With proper documentation and error handling in place, the API is ready for deployment and use in real-world scenarios. Great job!

In the next part, we will look at things like:

User and Profile Management
Authorization and Authentication
Pagination and Filtering
Testing and Pagination

You can find the complete code of the tutorial on Github

Musa Wakili ML