Merge pull request #76 from GabrielSalla/add-monitor-variables

Add monitor variables

Author: GabrielSalla

docs/monitor.md

@@ -56,9 +56,10 @@ To create a monitor, import specific dependencies from `monitor_utils`. Availabl
 - [`ValueRule`](#value-rule)
 - [`PriorityLevels`](#priority-levels)
 
-**Functions**
+**Available functions and modules**
 - [`query`](#query)
 - [`read_file`](#read-file)
+- [`variables`](#variables)
 
 To get started with a simple monitor, use the following imports:
 
@@ -360,5 +361,38 @@ The function takes 2 parameters:
 content = read_file("search_query.sql")
 ```
 
+## Variables
+The `variables` module allows storing and retrieving variables that can be used across executions of the monitor. This is useful for maintaining state or configuration information.
+
+Available functions are:
+
+**`get_variable`**
+
+The function takes one parameter:
+- `name`: The name of the variable to retrieve.
+
+Return the value of a variable. If the variable does not exist, returns `None`.
+
+**`set_variable`**
+
+The function takes two parameters:
+- `name`: The name of the variable to set.
+- `value`: The value to assign to the variable.
+
+Sets the value of a variable. If the variable does not exist yet, it's created.
+
+Both functions must be called from functions defined in the monitor base code. If they're called from any other Python file, they will raise an error as they won't be able to identify the monitor that's calling it.
+
+```python
+from monitor_utils import variables
+
+async def search() -> list[IssueDataType] | None:
+    # Set a variable
+    await variables.set_variable("my_var", "some_value")
+
+async def update(issues_data: list[IssueDataType]) -> list[IssueDataType] | None:
+    value = await variables.get_variable("my_var")
+```
+
 # Registering
 After creating the monitor, the next step is to register it on Sentinela. Check the [Registering a monitor](monitor_registering.md) documentation for more information.

migrations/versions/202505232108_ce56a57ff560_create_variables_table.py

@@ -0,0 +1,41 @@
+"""create variables table
+
+Revision ID: ce56a57ff560
+Revises: 247390255aee
+Create Date: 2025-05-23 21:08:54.608578
+
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "ce56a57ff560"
+down_revision: Union[str, None] = "247390255aee"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "Variables",
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column("monitor_id", sa.Integer()),
+        sa.Column("name", sa.String()),
+        sa.Column("value", sa.String(), nullable=True),
+        sa.Column("updated_at", sa.DateTime(timezone=True)),
+
+        sa.ForeignKeyConstraint(("monitor_id",), ["Monitors.id"]),
+    )
+    op.create_index(
+        "ix_Variables_monitor_id_name",
+        "Variables",
+        ["monitor_id", "name"],
+        unique=True,
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_Variables_monitor_id_name")
+    op.drop_table("Variables")

src/components/monitors_loader/monitor_module_type.py

@@ -1,11 +1,16 @@
+from pathlib import Path
 from typing import Protocol, TypedDict
 
 from data_models.monitor_options import AlertOptions, IssueOptions, MonitorOptions, ReactionOptions
 from notifications import BaseNotification
 
 
 class MonitorModule(Protocol):
-    """Class that represents a base monitor module structure to have a better code completion"""
+    """Class that represents a base monitor module structure"""
+
+    SENTINELA_MONITOR_ID: int
+    SENTINELA_MONITOR_NAME: str
+    SENTINELA_MONITOR_PATH: Path
 
     monitor_options: MonitorOptions
     issue_options: IssueOptions

src/components/monitors_loader/monitors_loader.py

@@ -179,8 +179,15 @@ async def _register_monitors() -> None:
         await _register_monitors_from_path(configs.sample_monitors_path)
 
 
-def _configure_monitor(monitor_module: MonitorModule) -> None:
+def _configure_monitor(
+    monitor_module: MonitorModule, monitor_id: int, monitor_name: str, monitor_path: Path
+) -> None:
     """Make the necessary configurations to the monitor's attributes"""
+    # Set the monitor's identification attributes
+    monitor_module.SENTINELA_MONITOR_ID = monitor_id
+    monitor_module.SENTINELA_MONITOR_NAME = monitor_name
+    monitor_module.SENTINELA_MONITOR_PATH = monitor_path
+
     # Add an empty reaction option if it was not configured
     if getattr(monitor_module, "reaction_options", None) is None:
         monitor_module.reaction_options = ReactionOptions()
@@ -274,7 +281,7 @@ async def _load_monitors(last_load_time: datetime | None) -> None:
 
             monitor_path = monitors_paths[monitor.id]
             monitor_module = cast(MonitorModule, module_loader.load_module_from_file(monitor_path))
-            _configure_monitor(monitor_module)
+            _configure_monitor(monitor_module, monitor.id, monitor.name, monitor_path)
 
             registry.add_monitor(monitor.id, monitor.name, monitor_module)
 

src/models/__init__.py

@@ -5,6 +5,7 @@
 from .monitor import Monitor
 from .notification import Notification, NotificationStatus
 from .utils.priority import AlertPriority
+from .variable import Variable
 
 __all__ = [
     "Alert",
@@ -17,4 +18,5 @@
     "Monitor",
     "Notification",
     "NotificationStatus",
+    "Variable",
 ]

src/models/variable.py

@@ -0,0 +1,24 @@
+from datetime import datetime
+
+from sqlalchemy import DateTime, ForeignKey, Integer, String
+from sqlalchemy.orm import Mapped, mapped_column
+
+from utils.time import now
+
+from .base import Base
+
+
+class Variable(Base):
+    __tablename__ = "Variables"
+
+    id: Mapped[int] = mapped_column(Integer(), primary_key=True)
+    monitor_id: Mapped[int] = mapped_column(ForeignKey("Monitors.id"))
+    name: Mapped[str] = mapped_column(String())
+    value: Mapped[str | None] = mapped_column(String(), nullable=True)
+    updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), insert_default=now)
+
+    async def set(self, value: str | None) -> None:
+        """Set the variable value"""
+        self.value = value
+        self.updated_at = now()
+        await self.save()

src/monitor_utils/__init__.py

@@ -16,6 +16,7 @@
 from databases import query
 from models.utils.priority import AlertPriority
 
+from . import variables
 from .read_file import read_file
 
 __all__ = [
@@ -31,4 +32,5 @@
     "ReactionOptions",
     "read_file",
     "ValueRule",
+    "variables",
 ]

src/monitor_utils/variables.py

@@ -0,0 +1,40 @@
+from types import ModuleType
+from typing import cast
+
+from models import Variable
+from utils.stack import get_caller
+
+"""The functions 'set_variable' and 'get_variable' must be called from inside a monitor to be able
+to identify the monitor ID. If called outside a monitor, they will raise an error."""
+
+
+def _get_monitor_id(monitor_module: ModuleType) -> int:
+    """Get the monitor ID from the monitor module"""
+    monitor_id = getattr(monitor_module, "SENTINELA_MONITOR_ID", None)
+    if monitor_id is None:
+        raise ValueError("Function called outside a monitor or the monitor was not loaded properly")
+
+    return cast(int, monitor_id)
+
+
+async def set_variable(name: str, value: str | None) -> None:
+    """Set a variable for the monitor. Must be called from inside a monitor or an error will be
+    raised."""
+    monitor_module, _ = get_caller()
+    monitor_id = _get_monitor_id(monitor_module)
+
+    variable = await Variable.get_or_create(monitor_id=monitor_id, name=name)
+    await variable.set(value)
+
+
+async def get_variable(name: str) -> str | None:
+    """Get a variable for the monitor, or None if it does not exist. Must be called from inside a
+    monitor or an error will be raised."""
+    monitor_module, _ = get_caller()
+    monitor_id = _get_monitor_id(monitor_module)
+
+    variable = await Variable.get(Variable.monitor_id == monitor_id, Variable.name == name)
+    if variable is None:
+        return None
+
+    return variable.value

src/utils/stack.py

@@ -0,0 +1,19 @@
+import inspect
+from types import ModuleType
+
+
+def get_caller(previous: int = 0) -> tuple[ModuleType, str]:
+    """Get the module of the caller function"""
+    stack_position = 2 + previous
+
+    stack = inspect.stack()
+
+    if len(stack) < stack_position:
+        raise IndexError(f"Could not access position {stack_position} in the stack")
+
+    caller_frame = stack[stack_position][0]
+    module = inspect.getmodule(caller_frame)
+    if module is None:
+        raise ValueError("Could not determine caller module")
+
+    return module, caller_frame.f_code.co_name

tests/components/monitors_loader/test_monitors_loader.py

@@ -496,20 +496,39 @@ async def test_register_monitors_no_sample_monitors(monkeypatch, clear_database)
             assert code_modules_dict[monitor.id].additional_files == {}
 
 
-async def test_configure_monitor(monkeypatch, sample_monitor: Monitor):
-    """'_configure_monitor' should populate the 'reaction_options' and 'notification_options' with
-    the default values"""
+@pytest.mark.parametrize(
+    "monitor_id, monitor_name, monitor_path",
+    [
+        (11, "abc", Path("/path/to/monitor")),
+        (22, "def", Path("/other_path/to/monitor2")),
+        (33, "ghi", Path("/one_more_path/monitor3")),
+    ],
+)
+async def test_configure_monitor(
+    monkeypatch, sample_monitor: Monitor, monitor_id, monitor_name, monitor_path
+):
+    """'_configure_monitor' should populate the monitor identification attributes,
+    'reaction_options' and 'notification_options' with the default values"""
     monitor_module = sample_monitor.code
     monkeypatch.setattr(monitor_module, "reaction_options", None, raising=False)
     monkeypatch.setattr(monitor_module, "notification_options", None, raising=False)
+    monkeypatch.setattr(monitor_module, "SENTINELA_MONITOR_ID", None)
+    monkeypatch.setattr(monitor_module, "SENTINELA_MONITOR_NAME", None)
+    monkeypatch.setattr(monitor_module, "SENTINELA_MONITOR_PATH", None)
 
     assert getattr(monitor_module, "reaction_options", None) is None
     assert getattr(monitor_module, "notification_options", None) is None
+    assert getattr(monitor_module, "SENTINELA_MONITOR_ID", None) is None
+    assert getattr(monitor_module, "SENTINELA_MONITOR_NAME", None) is None
+    assert getattr(monitor_module, "SENTINELA_MONITOR_PATH", None) is None
 
-    monitors_loader._configure_monitor(monitor_module)
+    monitors_loader._configure_monitor(monitor_module, monitor_id, monitor_name, monitor_path)
 
     assert isinstance(monitor_module.reaction_options, ReactionOptions)
     assert monitor_module.notification_options == []
+    assert monitor_module.SENTINELA_MONITOR_ID == monitor_id
+    assert monitor_module.SENTINELA_MONITOR_NAME == monitor_name
+    assert monitor_module.SENTINELA_MONITOR_PATH == monitor_path
 
 
 async def test_configure_monitor_notifications_setup(monkeypatch, sample_monitor: Monitor):
@@ -538,7 +557,7 @@ def reactions_list(self):
 
     monkeypatch.setattr(monitor_module, "notification_options", [MockNotification()], raising=False)
 
-    monitors_loader._configure_monitor(monitor_module)
+    monitors_loader._configure_monitor(monitor_module, 1, "monitor_1", Path(""))
 
     assert monitor_module.reaction_options.alert_updated == [do_something, do_nothing]
     assert monitor_module.reaction_options.alert_solved == [do_nothing, "do_nothing"]