fastapi-plugin-notification 0.1.0

Notification plugin for FastAPI projects using FastAPI SDK and Postmark provider.

FastAPI Plugin Notification

A notification plugin for FastAPI that integrates with FastAPI SDK and Postmark provider to send notifications via email and store them in a database.

Features

• 📧 Email Notifications: Send notifications via Postmark email provider
• 💾 Database Storage: Store notifications in MongoDB using ODMantic
• 🔐 User Ownership: Automatic ownership filtering based on user claims
• 📊 Notification Management:
  • Get all notifications for a user
  • Mark notifications as seen/acknowledged
  • Get unseen notification count
• 🚀 FastAPI Integration: Full CRUD API endpoints with authentication
• 📦 PyPI Package: Easy to install and use in any FastAPI project

Installation

Using UV:

uv add fastapi-plugin-notification

Or using pip:

pip install fastapi-plugin-notification

Quick Start

1. Initialize the Notification Client

from fastapi import FastAPI, Depends
from odmantic import AIOEngine
from fastapi_provider_postmark import PostmarkProvider
from fastapi_plugin_notification import NotificationClient

app = FastAPI()

# Initialize Postmark provider
postmark = PostmarkProvider(
    api_key="your-postmark-api-key",
    from_email="[e-mail]",
    from_name="My App",
)

# Initialize notification client (no DB engine needed)
notification_client = NotificationClient(
    postmark_provider=postmark,
    default_template_id=123456,  # Your Postmark template ID
)

# Dependency for notification client
def get_notification_client() -> NotificationClient:
    return notification_client

# Dependency for database engine
def get_db() -> AIOEngine:
    return AIOEngine(database="your_database")

2. Add Notification Routes to Your App

from fastapi_plugin_notification.app import create_notification_router

# Add notification routes
notification_router = create_notification_router(
    prefix="/notifications",
    get_db=lambda: engine,
)

app.include_router(notification_router)

3. Send Notifications

@app.post("/send-welcome")
async def send_welcome_notification(
    user_id: str,
    user_email: str,
    notification_client: NotificationClient = Depends(get_notification_client),
    db: AIOEngine = Depends(get_db),
):
    """Send a welcome notification to a user."""
    notification = await notification_client.send_notification(
        db_engine=db,
        name="welcome",
        user_id=user_id,
        user_email=user_email,
        claims=request.state.claims,  # Required: user claims for ownership verification
        metadata={
            "welcome_message": "Welcome to our platform!",
        },
        link="https://example.com/dashboard",
        link_name="Go to Dashboard",
        template_variables={
            "user_name": "John Doe",
            "action_url": "https://example.com/dashboard",
        },
    )
    return {"notification_id": notification.uuid, "message": "Notification sent"}

4. Get User Notifications

The plugin automatically provides these user-specific endpoints:

• GET /notifications/me - Get all notifications for the current user
• GET /notifications/me?seen=false - Get only unseen notifications
• GET /notifications/me/{notification_id} - Get a specific notification (automatically marks as seen)
• DELETE /notifications/me/{notification_id} - Delete a notification
• GET /notifications/count/unseen - Get count of unseen notifications

Usage Examples

Sending a Notification with Email

# In a FastAPI route handler
@app.post("/send-password-reset")
async def send_password_reset(
    user_id: str,
    user_email: str,
    notification_client: NotificationClient = Depends(get_notification_client),
    db: AIOEngine = Depends(get_db),
):
    notification = await notification_client.send_notification(
        db_engine=db,
        name="password_reset",
        user_id=user_id,
        user_email=user_email,
        claims=request.state.claims,  # Required: user claims for ownership verification
        metadata={
            "reset_token": "abc123",
            "expires_in": 3600,
        },
        link="https://example.com/reset-password?token=abc123",
        link_name="Reset Password",
        template_variables={
            "reset_link": "https://example.com/reset-password?token=abc123",
            "expires_in_minutes": 60,
        },
    )
    return {"notification_id": notification.uuid}

Getting User Notifications

# Get all notifications for a user
notifications = await notification_client.get_user_notifications(
    db_engine=db,
    user_id="user_123",
    page=1,
)

# Get only unseen notifications
unseen_notifications = await notification_client.get_user_notifications(
    db_engine=db,
    user_id="user_123",
    page=1,
    seen=False,
)

# Get unseen count
unseen_count = await notification_client.get_unseen_count(
    db_engine=db,
    user_id="user_123",
)

Marking Notifications as Seen

# Mark a specific notification as seen
notification = await notification_client.mark_as_seen(
    db_engine=db,
    notification_uuid="not_abc123",
)

Creating Notifications Without Email

# Create notification without sending email
notification = await notification_client.send_notification(
    db_engine=db,
    name="system_alert",
    user_id="user_123",
    user_email="[e-mail]",
    claims=request.state.claims,  # Required: user claims for ownership verification
    metadata={"alert_type": "info"},
    send_email=False,  # Don't send email
)

API Endpoints

Get My Notifications

GET /notifications/me?page=1&seen=false

Query Parameters:

• page (int): Page number (default: 1)
• seen (bool, optional): Filter by seen status

Response:

{
  "items": [
    {
      "uuid": "not_abc123",
      "name": "welcome",
      "user_id": "user_123",
      "user_email": "[e-mail]",
      "metadata": {},
      "link": "https://example.com/dashboard",
      "seen": false,
      "created_at": "2024-01-01T00:00:00Z",
      "updated_at": "2024-01-01T00:00:00Z",
      "deleted": false
    }
  ],
  "total": 1,
  "page": 1,
  "pages": 1,
  "size": 1
}

Get Unseen Count

GET /notifications/count/unseen

Response:

{
  "count": 5
}

Get My Notification (Auto-marks as Seen)

GET /notifications/me/{notification_id}

Note: This endpoint automatically marks the notification as seen: true when accessed. Only returns notifications belonging to the current user.

Response:

{
  "uuid": "not_abc123",
  "name": "welcome",
  "user_id": "user_123",
  "user_email": "[e-mail]",
  "metadata": {},
  "link": "https://example.com/dashboard",
  "link_name": "Go to Dashboard",
  "seen": true,
  "created_at": "2024-01-01T00:00:00Z",
  "updated_at": "2024-01-01T00:00:00Z",
  "deleted": false
}

Model Structure

The NotificationModel includes:

• uuid: Unique identifier (short UUID with "not" prefix)
• name: Name/type of the notification
• metadata: Additional metadata (dict)
• user_id: ID of the user who should receive the notification
• user_email: Email address of the user
• link: Call to action link (optional)
• link_name: Text to display on the call to action button/link in the email (optional)
• seen: Whether the notification has been seen (default: False)
• created_at: Creation timestamp
• updated_at: Last update timestamp
• deleted: Soft delete flag

Ownership and Permissions

The plugin uses FastAPI SDK's ownership rules to ensure users can only access their own notifications. The ownership is based on the user_id claim in the JWT token.

Required permissions:

• notification:create - Create notifications
• notification:read - Read notifications
• notification:update - Update notifications
• notification:delete - Delete notifications

Configuration

Custom Router Configuration

from fastapi_plugin_notification.app import create_notification_router

# Create router with custom prefix and tags
notification_router = create_notification_router(
    prefix="/notifications",
    tags=["notifications", "alerts"],
    get_db=lambda: engine,
)

Note: All routes are user-specific and automatically filter by the current user's ID from JWT claims.

Custom Template Variables

The notification client automatically includes:

• notification_name: The name of the notification
• user_email: The user's email
• link: The notification link
• link_name: The text for the call to action button/link
• All metadata fields
• All template_variables fields

These are merged and passed to the Postmark template. You can use link_name in your email template to display custom button text for the call to action link.

Error Handling

The plugin handles errors gracefully:

• If email sending fails, the notification is still stored in the database
• Errors are logged but don't interrupt the notification creation process
• Database errors are propagated as HTTP exceptions

Development

Setup

1. Clone the repository
2. Install dependencies:

uv sync

Running Tests

pytest

License

MIT License - see LICENSE file for details