Notifications

In DataJunction, notifications are crucial for keeping users informed about various activities, such as changes in the state of nodes. This document outlines how notifications are managed within DataJunction and how you can implement a custom notification solution using FastAPI’s dependency injection.

How Notifications are Used #

DataJunction uses notifications to alert users about significant events, such as updates to nodes or changes in their state. By handling notifications within the OSS project, DataJunction provides an opinionated take on where and when notifications should be sent.

Default Notification Implementation #

Out of the box, DataJunction includes a simple placeholder dependency that logs notifications. This implementation is designed to be straightforward and easily replaceable with a custom solution.

Placeholder Notification Implementation #

Here’s a brief look at the placeholder notification implementation:

import logging
from datajunction_server.database.history import History

_logger = logging.getLogger(__name__)

def get_notifier():
    """Returns a method for sending notifications for an event"""
    def notify(event: History):
        """Send a notification for an event"""
        _logger.debug(f"Sending notification for event %s", event)
    return notify

Custom Notification Implementation #

You can implement a custom notification system by using FastAPI’s dependency injection and injecting a get_notifier dependency. The custom notifier must handle the notify method, which processes the notification events.

Implementing a Custom Notifier #

To implement a custom notifier, create a function that processes notification events and use FastAPI’s dependency injection to inject your custom notifier.

Here’s an example of a custom notifier implementation:

from fastapi import Request

def custom_notify(event: History):
    """Custom logic to send notifications"""
    # Implement the logic to send notifications, e.g., via email or a messaging service
    ...

def get_custom_notifier(request: Request) -> callable:
    """Dependency for retrieving a custom notifier implementation"""
    # You can even add logic to choose between different notifiers based on request headers or other criteria
    return custom_notify

# Override the built-in get_notifier with your custom one
app.dependency_overrides[get_notifier] = get_custom_notifier

Example Usage #

Here’s how the get_notifier dependency is used within the application to send a notification:

notify = Depends(get_notifier())
event = History(
    id=1,
    entity_name="bar",
    entity_type=EntityType.NODE,
    activity_type=ActivityType.CREATE,
)
notify(event)

By customizing the get_notifier dependency, you can tailor the notification system to suit your specific needs, such as integrating with third-party services or implementing advanced notification configuration or logic.