☰
Navigation :

Using the Notification Service with the Atoti Python API

Overview

You can build and send notifications using the Atoti Notification Service with the Atoti Python API. Learn how to trigger the Atoti Python API’s REST services to publish notifications.

Steps

  1. Download the notification-service jar from Artifactory.
  2. Download the bas-extension-notification package from Artifactory.
  3. Build the bas-extension-notification package using pnpm build.
  4. Start your session with the notification-service jar and the bas-extension-notification package.

Sample Notebook

The following notebook demonstrates how to use the Atoti Notification Service with the Atoti Python API.

import os
from pathlib import WindowsPath
import sys
import requests
import json
import atoti as tt

note

Before starting the session, ensure you have an environment variable named ATOTI_LICENSE that has your Base64 encoded license. This is the same value as the ACTIVEPIVOT_LICENSE.

1. Start the session

This is the only configuration required to install the notification service in the Python session. Ensure you include the following:

  • extra_jars with the path to the notification-service jar
  • app_extensions with the path to the bas-extension-notification package
session = tt.Session.start({
    'extra_jars': [ 
        '../target/notification-service-1.0.1.jar'
    ],
    'app_extensions': {
        # Please change this
        '@activeviam/bas-extension-notification': WindowsPath('C:/path/to/bas-extension-notification/dist')
    }
})

Perform regular session activity until you are ready to send the notification.

sales_table = session.read_csv("data/sales.csv", keys={"Sale ID"}, table_name="Sales")
cube = session.create_cube(sales_table)

2. Build the notification

# this is the endpoint to publish a single notification
url = str(session.link) + "/notification-service/rest/v1/notification"

payload = {
    'id': 0,
    'title': 'Data loaded',
    'description': 'Sales data has been loaded',
    "severity": "LOW",
    'classifiers': [ "startup" ],
    'defaultActions': {
        'markRead' : {
            'actionName': 'MARK_READ',
            'confirmationMessage': 'Marked notification as read.',
            'errorMessage': 'Failed to mark notification as read.'
        },
        'markUnread' : {
            'actionName': 'MARK_UNREAD',
            'confirmationMessage': 'Marked notification as unread.',
            'errorMessage': 'Failed to mark notification as unread.'
        },
        'delete' : {
            'actionName': 'DELETE',
            'confirmationMessage': 'Deleted notification.',
            'errorMessage': 'Failed to delete notification.'
        },
    },
    'applicationName': 'Atoti',
    'status': ''
}
headers = {'Content-Type': 'application/json'}

3. Publish the notification

try:
    response = requests.post(url, json=payload, headers=headers)
    response.raise_for_status()  # Raise HTTPError for bad responses (4xx or 5xx)

    print("Request successful!")
    print("Status code:", response.status_code)
    print("Response body:", response.text)

except requests.exceptions.RequestException as e:
    print(f"Request failed: {e}")
except json.JSONDecodeError:
    print("Failed to decode JSON response")

Stop execution so you can navigate to the UI, publish the next notification and see it being published in real time.

stop

payload = {
    'id': 1,
    'title': 'Important reminder',
    'description': "Don't forget to keep selling Atoti!",
    "severity": "HIGH",
    'classifiers': [ 'revenue' ],
    'defaultActions': {
        'markRead' : {
            'actionName': 'MARK_READ',
            'confirmationMessage': 'Marked notification as read.',
            'errorMessage': 'Failed to mark notification as read.'
        },
        'markUnread' : {
            'actionName': 'MARK_UNREAD',
            'confirmationMessage': 'Marked notification as unread.',
            'errorMessage': 'Failed to mark notification as unread.'
        },
        'delete' : {
            'actionName': 'DELETE',
            'confirmationMessage': 'Deleted notification.',
            'errorMessage': 'Failed to delete notification.'
        },
    },
    'applicationName': 'Atoti',
    'status': ''
}

try:
    response = requests.post(url, json=payload, headers=headers)
    response.raise_for_status()  # Raise HTTPError for bad responses (4xx or 5xx)

    print("Request successful!")
    print("Status code:", response.status_code)
    print("Response body:", response.text)

except requests.exceptions.RequestException as e:
    print(f"Request failed: {e}")
except json.JSONDecodeError:
    print("Failed to decode JSON response")