Building a Notification Service with FastAPI and Python

FastAPI is a web framework built for developing APIs with Python. It enables developers to create applications quickly and securely, offering features such as data validation, dependency injection, and support for handling asynchronous requests. In this guide, we will build a simple notification service using FastAPI and SQLite, where the primary focus is on managing notifications in a database and exposing those via API endpoints. This service could be part of a larger system that tracks various types of events or alerts for users.

Setting Up the Project

Step 1: Organize the Project Structure

Start by creating a directory for your project. Organize your files like this:

notification_service/
├── main.py
├── models.py
├── database.py
├── notification.db

Each file has a distinct responsibility. The main.py file contains the API routes for FastAPI. The models.py file holds the database schema for our notification records. The database.py file handles the database connection logic, and notification.db is the SQLite database where we store notification data.

Step 2: Install Dependencies

We will need a few libraries to build this notification service. Install them with:

pip install fastapi uvicorn sqlalchemy pydantic sqlite3

These packages will help us set up FastAPI, run it with Uvicorn, and manage our SQLite database using SQLAlchemy.

Creating the Database Model

Let’s define the database model using SQLAlchemy. In models.py, we’ll define a table for storing notification data.

from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime

Base = declarative_base()

class Notification(Base):
    __tablename__ = "notifications"

    id = Column(Integer, primary_key=True, index=True)
    recipient = Column(String, index=True)
    message = Column(String)
    status = Column(String, default="pending")
    created_at = Column(DateTime, default=datetime.utcnow)

In this Notification model, we define the following fields:

• id: A unique identifier for each notification.
• recipient: A field to store who the notification is for.
• message: The actual content of the notification.
• status: A field to track the state of the notification (e.g., “pending” or “sent”).
• created_at: A timestamp to record when the notification was created.

Database Connection

Now, let’s set up the database connection in database.py. We’ll use SQLite for this example, which is simple to set up and works well for smaller projects.

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from models import Base

SQLALCHEMY_DATABASE_URL = "sqlite:///./notification.db"

engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

def init_db():
    Base.metadata.create_all(bind=engine)

In this file, we define the connection to the SQLite database and create a SessionLocal class that we will use to interact with the database. The init_db() function initializes the database and creates any tables that do not already exist.

FastAPI Application

Now let’s move on to the main file, main.py, where we define the FastAPI application and the API endpoints for managing notifications.

from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from pydantic import BaseModel
from models import Notification
from database import SessionLocal, init_db

app = FastAPI()

# Initialize the database
init_db()

# Dependency for database session
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# Pydantic model for request validation
class NotificationRequest(BaseModel):
    recipient: str
    message: str

@app.post("/notifications/")
async def create_notification(notification_req: NotificationRequest, db: Session = Depends(get_db)):
    notification = Notification(
        recipient=notification_req.recipient,
        message=notification_req.message,
        status="pending",
    )
    db.add(notification)
    db.commit()
    db.refresh(notification)
    return {"status": "Notification created", "notification_id": notification.id}

@app.get("/notifications/{notification_id}")
async def read_notification(notification_id: int, db: Session = Depends(get_db)):
    notification = db.query(Notification).filter(Notification.id == notification_id).first()
    if notification is None:
        return {"error": "Notification not found"}
    return {
        "id": notification.id,
        "recipient": notification.recipient,
        "message": notification.message,
        "status": notification.status,
        "created_at": notification.created_at,
    }

@app.put("/notifications/{notification_id}")
async def update_notification_status(notification_id: int, status: str, db: Session = Depends(get_db)):
    notification = db.query(Notification).filter(Notification.id == notification_id).first()
    if notification is None:
        return {"error": "Notification not found"}
    
    notification.status = status
    db.commit()
    return {"status": "Notification updated", "notification_id": notification.id}

Key Features in main.py

1. Database Dependency: Each request gets its own database session using get_db(). This ensures the connection is closed properly after each request.
2. Pydantic Validation: NotificationRequest is a model that validates incoming request data. This ensures that all notifications include a recipient and message.
3. CRUD Operations: We define three main routes:

• POST /notifications/: Creates a new notification and stores it in the database.
• GET /notifications/{notification_id}: Retrieves the details of a notification by its ID.
• PUT /notifications/{notification_id}: Updates the status of a notification (e.g., from “pending” to “sent”).

Running the Service

To run the service, use Uvicorn. Start the FastAPI application by running the following command in your terminal:

uvicorn main:app --reload

This will launch the FastAPI server at http://127.0.0.1:8000. You can now interact with the service using tools like Postman or cURL.

Testing the API Endpoints

You can test the service by creating and managing notifications using cURL. Here’s an example of how to create a new notification:

curl -X 'POST' \
  'http://127.0.0.1:8000/notifications/' \
  -H 'Content-Type: application/json' \
  -d '{
  "recipient": "john.doe@example.com",
  "message": "You have a new alert."
}'

Once created, you can retrieve the notification using its ID:

curl -X 'GET' 'http://127.0.0.1:8000/notifications/1'

To update the status of a notification, use the following request:

curl -X 'PUT' \
  'http://127.0.0.1:8000/notifications/1?status=sent'

These endpoints allow you to manage notifications in a real-world way. Your system could be expanded further to integrate more features, but this setup offers a clear example of how FastAPI can be used to build a functional service.

Thank you for following along with this tutorial. We hope you found it helpful and informative. If you have any questions, or if you would like to suggest new Python code examples or topics for future tutorials/articles, please feel free to join and comment. Your feedback and suggestions are always welcome!

You can find the same tutorial on Medium.com.