API - Added notifications API endpoints (#3103)

changedetectionio/api/Notifications.py

@@ -0,0 +1,145 @@
+from flask_expects_json import expects_json
+from flask_restful import Resource
+from . import auth
+from flask_restful import abort, Resource
+from flask import request
+from . import auth
+from . import schema_create_notification_urls, schema_delete_notification_urls
+
+class Notifications(Resource):
+    def __init__(self, **kwargs):
+        # datastore is a black box dependency
+        self.datastore = kwargs['datastore']
+
+    @auth.check_token
+    def get(self):
+        """
+        @api {get} /api/v1/notifications Return Notification URL List
+        @apiDescription Return the Notification URL List from the configuration
+        @apiExample {curl} Example usage:
+            curl http://localhost:5000/api/v1/notifications -H"x-api-key:813031b16330fe25e3780cf0325daa45"
+            HTTP/1.0 200
+            {
+                'notification_urls': ["notification-urls-list"]
+            }
+        @apiName Get
+        @apiGroup Notifications
+        """
+
+        notification_urls = self.datastore.data.get('settings', {}).get('application', {}).get('notification_urls', [])        
+
+        return {
+                'notification_urls': notification_urls,
+               }, 200
+    
+    @auth.check_token
+    @expects_json(schema_create_notification_urls)
+    def post(self):
+        """
+        @api {post} /api/v1/notifications Create Notification URLs
+        @apiDescription Add one or more notification URLs from the configuration
+        @apiExample {curl} Example usage:
+            curl http://localhost:5000/api/v1/notifications/batch -H"x-api-key:813031b16330fe25e3780cf0325daa45" -H "Content-Type: application/json" -d '{"notification_urls": ["url1", "url2"]}'
+        @apiName CreateBatch
+        @apiGroup Notifications
+        @apiSuccess (201) {Object[]} notification_urls List of added notification URLs
+        @apiError (400) {String} Invalid input
+        """
+
+        json_data = request.get_json()
+        notification_urls = json_data.get("notification_urls", [])
+
+        from wtforms import ValidationError
+        try:
+            validate_notification_urls(notification_urls)
+        except ValidationError as e:
+            return str(e), 400
+
+        added_urls = []
+
+        for url in notification_urls:
+            clean_url = url.strip()
+            added_url = self.datastore.add_notification_url(clean_url)
+            if added_url:
+                added_urls.append(added_url)
+
+        if not added_urls:
+            return "No valid notification URLs were added", 400
+
+        return {'notification_urls': added_urls}, 201
+    
+    @auth.check_token
+    @expects_json(schema_create_notification_urls)
+    def put(self):
+        """
+        @api {put} /api/v1/notifications Replace Notification URLs
+        @apiDescription Replace all notification URLs with the provided list (can be empty)
+        @apiExample {curl} Example usage:
+            curl -X PUT http://localhost:5000/api/v1/notifications -H"x-api-key:813031b16330fe25e3780cf0325daa45" -H "Content-Type: application/json" -d '{"notification_urls": ["url1", "url2"]}'
+        @apiName Replace
+        @apiGroup Notifications
+        @apiSuccess (200) {Object[]} notification_urls List of current notification URLs
+        @apiError (400) {String} Invalid input
+        """
+        json_data = request.get_json()
+        notification_urls = json_data.get("notification_urls", [])
+
+        from wtforms import ValidationError
+        try:
+            validate_notification_urls(notification_urls)
+        except ValidationError as e:
+            return str(e), 400
+        
+        if not isinstance(notification_urls, list):
+            return "Invalid input format", 400
+
+        clean_urls = [url.strip() for url in notification_urls if isinstance(url, str)]
+        self.datastore.data['settings']['application']['notification_urls'] = clean_urls
+        self.datastore.needs_write = True
+
+        return {'notification_urls': clean_urls}, 200
+        
+    @auth.check_token
+    @expects_json(schema_delete_notification_urls)
+    def delete(self):
+        """
+        @api {delete} /api/v1/notifications Delete Notification URLs
+        @apiDescription Deletes one or more notification URLs from the configuration
+        @apiExample {curl} Example usage:
+            curl http://localhost:5000/api/v1/notifications -X DELETE -H"x-api-key:813031b16330fe25e3780cf0325daa45" -H "Content-Type: application/json" -d '{"notification_urls": ["url1", "url2"]}'
+        @apiParam {String[]} notification_urls The notification URLs to delete.
+        @apiName Delete
+        @apiGroup Notifications
+        @apiSuccess (204) {String} OK Deleted
+        @apiError (400) {String} No matching notification URLs found.
+        """
+
+        json_data = request.get_json()
+        urls_to_delete = json_data.get("notification_urls", [])
+        if not isinstance(urls_to_delete, list):
+            abort(400, message="Expected a list of notification URLs.")
+
+        notification_urls = self.datastore.data['settings']['application'].get('notification_urls', [])
+        deleted = []
+
+        for url in urls_to_delete:
+            clean_url = url.strip()
+            if clean_url in notification_urls:
+                notification_urls.remove(clean_url)
+                deleted.append(clean_url)
+
+        if not deleted:
+            abort(400, message="No matching notification URLs found.")
+
+        self.datastore.data['settings']['application']['notification_urls'] = notification_urls
+        self.datastore.needs_write = True
+
+        return 'OK', 204
+    
+def validate_notification_urls(notification_urls):
+    from changedetectionio.forms import ValidateAppRiseServers
+    validator = ValidateAppRiseServers()
+    class DummyForm: pass
+    dummy_form = DummyForm()
+    field = type("Field", (object,), {"data": notification_urls, "gettext": lambda self, x: x})()
+    validator(dummy_form, field)

changedetectionio/api/__init__.py

@@ -19,8 +19,15 @@ schema_create_tag['required'] = ['title']
 schema_update_tag = copy.deepcopy(schema_tag)
 schema_update_tag['additionalProperties'] = False
 
+schema_notification_urls = copy.deepcopy(schema)
+schema_create_notification_urls = copy.deepcopy(schema_notification_urls)
+schema_create_notification_urls['required'] = ['notification_urls']
+schema_delete_notification_urls = copy.deepcopy(schema_notification_urls)
+schema_delete_notification_urls['required'] = ['notification_urls']
+
 # Import all API resources
 from .Watch import Watch, WatchHistory, WatchSingleHistory, CreateWatch
 from .Tags import Tags, Tag
 from .Import import Import
 from .SystemInfo import SystemInfo
+from .Notifications import Notifications

changedetectionio/flask_app.py

@@ -33,7 +33,7 @@ from loguru import logger
 
 from changedetectionio import __version__
 from changedetectionio import queuedWatchMetaData
-from changedetectionio.api import Watch, WatchHistory, WatchSingleHistory, CreateWatch, Import, SystemInfo, Tag, Tags
+from changedetectionio.api import Watch, WatchHistory, WatchSingleHistory, CreateWatch, Import, SystemInfo, Tag, Tags, Notifications
 from changedetectionio.api.Search import Search
 from .time_handler import is_within_schedule
 
@@ -285,7 +285,8 @@ def changedetection_app(config=None, datastore_o=None):
     watch_api.add_resource(Search, '/api/v1/search',
                            resource_class_kwargs={'datastore': datastore})
 
-
+    watch_api.add_resource(Notifications, '/api/v1/notifications',
+                           resource_class_kwargs={'datastore': datastore})
 
     @login_manager.user_loader
     def user_loader(email):

changedetectionio/store.py

@@ -964,3 +964,25 @@ class ChangeDetectionStore:
                         f_d.write(zlib.compress(f_j.read()))
                         os.unlink(json_path)
 
+    def add_notification_url(self, notification_url):
+        
+        logger.debug(f">>> Adding new notification_url - '{notification_url}'")
+
+        notification_urls = self.data['settings']['application'].get('notification_urls', [])
+
+        if notification_url in notification_urls:
+            return notification_url
+
+        with self.lock:
+            notification_urls = self.__data['settings']['application'].get('notification_urls', [])
+
+            if notification_url in notification_urls:
+                return notification_url
+
+            # Append and update the datastore
+            notification_urls.append(notification_url)
+            self.__data['settings']['application']['notification_urls'] = notification_urls
+            self.needs_write = True
+
+        return notification_url
+

changedetectionio/tests/test_api_notifications.py

@@ -0,0 +1,108 @@
+#!/usr/bin/env python3
+
+from flask import url_for
+from .util import live_server_setup
+import json
+
+def test_api_notifications_crud(client, live_server):
+    live_server_setup(live_server)
+    api_key = live_server.app.config['DATASTORE'].data['settings']['application'].get('api_access_token')
+
+    # Confirm notifications are initially empty
+    res = client.get(
+        url_for("notifications"),
+        headers={'x-api-key': api_key}
+    )
+    assert res.status_code == 200
+    assert res.json == {"notification_urls": []}
+
+    # Add notification URLs
+    test_urls = ["posts://example.com/notify1", "posts://example.com/notify2"]
+    res = client.post(
+        url_for("notifications"),
+        data=json.dumps({"notification_urls": test_urls}),
+        headers={'content-type': 'application/json', 'x-api-key': api_key}
+    )
+    assert res.status_code == 201
+    for url in test_urls:
+        assert url in res.json["notification_urls"]
+
+    # Confirm the notification URLs were added
+    res = client.get(
+        url_for("notifications"),
+        headers={'x-api-key': api_key}
+    )
+    assert res.status_code == 200
+    for url in test_urls:
+        assert url in res.json["notification_urls"]
+
+    # Delete one notification URL
+    res = client.delete(
+        url_for("notifications"),
+        data=json.dumps({"notification_urls": [test_urls[0]]}),
+        headers={'content-type': 'application/json', 'x-api-key': api_key}
+    )
+    assert res.status_code == 204
+
+    # Confirm it was removed and the other remains
+    res = client.get(
+        url_for("notifications"),
+        headers={'x-api-key': api_key}
+    )
+    assert res.status_code == 200
+    assert test_urls[0] not in res.json["notification_urls"]
+    assert test_urls[1] in res.json["notification_urls"]
+
+    # Try deleting a non-existent URL
+    res = client.delete(
+        url_for("notifications"),
+        data=json.dumps({"notification_urls": ["posts://nonexistent.com"]}),
+        headers={'content-type': 'application/json', 'x-api-key': api_key}
+    )
+    assert res.status_code == 400
+
+    res = client.post(
+        url_for("notifications"),
+        data=json.dumps({"notification_urls": test_urls}),
+        headers={'content-type': 'application/json', 'x-api-key': api_key}
+    )
+    assert res.status_code == 201
+
+    # Replace with a new list
+    replacement_urls = ["posts://new.example.com"]
+    res = client.put(
+        url_for("notifications"),
+        data=json.dumps({"notification_urls": replacement_urls}),
+        headers={'content-type': 'application/json', 'x-api-key': api_key}
+    )
+    assert res.status_code == 200
+    assert res.json["notification_urls"] == replacement_urls
+
+    # Replace with an empty list
+    res = client.put(
+        url_for("notifications"),
+        data=json.dumps({"notification_urls": []}),
+        headers={'content-type': 'application/json', 'x-api-key': api_key}
+    )
+    assert res.status_code == 200
+    assert res.json["notification_urls"] == []
+
+    # Provide an invalid AppRise URL to trigger validation error
+    invalid_urls = ["ftp://not-app-rise"]
+    res = client.post(
+        url_for("notifications"),
+        data=json.dumps({"notification_urls": invalid_urls}),
+        headers={'content-type': 'application/json', 'x-api-key': api_key}
+    )
+    assert res.status_code == 400
+    assert "is not a valid AppRise URL." in res.data.decode()
+
+    res = client.put(
+        url_for("notifications"),
+        data=json.dumps({"notification_urls": invalid_urls}),
+        headers={'content-type': 'application/json', 'x-api-key': api_key}
+    )
+    assert res.status_code == 400
+    assert "is not a valid AppRise URL." in res.data.decode()
+
+