Source: Building a Notification Service with FastAPI 🚀 | by Joël-Steve N. | Aug, 2024 | Medium

A Guide to Implementing Push Notifications with OneSignal

Push notifications have become essential for modern applications. Whether to inform users about new messages, updates, or alerts, they are an effective way to keep users engaged. In this article, we’ll explore how to create a push notification service using FastAPI and OneSignal, a popular and robust push notification service.

Why FastAPI and OneSignal? 🌟

FastAPI is known for its speed, ease of use, and robust integration with modern tools. Here are a few reasons why FastAPI is ideal for our notification service:

• Performance: Built on ASGI, it offers exceptional performance.
• Ease of Use: With intuitive syntax, it’s easy to learn and use.
• Automatic Documentation: Automatically generates interactive documentation with Swagger UI.

OneSignal is a powerful push notification service that supports multiple platforms including web, mobile, and email. It provides a simple API for sending notifications and managing user segments.

Prerequisites 📋

Before we start, make sure you have the following installed:

• Python 3.7+
• FastAPI
• Uvicorn (ASGI server)
• OneSignal SDK for Python

Installation 🚀

Let’s start by installing the necessary dependencies:

pip install fastapi uvicorn requests

Additionally, you need to install the OneSignal Python SDK:

pip install onesignal-sdk

Project Structure 📁

Here’s the structure of our project:

notifications_service/
├── main.py
├── notifications/
│   ├── __init__.py
│   ├── models.py
│   ├── schemas.py
│   ├── routes.py
│   └── utils.py
├── auth/
│   ├── __init__.py
│   ├── jwt_handler.py
└── requirements.txt

OneSignal Configuration 🔧

1. Sign up for a OneSignal account.
2. Create a new app and get the app_id and API key.

Implementing the Notification Service 💻

1. Set Up FastAPI

In main.py, we’ll set up our FastAPI application:

from fastapi import FastAPI, Depends
from notifications.routes import router as notifications_router
from auth.jwt_handler import JWTBearer

app = FastAPI()

app.include_router(notifications_router, prefix="/notifications", dependencies=[Depends(JWTBearer())])

@app.get("/")
def read_root():
    return {"message": "Welcome to the Notification Service!"}

2. Models and Schemas

In notifications/models.py, define our data models:

from pydantic import BaseModel

class Notification(BaseModel):
    title: str
    message: str
    user_id: str

In notifications/schemas.py, define the Pydantic schemas:

from pydantic import BaseModel

class NotificationSchema(BaseModel):
    title: str
    message: str
    user_id: str

class NotificationResponse(BaseModel):
    success: bool
    message: str

3. Routes

In notifications/routes.py, we’ll define our endpoints:

from fastapi import APIRouter, HTTPException
from notifications.models import Notification
from notifications.schemas import NotificationSchema, NotificationResponse
from notifications.utils import send_notification

router = APIRouter()

@router.post("/", response_model=NotificationResponse)
async def create_notification(notification: NotificationSchema):
    success = send_notification(notification)
    if not success:
        raise HTTPException(status_code=500, detail="Notification failed")
    return NotificationResponse(success=True, message="Notification sent successfully")

4. Sending Notifications

In notifications/utils.py, we’ll implement the logic to send notifications via OneSignal:

import onesignal as onesignal_sdk
from notifications.models import Notification

ONESIGNAL_APP_ID = 'your_onesignal_app_id'
ONESIGNAL_API_KEY = 'your_onesignal_api_key'

onesignal_client = onesignal_sdk.Client(app_id=ONESIGNAL_APP_ID, rest_api_key=ONESIGNAL_API_KEY)

def send_notification(notification: Notification):
    notification_body = {
        'headings': {'en': notification.title},
        'contents': {'en': notification.message},
        'include_player_ids': [notification.user_id]
    }

    response = onesignal_client.send_notification(notification_body)
    return response.status_code == 200

5. Authentication

In auth/jwt_handler.py, we’ll handle JWT authentication:

import jwt
from fastapi import Request, HTTPException

SECRET_KEY = "your_secret_key"

def JWTBearer():
    def verify_jwt(request: Request):
        token = request.headers.get("Authorization")
        if not token:
            raise HTTPException(status_code=403, detail="Token is missing")
        try:
            payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        except jwt.ExpiredSignatureError:
            raise HTTPException(status_code=403, detail="Token has expired")
        except jwt.InvalidTokenError:
            raise HTTPException(status_code=403, detail="Invalid token")
    return verify_jwt

Testing the Service 🛠️

To test our service, we’ll use Uvicorn to start the application:

uvicorn main:app --reload

Then, we can use a tool like Postman to send POST requests to http://localhost:8000/notifications with the following data:

{
    "title": "Hello",
    "message": "This is a test notification",
    "user_id": "<user_device_token>"
}

Conclusion 🎉

And there you have it! We have built a simple yet effective push notification service using FastAPI and OneSignal. This guide has shown you how to structure your project, configure OneSignal, and implement basic functionalities to send notifications.

Remember to secure your application and handle errors appropriately in production. FastAPI offers numerous possibilities for creating performant and scalable services.

Stay Connected 🌐

If you enjoyed this article, feel free to share it and follow my profile for more content like this!

Thanks for reading and happy coding! 💻🚀