Notifications backend

pydatalab/src/pydatalab/config.py

@@ -232,6 +232,11 @@ class ServerConfig(BaseSettings):
         description="A dictionary containing SMTP settings for sending emails for account registration.",
     )
 
+    ENABLE_NOTIFICATIONS: bool = Field(
+        False,
+        description="Whether to enable in-app notifications and their API endpoints.",
+    )
+
     MAX_CONTENT_LENGTH: int = Field(
         10 * 1000**3,
         description=r"""Direct mapping to the equivalent Flask setting. In practice, limits the file size that can be uploaded.

pydatalab/src/pydatalab/feature_flags.py

@@ -7,7 +7,7 @@
 from pydatalab.config import CONFIG
 from pydatalab.logger import LOGGER
 
-__all__ = ("FEATURE_FLAGS", "check_feature_flags", "FeatureFlags")
+__all__ = ("FEATURE_FLAGS", "check_feature_flags", "FeatureFlags", "NotificationFeatures")
 
 
 class AuthMechanisms(BaseModel):
@@ -23,10 +23,15 @@ class AIIntegrations(BaseModel):
     anthropic: bool = False
 
 
+class NotificationFeatures(BaseModel):
+    enabled: bool = False
+
+
 class FeatureFlags(BaseModel):
     auth_mechanisms: AuthMechanisms = AuthMechanisms()
     ai_integrations: AIIntegrations = AIIntegrations()
     email_notifications: bool = False
+    notifications: NotificationFeatures = NotificationFeatures()
 
 
 FEATURE_FLAGS: FeatureFlags = FeatureFlags()
@@ -59,6 +64,8 @@ def check_feature_flags(app):
 
     """
 
+    FEATURE_FLAGS.notifications = NotificationFeatures(enabled=CONFIG.ENABLE_NOTIFICATIONS)
+
     if CONFIG.EMAIL_AUTH_SMTP_SETTINGS is None:
         LOGGER.warning(
             "No email auth SMTP settings provided, email registration will not be enabled."

pydatalab/src/pydatalab/main.py

@@ -226,6 +226,10 @@ def register_endpoints(app: Flask):
     versions = ["", f"v{major}", f"v{major}.{minor}", f"v{major}.{minor}.{patch}"]
 
     for bp in BLUEPRINTS:
+        if bp.name == "notifications" and not CONFIG.ENABLE_NOTIFICATIONS:
+            LOGGER.info("Skipping notification routes because notifications are disabled.")
+            continue
+
         for ver in versions:
             app.register_blueprint(
                 bp, url_prefix=f"{CONFIG.ROOT_PATH}{ver}", name=f"{ver}/{bp.name}"

pydatalab/src/pydatalab/models/__init__.py

@@ -4,6 +4,11 @@
 from pydatalab.models.collections import Collection
 from pydatalab.models.equipment import Equipment
 from pydatalab.models.files import File
+from pydatalab.models.notifications import (
+    Notification,
+    NotificationGrouping,
+    NotificationOccurrence,
+)
 from pydatalab.models.people import Person
 from pydatalab.models.samples import Sample
 from pydatalab.models.starting_materials import StartingMaterial
@@ -25,5 +30,8 @@
     "Collection",
     "Equipment",
     "ItemVersion",
+    "Notification",
+    "NotificationGrouping",
+    "NotificationOccurrence",
     "ITEM_MODELS",
 )

pydatalab/src/pydatalab/models/notifications.py

@@ -0,0 +1,116 @@
+from datetime import datetime, timezone
+from enum import Enum
+
+from pydantic import BaseModel, Field, root_validator
+
+from pydatalab.models.entries import Entry
+from pydatalab.models.utils import JSON_ENCODERS, PyObjectId
+
+
+class NotificationLevel(str, Enum):
+    LOW = "low"
+    NORMAL = "normal"
+    IMPORTANT = "important"
+    URGENT = "urgent"
+    CRITICAL = "critical"
+
+    @property
+    def priority(self) -> int:
+        return {
+            NotificationLevel.LOW: 10,
+            NotificationLevel.NORMAL: 20,
+            NotificationLevel.IMPORTANT: 30,
+            NotificationLevel.URGENT: 40,
+            NotificationLevel.CRITICAL: 50,
+        }[self]
+
+
+class NotificationGroupPolicy(str, Enum):
+    ONCE = "once"
+    WINDOW = "window"
+
+
+class NotificationGrouping(BaseModel):
+    """Rules for grouping repeated notification occurrences."""
+
+    key: str = Field(
+        ...,
+        min_length=1,
+        max_length=500,
+        description="Stable key used to group repeated notifications from the same thing.",
+    )
+    policy: NotificationGroupPolicy = Field(
+        NotificationGroupPolicy.WINDOW,
+        description="How repeated notification occurrences should be grouped.",
+    )
+    window_seconds: int | None = Field(
+        86400,
+        ge=1,
+        description=(
+            "Minimum interval before a grouped notification can create a new notification document."
+        ),
+    )
+    max_occurrences: int = Field(
+        100,
+        ge=1,
+        description="Maximum number of occurrences before a new notification document is created.",
+    )
+
+    @root_validator
+    def validate_grouping_policy(cls, values):
+        policy = NotificationGroupPolicy(values.get("policy"))
+        if policy == NotificationGroupPolicy.WINDOW and values.get("window_seconds") is None:
+            raise ValueError("window_seconds must be provided for window grouping.")
+        if policy == NotificationGroupPolicy.ONCE:
+            values["window_seconds"] = None
+        return values
+
+    class Config:
+        json_encoders = JSON_ENCODERS
+        use_enum_values = True
+
+
+class NotificationOccurrence(BaseModel):
+    """A single occurrence represented by a grouped notification."""
+
+    occurred_at: datetime = Field(default_factory=lambda: datetime.now(tz=timezone.utc))
+    summary: str | None = Field(None, max_length=1000)
+    message: str | None = Field(None, max_length=5000)
+    level: NotificationLevel = Field(NotificationLevel.NORMAL)
+    is_new: bool = Field(
+        False,
+        description="Whether this occurrence arrived since the notification was last read.",
+    )
+
+    class Config:
+        json_encoders = JSON_ENCODERS
+        use_enum_values = True
+
+
+class Notification(Entry):
+    """A notification addressed to one user."""
+
+    type: str = Field("notifications", const=True)
+    recipient_id: PyObjectId = Field(..., description="ID of the user receiving the notification")
+    title: str = Field(..., min_length=1, max_length=200)
+    summary: str | None = Field(
+        None,
+        max_length=1000,
+        description="Short text shown in compact notification lists.",
+    )
+    message: str | None = Field(None, max_length=5000)
+    level: NotificationLevel = Field(NotificationLevel.NORMAL)
+    created_at: datetime = Field(default_factory=lambda: datetime.now(tz=timezone.utc))
+    created_by: PyObjectId | None = Field(None, description="User ID that created the notification")
+    read_at: datetime | None = None
+    archived_at: datetime | None = None
+    grouping: NotificationGrouping | None = None
+    occurrence_count: int = Field(1, ge=1)
+    last_occurred_at: datetime = Field(default_factory=lambda: datetime.now(tz=timezone.utc))
+    occurrences: list[NotificationOccurrence] | None = Field(
+        None,
+        description="Individual occurrence details for grouped notifications.",
+    )
+
+    class Config(Entry.Config):
+        use_enum_values = True

pydatalab/src/pydatalab/mongo.py

@@ -370,6 +370,38 @@ def create_group_fts():
         "refcode", unique=True, name="unique refcode counter", background=background
     )
 
+    from pydatalab.config import CONFIG
+
+    if CONFIG.ENABLE_NOTIFICATIONS:
+        ret += db.notifications.create_index(
+            [("recipient_id", pymongo.ASCENDING), ("created_at", pymongo.DESCENDING)],
+            name="notification recipient and created",
+            background=background,
+        )
+        ret += db.notifications.create_index(
+            [("recipient_id", pymongo.ASCENDING), ("read_at", pymongo.ASCENDING)],
+            name="notification recipient and read",
+            background=background,
+        )
+        ret += db.notifications.create_index(
+            [("recipient_id", pymongo.ASCENDING), ("archived_at", pymongo.ASCENDING)],
+            name="notification recipient and archived",
+            background=background,
+        )
+        ret += db.notifications.create_index(
+            [
+                ("recipient_id", pymongo.ASCENDING),
+                ("grouping.key", pymongo.ASCENDING),
+                ("grouping.policy", pymongo.ASCENDING),
+                ("grouping.window_seconds", pymongo.ASCENDING),
+                ("grouping.max_occurrences", pymongo.ASCENDING),
+                ("occurrence_count", pymongo.ASCENDING),
+                ("last_occurred_at", pymongo.DESCENDING),
+            ],
+            name="notification grouping lookup",
+            background=background,
+        )
+
     return ret