Implement Database Models for Notification Rules and Channels
- Dominant language
- Python
- Stars
- 670
- Forks
- 183
- Avg merge
- 17h 7m
- Merged PRs (30d)
- 358
Description
## User Story
As a system administrator, I want to store notification rules and channel configurations in the database, so that I can dynamically manage notification routing without code deployment.
## Context
This story defines the database schema for storing:
- Notification routing rules (which events trigger which channels)
- Channel configurations (webhook URLs, SMTP settings, Slack webhooks, etc.)
- Message templates for customization
- User preferences for notification subscriptions
## Functional Requirements
### 1. Database Models
#### NotificationRule Table
```python
class NotificationRule(Base):
__tablename__ = \"notification_rules\"
id = sa.Column(UUID, primary_key=True, default=uuid4)
name = sa.Column(sa.String(255), nullable=False)
description = sa.Column(sa.Text)
# Event matching
event_type = sa.Column(sa.String(255), nullable=False, index=True)
event_filter = sa.Column(JSONB) # Optional filtering conditions
# Target configuration
channel_id = sa.Column(UUID, sa.ForeignKey(\"notification_channels.id\"), nullable=False)
# Message template
message_template = sa.Column(sa.Text, nullable=False)
# Control
enabled = sa.Column(sa.Boolean, default=True, nullable=False, index=True)
priority = sa.Column(sa.Integer, default=0) # Execution order
# Metadata
created_by = sa.Column(UUID, sa.ForeignKey(\"users.id\"))
created_at = sa.Column(sa.DateTime, default=datetime.utcnow)
updated_at = sa.Column(sa.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# Relationships
channel = relationship(\"NotificationChannel\", back_populates=\"rules\")
creator = relationship(\"User\")
```
#### NotificationChannel Table
```python
class NotificationChannel(Base):
__tablename__ = \"notification_channels\"
id = sa.Column(UUID, primary_key=True, default=uuid4)
name = sa.Column(sa.String(255), nullable=False, unique=True)
description = sa.Column(sa.Text)
# Channel type
channel_type = sa.Column(
sa.Enum(\"webhook\", \"email\", \"slack\", \"discord\", name=\"notification_channel_type\"),
nullable=False
)
# Configuration (type-specific)
config = sa.Column(JSONB, nullable=False)
# Examples:
# webhook: {\"url\": \"...\", \"method\": \"POST\", \"headers\": {...}, \"retry\": {...}}
# email: {\"smtp_host\": \"...\", \"smtp_port\": 587, \"from_addr\": \"...\", \"to_addrs\": [...]}
# slack: {\"webhook_url\": \"...\", \"channel\": \"#notifications\"}
# Control
enabled = sa.Column(sa.Boolean, default=True, nullable=False, index=True)
# Metadata
created_by = sa.Column(UUID, sa.ForeignKey(\"users.id\"))
created_at = sa.Column(sa.DateTime, default=datetime.utcnow)
updated_at = sa.Column(sa.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# Relationships
rules = relationship(\"NotificationRule\", back_populates=\"channel\")
creator = relationship(\"User\")
```
#### NotificationLog Table (Optional for audit)
```python
class NotificationLog(Base):
__tablename__ = \"notification_logs\"
id = sa.Column(UUID, primary_key=True, default=uuid4)
# Event info
event_id = sa.Column(sa.String(255), nullable=False, index=True)
event_type = sa.Column(sa.String(255), nullable=False, index=True)
# Delivery info
rule_id = sa.Column(UUID, sa.ForeignKey(\"notification_rules.id\"))
channel_id = sa.Column(UUID, sa.ForeignKey(\"notification_channels.id\"))
# Status
status = sa.Column(
sa.Enum(\"pending\", \"sent\", \"failed\", name=\"notification_status\"),
nullable=False,
index=True
)
error_message = sa.Column(sa.Text)
# Timing
created_at = sa.Column(sa.DateTime, default=datetime.utcnow, index=True)
sent_at = sa.Column(sa.DateTime)
# Payload snapshot
payload = sa.Column(JSONB) # For debugging
```
### 2. Database Migration
Create Alembic migration script:
```python
# versions/XXXXXX_add_notification_tables.py
def upgrade():
# Create enum types
op.execute(\"CREATE TYPE notification_channel_type AS ENUM ('webhook', 'email', 'slack', 'discord')\")
op.execute(\"CREATE TYPE notification_status AS ENUM ('pending', 'sent', 'failed')\")
# Create tables
op.create_table(
'notification_channels',
# ... columns
)
op.create_table(
'notification_rules',
# ... columns
)
op.create_table(
'notification_logs',
# ... columns
)
# Create indexes
op.create_index('ix_notification_rules_event_type', 'notification_rules', ['event_type'])
op.create_index('ix_notification_rules_enabled', 'notification_rules', ['enabled'])
# ...
def downgrade():
op.drop_table('notification_logs')
op.drop_table('notification_rules')
op.drop_table('notification_channels')
op.execute(\"DROP TYPE notification_status\")
op.execute(\"DROP TYPE notification_channel_type\")
```
### 3. Configuration Schemas
Define Pydantic schemas for validation:
```python
# Channel configurations
class WebhookConfig(BaseModel):
url: str
method: str = \"POST\"
headers: dict[str, str] = {}
timeout: int = 30
retry: RetryConfig = RetryConfig()
class EmailConfig(BaseModel):
smtp_host: str
smtp_port: int = 587
smtp_user: str
smtp_password: SecretStr
from_addr: str
to_addrs: list[str]
class SlackConfig(BaseModel):
webhook_url: str
channel: str | None = None
username: str = \"Backend.AI\"
```
### 4. Repository Layer
Implement repository for database operations:
```python
class NotificationRuleRepository:
async def find_by_event_type(
self,
db_session,
event_type: str,
enabled_only: bool = True
) -> list[NotificationRule]:
\"\"\"Find rules matching the event type\"\"\"
async def create(self, db_session, rule: NotificationRuleInput) -> NotificationRule:
\"\"\"Create a new notification rule\"\"\"
async def update(self, db_session, rule_id: UUID, updates: dict) -> NotificationRule:
\"\"\"Update an existing rule\"\"\"
async def delete(self, db_session, rule_id: UUID) -> None:
\"\"\"Delete a rule\"\"\"
```
## Acceptance Criteria
- [ ] Database models for NotificationRule, NotificationChannel, NotificationLog are created
- [ ] Alembic migration script creates tables with proper indexes
- [ ] JSONB columns have proper structure validation
- [ ] Foreign key relationships are correctly defined
- [ ] Repository layer provides CRUD operations
- [ ] Configuration schemas validate channel-specific settings
- [ ] Unit tests verify model relationships and constraints
- [ ] Migration can be applied and rolled back successfully
## Implementation Notes
- Location: `src/ai/backend/manager/models/notification.py`
- Migration: `alembic/versions/XXXXXX_add_notification_tables.py`
- Repository: Consider using existing RBAC mixin patterns if applicable
- Use JSONB for flexible configuration storage
- Add indexes for frequently queried fields (event_type, enabled)
## Dependencies
- Depends on: BA-2852 (Core Infrastructure) for NotificationMessage definition
JIRA Issue: BA-2853
Contributor guide
Assessment
This issue has not been assessed yet.