A simple way to Send Notion Updates to Mattermost using Python

In the digital age, effective communication and organization are paramount. Notion and Mattermost are two technologies that have revolutionized the way teams collaborate and manage tasks. This article will delve into what these technologies are, their functionalities, and how to Send Notion Updates to Mattermost using Python.

There are a lot of integration websites like zapier, integrately, pabbly which provides a GUI based ways but joy of coding our own solution is always interesting and joyful thing to do.

Notion: A Brief Overview

Notion is an all-in-one workspace where you can write, plan, collaborate, and organize. It essentially provides components such as notes, databases, kanban boards, wikis, calendars, and reminders. Teams and individuals use Notion to create custom systems that match their style of work.

Mattermost: A Brief Overview

Mattermost is an open-source, self-hosted online chat service with file sharing, search, and integrations. It’s designed as an internal chat for organizations and companies, and mostly markets itself as an open-source alternative to Slack and Microsoft Teams.

Python: The Bridge Between Notion and Mattermost

To send Notion updates notifications to Mattermost using Python, we’ll need to use the Notion API, Mattermost Webhooks, and a Python script to tie everything together.

Step 1: Setting Up Notion API

First, you need to set up the Notion API. Notion’s API allows you to retrieve, update, delete, and create data in Notion. You’ll need to create an integration, which will provide you with a token that you’ll use to authenticate your requests.

Step 2: Setting Up Mattermost Webhooks

Next, set up a webhook in Mattermost. A webhook is a way for an app to provide other applications with real-time information. In Mattermost, incoming webhooks are a simple way to post messages from external sources into Mattermost channels.

Step 3: Writing the Script to Send Notion Updates to Mattermost using Python

Now, you can write a Python script that uses the Notion API to fetch updates and the Mattermost webhook to send these updates as notifications. You’ll need to use the notion_client, httpx, requests libraries in Python to make HTTP requests to both the Notion API and the Mattermost webhook. This can be easily installed using PIP.

The below script was build using Python 3.8.

import os
import time
from datetime import datetime, timedelta
from notion_client import Client
import requests
import httpx
from time import sleep

NOTION_API_KEY = "xxx"
MATTERMOST_WEBHOOK_URL = "yyy"

notion = Client(auth=NOTION_API_KEY)

def retry_request(func, *args, **kwargs):
    max_retries = 5
    backoff_factor = 2
    for retry_count in range(max_retries):
        try:
            return func(*args, **kwargs)
        except httpx.HTTPStatusError as e:
            if e.response.status_code == 502 and retry_count < max_retries - 1:
                sleep_time = backoff_factor ** retry_count
                print(f"Request failed with a 502 error. Retrying in {sleep_time} seconds...")
                sleep(sleep_time)
            else:
                raise

def list_databases():
    databases = []
    cursor = None
    while True:
        response = retry_request(notion.search, filter={"property": "object", "value": "database"}, start_cursor=cursor)
        databases.extend(response["results"])
        cursor = response.get("next_cursor")
        if not cursor:
            break
    return databases

def fetch_database_items(database_id):
    one_minute_ago = (datetime.utcnow() - timedelta(minutes=5)).isoformat() + "Z"
    pages = []
    cursor = None

    while True:
        result = retry_request(notion.databases.query, database_id, start_cursor=cursor)
        pages.extend(result["results"])
        cursor = result.get("next_cursor")
        if not cursor:
            break

    return [item for item in pages if item["last_edited_time"] > one_minute_ago]

def fetch_user(user_id):
    return retry_request(notion.users.retrieve, user_id)

def send_to_mattermost(item, user):
    title = None

    for property_key, property_value in item["properties"].items():
        if property_value["type"] == "title":
            title = property_value["title"][0]["plain_text"]
            break
        elif property_value["type"] == "rich_text":
            if property_value["rich_text"]:
                title = property_value["rich_text"][0]["plain_text"]
                break
            else:
                title = "No content"
                break

    if title is None:
        title = "Title not found"

    url = f"https://www.notion.so/{item['id'].replace('-', '')}"
    if user:
        message = f"**{title}** has been updated by {user['name']}.\nLink: {url}"
    else:
        message = f"**{title}** has been updated.\nLink: {url}"

    print(f"Sending message to Mattermost: {message}")

    response = requests.post(
        MATTERMOST_WEBHOOK_URL,
        json={"text": message}
    )
    if response.status_code != 200:
        raise ValueError(f"Request to Mattermost returned an error: {response.status_code}, {response.text}")


def fetch_last_editor(database_id, last_edited_time):
    blocks = retry_request(notion.blocks.children.list, database_id)
    for block in blocks["results"]:
        if block.get("last_edited_time") == last_edited_time:
            last_edited_by_id = block["last_edited_by"]["id"]
            return fetch_user(last_edited_by_id)
    return None



def main():
    databases = list_databases()
    print(f"Found {len(databases)} databases")

    while True:
        for database in databases:
            updated_items = fetch_database_items(database["id"])
            print(f"Found {len(updated_items)} updated items for database {database['title']}")
            for item in updated_items:
                last_edited_time = item["last_edited_time"]
                last_editor = fetch_last_editor(database["id"], last_edited_time)
                send_to_mattermost(item, last_editor)
        time.sleep(300)

if __name__ == "__main__":
    main()

Replace “xxx” and “yyy” with the corresponding notion API key and MatterMost webhook URL.

Save it as notion_mattermost_notifier.py

Step 4: Executing the Python Script

You can simple run the script in the same way you execute every python codes, depending on how its installed.

python notion_mattermost_notifier.py

OR

python3 notion_mattermost_notifier.py

The script runs every 5 minutes and check for notion updates. If any notion sheets are updated in last 5 minutes, it will send a notification

It can be run in background by appending & or by running it as a service.

Catches

Of course, yes. It is not a complete script, as Notion API is still not a complete solution. Below are some problems or catches which I couldn’t resolve yet

• All notion databases will need a field “Last Edited” which shows the time when it was edited last. The script checks that field to find the latest updates.
• Any update to the comments for the databases items will not be considered as an update. Only updates to status, body of the text etc. works

End Notes

By integrating Notion and Mattermost using Python, you can create a powerful system that keeps your team updated on all changes and updates in Notion. This not only improves communication within your team, but also enhances productivity by ensuring everyone is on the same page.

Any changes to the script will be updated here as well. Comments are always welcome.