event bus: initial implementation

Author: Pablo Panero

invenio_records_resources/config.py

@@ -1,6 +1,6 @@
 # -*- coding: utf-8 -*-
 #
-# Copyright (C) 2020 CERN.
+# Copyright (C) 2020-2022 CERN.
 # Copyright (C) 2020 Northwestern University.
 #
 # Invenio-Records-Resources is free software; you can redistribute it and/or
@@ -16,3 +16,7 @@
 SITE_UI_URL = "https://127.0.0.1:5000"
 
 SITE_API_URL = "https://127.0.0.1:5000/api"
+
+RECORDS_RESOURCES_EVENTS_HANDLERS = {}
+
+RECORDS_RESOURCES_EVENTS_QUEUE = "events"

invenio_records_resources/services/events/__init__.py

@@ -0,0 +1,19 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (C) 2022 CERN.
+#
+# Invenio-Records-Resources is free software; you can redistribute it and/or
+# modify it under the terms of the MIT License; see LICENSE file for more
+# details.
+
+"""Module for event driven actions support."""
+
+from .bus import EventBus
+from .events import Event
+from .handlers import EventHandler
+
+__all__ = (
+    "Event",
+    "EventHandler",
+    "EventBus"
+)

invenio_records_resources/services/events/bus.py

@@ -0,0 +1,38 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (C) 2022 CERN.
+#
+# Invenio-Records-Resources is free software; you can redistribute it and/or
+# modify it under the terms of the MIT License; see LICENSE file for more
+# details.
+
+"""Events bus module."""
+
+from pickle import dumps, loads
+
+from flask import current_app
+from invenio_queues.proxies import current_queues
+
+
+class EventBus:
+    """Event bus."""
+
+    def __init__(self, queue_name=None):
+        """Constructor."""
+        self._queue_name = queue_name or \
+            current_app.config["RECORDS_RESOURCES_EVENTS_QUEUE"]
+        self._queue = None
+
+        for name, queue in current_queues.queues.items():
+            if name == self._queue_name:
+                self._queue = queue
+                break
+
+    def publish(self, event):
+        """Publish an event to the bus queue."""
+        return self._queue.publish([dumps(event)])
+
+    def consume(self):
+        """Consume an event from the bus queue."""
+        for event in self._queue.consume():  # consume() returns a generator
+            yield loads(event)

invenio_records_resources/services/events/events.py

@@ -0,0 +1,37 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (C) 2022 CERN.
+#
+# Invenio-Records-Resources is free software; you can redistribute it and/or
+# modify it under the terms of the MIT License; see LICENSE file for more
+# details.
+
+"""Events module."""
+
+from dataclasses import dataclass
+from datetime import datetime
+from typing import ClassVar
+
+
+@dataclass
+class Event:
+    """Base event."""
+
+    created: datetime
+    type: str
+    action: str
+
+
+@dataclass
+class RecordEvent(Event):
+    """Record related events."""
+
+    recid: str
+    type: ClassVar[str] = "RECORD"
+
+
+@dataclass
+class RecordCreatedEvent(RecordEvent):
+    """Record related events."""
+
+    action: ClassVar[str] = "PUBLISHED"

invenio_records_resources/services/events/handlers.py

@@ -0,0 +1,20 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (C) 2022 CERN.
+#
+# Invenio-Records-Resources is free software; you can redistribute it and/or
+# modify it under the terms of the MIT License; see LICENSE file for more
+# details.
+
+"""Event handlers module."""
+
+from abc import ABC, abstractmethod
+
+
+class EventHandler(ABC):
+    """Abstract event handler class."""
+
+    @abstractmethod
+    def handle(self, event):
+        """Handle an event."""
+        pass

invenio_records_resources/services/uow.py

@@ -106,6 +106,8 @@ def on_commit(self, uow):
 
 from invenio_db import db
 
+from .events import EventBus
+
 
 #
 # Unit of work operations
@@ -199,6 +201,21 @@ def on_post_commit(self, uow):
         self._celery_task.delay(*self._args, **self._kwargs)
 
 
+class EventOp(Operation):
+    """A task to send an event.
+
+    All events will be sent after the commit phase.
+    """
+
+    def __init__(self, event, *args, **kwargs):
+        """Constructor."""
+        self._event = event
+
+    def on_post_commit(self, uow):
+        """Publish the event to the bus."""
+        uow._event_bus.publish(self._event)
+
+
 #
 # Unit of work context manager
 #
@@ -215,6 +232,7 @@ def __init__(self, session=None):
         """Initialize unit of work context."""
         self._session = session or db.session
         self._operations = []
+        self._event_bus = EventBus()
         self._dirty = False
 
     def __enter__(self):

run-tests.sh

@@ -31,7 +31,7 @@ trap cleanup EXIT
 
 python -m check_manifest --ignore ".*-requirements.txt"
 python -m sphinx.cmd.build -qnNW docs docs/_build/html
-eval "$(docker-services-cli up --db ${DB:-postgresql} --search ${SEARCH:-elasticsearch} --cache ${CACHE:-redis} --env)"
+eval "$(docker-services-cli up --db ${DB:-postgresql} --search ${SEARCH:-elasticsearch} --cache ${CACHE:-redis} --mq ${MQ:-rabbitmq} --env)"
 python -m pytest $@
 tests_exit_code=$?
 python -m sphinx.cmd.build -qnNW -b doctest docs docs/_build/doctest

setup.py

@@ -67,6 +67,7 @@
     "invenio-indexer>=1.2.1",
     "invenio-jsonschemas>=1.1.3",
     "invenio-pidstore>=1.2.2",
+    "invenio-queues>=1.0.0a4",
     "invenio-records-permissions>=0.13.0,<0.14.0",
     "invenio-records>=1.6.0",
     "luqum>=0.11.0",

tests/services/events/conftest.py

@@ -0,0 +1,37 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (C) 2022 CERN.
+#
+# Invenio-Records-Resources is free software; you can redistribute it and/or
+# modify it under the terms of the MIT License; see LICENSE file for more
+# details.
+
+"""Pytest configuration.
+
+See https://pytest-invenio.readthedocs.io/ for documentation on which test
+fixtures are available.
+"""
+
+import pytest
+from kombu import Exchange
+
+
+@pytest.fixture(scope="module")
+def app_config(app_config):
+    """Application configuration."""
+    queue_name = "test-events"
+    app_config["RECORDS_RESOURCES_EVENT_QUEUE"] = queue_name
+
+    exchange = Exchange(
+        queue=queue_name,
+        type="direct",
+        delivery_mode="persistent",  # in-memory and disk
+    )
+
+    app_config["QUEUES_DEFINITIONS"] = [
+        {"name": queue_name, "exchange": exchange}
+    ]
+
+    app_config["CELERY_ACCEPT_CONTENT"] = ["json", "msgpack", "yaml", "pickle"]
+
+    return app_config

tests/services/events/test_bus.py

@@ -0,0 +1,25 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (C) 2022 CERN.
+#
+# Invenio-Records-Resources is free software; you can redistribute it and/or
+# modify it under the terms of the MIT License; see LICENSE file for more
+# details.
+
+"""Event bus test."""
+
+from datetime import datetime
+from time import sleep
+
+from invenio_records_resources.services.events import EventBus
+from invenio_records_resources.services.events.events import RecordCreatedEvent
+
+
+def test_bus_publish_consume(app):
+    bus = EventBus("test-events")
+    event = RecordCreatedEvent(created=datetime.now(), recid="12345-abcde")
+
+    bus.publish(event)
+    sleep(10)
+    consumed_event = bus.consume()
+    assert event == next(consumed_event)