feat: send notifications when an alert is received (#272)

* feat: send notifications when an alert is received

- send and log a notification message via email (or another type, to be implemented) to the recipients subscribed to a group
- add tables and routes: notifications and recipients
- allow python >= 3.8, <4 ; update poetry.lock

* add subject ; not working

* fix: pydantic validators and code quality

* fix: reinstate postgresql extra for dependency `databases`

* test: add library stubs to mypy test in CI

* fix: headers

* fix: quack suggestions

* telegram instead of email

* refactor: change return value of create_event_if_inexistant

* refactor: models, send_telegram_msg, tests

* fix: header

* fix header ; add test ; CI: fix ruff version, add telegram secrets

* fix: remove sleep and untested line

* refactor send_telegram_msg to remove untested line

* doc: improve / fix docstring

---------

Co-authored-by: Bruno Lenzi <[email protected]>

.github/workflows/style.yml

@@ -22,7 +22,7 @@ jobs:
           architecture: x64
       - name: Run ruff
         run: |
-          pip install ruff
+          pip install ruff==0.0.278
           ruff --version
           ruff check --diff .
 

.github/workflows/tests.yml

@@ -29,6 +29,8 @@ jobs:
           S3_ACCESS_KEY: ${{ secrets.S3_ACCESS_KEY }}
           S3_SECRET_KEY: ${{ secrets.S3_SECRET_KEY }}
           S3_REGION: ${{ secrets.S3_REGION }}
+          TELEGRAM_TOKEN: ${{ secrets.TELEGRAM_TOKEN }}
+          TELEGRAM_TEST_CHAT_ID: ${{ secrets.TELEGRAM_TEST_CHAT_ID }}
         run: |
           docker build src/. -t pyroapi:python3.8-alpine3.10
           docker-compose -f docker-compose-dev.yml up -d --build

README.md

@@ -117,6 +117,7 @@ Optionally, the following information can be added:
 - `SENTRY_DSN`: the URL of the [Sentry](https://sentry.io/) project, which monitors back-end errors and report them back.
 - `SENTRY_SERVER_NAME`: the server tag to apply to events.
 - `BUCKET_MEDIA_FOLDER`: the optional subfolder to put the media files in
+- `TELEGRAM_TOKEN`: to send notifications via telegram for a new alert (once per event)
 
 So your `.env` file should look like something similar to:
 ```

docker-compose-dev.yml

@@ -25,6 +25,8 @@ services:
       - S3_ACCESS_KEY=${S3_ACCESS_KEY}
       - S3_SECRET_KEY=${S3_SECRET_KEY}
       - S3_REGION=${S3_REGION}
+      - TELEGRAM_TOKEN=${TELEGRAM_TOKEN}
+      - TELEGRAM_TEST_CHAT_ID=${TELEGRAM_TEST_CHAT_ID}
     depends_on:
       - db
   db:

docker-compose.yml

@@ -24,6 +24,7 @@ services:
       - S3_REGION=${S3_REGION}
       - SENTRY_DSN=${SENTRY_DSN}
       - SENTRY_SERVER_NAME=${SENTRY_SERVER_NAME}
+      - TELEGRAM_TOKEN=${TELEGRAM_TOKEN}
     depends_on:
       - db
   db:

src/app/api/crud/alerts.py

@@ -4,7 +4,7 @@
 # See LICENSE or go to <https://opensource.org/licenses/Apache-2.0> for full license details.
 
 from datetime import datetime, timedelta
-from typing import Optional
+from typing import Optional, Tuple
 
 from sqlalchemy import and_
 
@@ -30,14 +30,19 @@ async def resolve_previous_alert(device_id: int) -> Optional[AlertOut]:
     return AlertOut(**entries[0]) if len(entries) > 0 else None
 
 
-async def create_event_if_inexistant(payload: AlertIn) -> int:
+async def create_event_if_inexistant(payload: AlertIn) -> Tuple[Optional[int], bool]:
+    """Return the id of the event to be associated with the alert and a boolean flag to tell if a new event was created
+
+    Args:
+        payload: alert object
+
+    Returns: tuple (int, bool) -> (event_id, new alert ?)
+
+    """
     # check whether there is an alert in the last 30 min by the same device
     previous_alert = await resolve_previous_alert(payload.device_id)
     if previous_alert is None:
         # Create an event & get the ID
         event = await create_event(EventIn(lat=payload.lat, lon=payload.lon, start_ts=datetime.utcnow()))
-        event_id = event["id"]
-    # Get event ref
-    else:
-        event_id = previous_alert.event_id
-    return event_id
+        return event["id"], True
+    return previous_alert.event_id, False

src/app/api/endpoints/alerts.py

@@ -4,6 +4,7 @@
 # See LICENSE or go to <https://opensource.org/licenses/Apache-2.0> for full license details.
 
 from functools import partial
+from string import Template
 from typing import List, cast
 
 from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Path, Security, status
@@ -13,10 +14,13 @@
 from app.api.crud.authorizations import check_group_read, is_admin_access
 from app.api.crud.groups import get_entity_group_id
 from app.api.deps import get_current_access, get_current_device, get_db
+from app.api.endpoints.devices import get_device
+from app.api.endpoints.notifications import send_notification
+from app.api.endpoints.recipients import fetch_recipients_for_group
 from app.api.external import post_request
 from app.db import alerts, events, media
 from app.models import Access, AccessType, Alert, Device, Event
-from app.schemas import AlertBase, AlertIn, AlertOut, DeviceOut
+from app.schemas import AlertBase, AlertIn, AlertOut, DeviceOut, NotificationIn, RecipientOut
 
 router = APIRouter()
 
@@ -33,6 +37,24 @@ async def alert_notification(payload: AlertOut):
     # Post the payload to each URL
     map(partial(post_request, payload=payload), webhook_urls)
 
+    # Send notification to the recipients of the same group as the device that issued the alert
+    device: DeviceOut = DeviceOut.parse_obj(await get_device(payload.device_id))
+    # N.B.: "or 0" is just for mypy, to convert Optional[int] -> int ; should never happen
+    for item in await fetch_recipients_for_group(await get_entity_group_id(alerts, payload.id) or 0):
+        recipient: RecipientOut = RecipientOut.parse_obj(item)
+        # Information to be added to subject and message: safe_substitute accepts fields that are not present
+        info = {
+            "alert_id": payload.id,
+            "event_id": payload.event_id,
+            "date": "?" if payload.created_at is None else payload.created_at.isoformat(sep=" ", timespec="seconds"),
+            "device_id": device.id,
+            "device_name": device.login,
+        }
+        subject: str = Template(recipient.subject_template).safe_substitute(**info)
+        message: str = Template(recipient.message_template).safe_substitute(**info)
+        notification = NotificationIn(alert_id=payload.id, recipient_id=recipient.id, subject=subject, message=message)
+        await send_notification(notification)
+
 
 @router.post("/", response_model=AlertOut, status_code=status.HTTP_201_CREATED, summary="Create a new alert")
 async def create_alert(
@@ -41,19 +63,21 @@ async def create_alert(
     _=Security(get_current_access, scopes=[AccessType.admin]),
 ):
     """
-    Creates a new alert based on the given information
+    Creates a new alert based on the given information and send a notification if it is the first alert of the event
 
     Below, click on "Schema" for more detailed information about arguments
     or "Example Value" to get a concrete idea of arguments
     """
     if payload.media_id is not None:
         await check_media_existence(payload.media_id)
 
+    new_event: bool = False
     if payload.event_id is None:
-        payload.event_id = await crud.alerts.create_event_if_inexistant(payload)
-    alert = await crud.create_entry(alerts, payload)
+        payload.event_id, new_event = await crud.alerts.create_event_if_inexistant(payload)
+    alert = AlertOut.parse_obj(await crud.create_entry(alerts, payload))
     # Send notification
-    background_tasks.add_task(alert_notification, alert)  # type: ignore[arg-type]
+    if new_event:
+        background_tasks.add_task(alert_notification, alert)
     return alert
 
 

src/app/api/endpoints/notifications.py

@@ -0,0 +1,64 @@
+# Copyright (C) 2020-2023, Pyronear.
+
+# This program is licensed under the Apache License 2.0.
+# See LICENSE or go to <https://opensource.org/licenses/Apache-2.0> for full license details.
+
+from typing import List
+
+from fastapi import APIRouter, HTTPException, Path, Security, status
+
+from app.api import crud
+from app.api.deps import get_current_access
+from app.api.endpoints.recipients import get_recipient
+from app.db import notifications
+from app.models import AccessType, NotificationType
+from app.schemas import NotificationIn, NotificationOut, RecipientOut
+from app.services import send_telegram_msg
+
+router = APIRouter(dependencies=[Security(get_current_access, scopes=[AccessType.admin])])
+
+
+@router.post(
+    "/", response_model=NotificationOut, status_code=status.HTTP_201_CREATED, summary="Send and log notification"
+)
+async def send_notification(payload: NotificationIn):
+    """
+    Send a notification to the recipients of the same group as the device that issued the alert; log notification to db
+
+    Below, click on "Schema" for more detailed information about arguments
+    or "Example Value" to get a concrete idea of arguments
+    """
+    recipient = RecipientOut(**(await get_recipient(recipient_id=payload.recipient_id)))
+    if recipient.notification_type == NotificationType.telegram:
+        send_telegram_msg(recipient.address, payload.message)
+    else:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid NotificationType, not treated")
+
+    notification: NotificationOut = NotificationOut(**(await crud.create_entry(notifications, payload)))
+    return notification
+
+
+@router.get(
+    "/{notification_id}/", response_model=NotificationOut, summary="Get information about a specific notification"
+)
+async def get_notification(notification_id: int = Path(..., gt=0)):
+    """
+    Retrieve information about the notification with the given notification_id
+    """
+    return await crud.get_entry(notifications, notification_id)
+
+
+@router.get("/", response_model=List[NotificationOut], summary="Get the list of all notifications")
+async def fetch_notifications():
+    """
+    Retrieves the list of all notifications and their information
+    """
+    return await crud.fetch_all(notifications)
+
+
+@router.delete("/{notification_id}/", response_model=NotificationOut, summary="Delete a specific notification")
+async def delete_notification(notification_id: int = Path(..., gt=0)):
+    """
+    Based on a notification_id, deletes the specified notification from the db
+    """
+    return await crud.delete_entry(notifications, notification_id)

src/app/api/endpoints/recipients.py

@@ -0,0 +1,74 @@
+# Copyright (C) 2020-2023, Pyronear.
+
+# This program is licensed under the Apache License 2.0.
+# See LICENSE or go to <https://opensource.org/licenses/Apache-2.0> for full license details.
+
+from typing import List
+
+from fastapi import APIRouter, Security, status
+from pydantic import PositiveInt
+
+from app.api import crud
+from app.api.deps import get_current_access
+from app.db import recipients
+from app.models import AccessType
+from app.schemas import RecipientIn, RecipientOut
+
+router = APIRouter(dependencies=[Security(get_current_access, scopes=[AccessType.admin])])
+
+
+@router.post("/", response_model=RecipientOut, status_code=status.HTTP_201_CREATED, summary="Create a new recipient")
+async def create_recipient(
+    payload: RecipientIn,
+):
+    """
+    Creates a new recipient based on the given information
+
+    Below, click on "Schema" for more detailed information about arguments
+    or "Example Value" to get a concrete idea of arguments
+    """
+    return await crud.create_entry(recipients, payload)
+
+
+@router.get("/{recipient_id}/", response_model=RecipientOut, summary="Get information about a specific recipient")
+async def get_recipient(recipient_id: PositiveInt):
+    """
+    Retrieve information about the recipient with the given recipient_id
+    """
+    return await crud.get_entry(recipients, recipient_id)
+
+
+@router.get("/", response_model=List[RecipientOut], summary="Get the list of all recipients")
+async def fetch_recipients():
+    """
+    Retrieves the list of all recipients and their information
+    """
+    return await crud.fetch_all(recipients)
+
+
+@router.get(
+    "/group-recipients/{group_id}/",
+    response_model=List[RecipientOut],
+    summary="Get the list of all recipients for the given group",
+)
+async def fetch_recipients_for_group(group_id: PositiveInt):
+    """
+    Retrieves the list of all recipients for the given group and their information
+    """
+    return await crud.fetch_all(recipients, {"group_id": group_id})
+
+
+@router.put("/{recipient_id}/", response_model=RecipientOut, summary="Update information about a specific recipient")
+async def update_recipient(payload: RecipientIn, recipient_id: PositiveInt):
+    """
+    Given its ID, updates information about the specified recipient
+    """
+    return await crud.update_entry(recipients, payload, recipient_id)
+
+
+@router.delete("/{recipient_id}/", response_model=RecipientOut, summary="Delete a specific recipient")
+async def delete_recipient(recipient_id: PositiveInt):
+    """
+    Based on a recipient_id, deletes the specified recipient
+    """
+    return await crud.delete_entry(recipients, recipient_id)

src/app/config.py

@@ -56,3 +56,6 @@
 # Sentry
 SENTRY_DSN: Optional[str] = os.getenv("SENTRY_DSN")
 SENTRY_SERVER_NAME: Optional[str] = os.getenv("SENTRY_SERVER_NAME")
+
+# Telegram
+TELEGRAM_TOKEN: Optional[str] = os.getenv("TELEGRAM_TOKEN")

src/app/db/tables.py

@@ -4,7 +4,20 @@
 # See LICENSE or go to <https://opensource.org/licenses/Apache-2.0> for full license details.
 
 
-from app.models import Access, Alert, Device, Event, Group, Installation, Media, Site, User, Webhook
+from app.models import (
+    Access,
+    Alert,
+    Device,
+    Event,
+    Group,
+    Installation,
+    Media,
+    Notification,
+    Recipient,
+    Site,
+    User,
+    Webhook,
+)
 
 from .base_class import Base
 
@@ -20,6 +33,8 @@
     "installations",
     "alerts",
     "webhooks",
+    "recipients",
+    "notifications",
 ]
 
 users = User.__table__
@@ -32,5 +47,7 @@
 installations = Installation.__table__
 alerts = Alert.__table__
 webhooks = Webhook.__table__
+recipients = Recipient.__table__
+notifications = Notification.__table__
 
 metadata = Base.metadata

src/app/main.py

@@ -21,6 +21,8 @@
     installations,
     login,
     media,
+    notifications,
+    recipients,
     sites,
     users,
     webhooks,
@@ -70,6 +72,8 @@ async def shutdown():
 app.include_router(alerts.router, prefix="/alerts", tags=["alerts"])
 app.include_router(accesses.router, prefix="/accesses", tags=["accesses"])
 app.include_router(webhooks.router, prefix="/webhooks", tags=["webhooks"])
+app.include_router(recipients.router, prefix="/recipients", tags=["recipients"])
+app.include_router(notifications.router, prefix="/notifications", tags=["notifications"])
 
 
 # Middleware

src/app/models/__init__.py

@@ -8,3 +8,5 @@
 from .site import *
 from .user import *
 from .webhook import *
+from .recipient import *
+from .notification import *

src/app/models/alert.py

@@ -28,6 +28,7 @@ class Alert(Base):
     device: RelationshipProperty = relationship("Device", back_populates="alerts")
     event: RelationshipProperty = relationship("Event", back_populates="alerts")
     media: RelationshipProperty = relationship("Media", back_populates="alerts")
+    notifications: RelationshipProperty = relationship("Notification", back_populates="alert")
 
     def __repr__(self):
         return f"<Alert(device_id='{self.device_id}', event_id='{self.event_id}', media_id='{self.media_id}'>"

src/app/models/group.py

@@ -20,6 +20,7 @@ class Group(Base):
 
     accesses: RelationshipProperty = relationship("Access", back_populates="group")
     sites: RelationshipProperty = relationship("Site", back_populates="group")
+    recipients: RelationshipProperty = relationship("Recipient", back_populates="group")
 
     def __repr__(self):
         return f"<Group(name='{self.name}')>"