fastapi-plugin-notification 0.1.0
Notification plugin for FastAPI projects using FastAPI SDK and Postmark provider.
Requires Python: >=3.12

# 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:

```bash
uv add fastapi-plugin-notification
```

Or using pip:

```bash
pip install fastapi-plugin-notification
```

## Quick Start

### 1. Initialize the Notification Client

```python
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="noreply@example.com",
    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

```python
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

```python
@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

```python
# 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

```python
# 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

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

### Creating Notifications Without Email

```python
# Create notification without sending email
notification = await notification_client.send_notification(
    db_engine=db,
    name="system_alert",
    user_id="user_123",
    user_email="user@example.com",
    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

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

**Query Parameters:**
- `page` (int): Page number (default: 1)
- `seen` (bool, optional): Filter by seen status

**Response:**
```json
{
  "items": [
    {
      "uuid": "not_abc123",
      "name": "welcome",
      "user_id": "user_123",
      "user_email": "user@example.com",
      "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

```http
GET /notifications/count/unseen
```

**Response:**
```json
{
  "count": 5
}
```

### Get My Notification (Auto-marks as Seen)

```http
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:**
```json
{
  "uuid": "not_abc123",
  "name": "welcome",
  "user_id": "user_123",
  "user_email": "user@example.com",
  "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

```python
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:
```bash
uv sync
```

### Running Tests

```bash
pytest
```

## License

MIT License - see LICENSE file for details