c1e06d4c0a
Some checks failed
Aegis CI / lint-and-test (push) Has been cancelled
- Phase 6.1: WebhookConfig model, CRUD router (/api/v1/webhooks, admin-only), dispatch_webhook() with HMAC signing; integrated into test validation, campaign completion, and MITRE sync job - Phase 7.1: SMTP email service with send_test_validated_email, send_campaign_completed_email, send_new_mitre_techniques_email; notify_role_with_email() added to notification_service - Phase 7.2: notification_preferences and jira_account_id on User model; PATCH /users/me/preferences endpoint; Alembic migrations b031phase6 and b032phase7 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
42 lines
1.4 KiB
Python
42 lines
1.4 KiB
Python
"""Phase 7.2: user notification_preferences and jira_account_id columns.
|
|
|
|
Revision ID: b032phase7
|
|
Revises: b031phase6
|
|
"""
|
|
from typing import Sequence, Union
|
|
import sqlalchemy as sa
|
|
from sqlalchemy.dialects import postgresql
|
|
from alembic import op
|
|
|
|
revision: str = "b032phase7"
|
|
down_revision: Union[str, None] = "b031phase6"
|
|
branch_labels: Union[str, Sequence[str], None] = None
|
|
depends_on: Union[str, Sequence[str], None] = None
|
|
|
|
_DEFAULT_PREFS = '{"email_on_test_validated": true, "email_on_campaign_completed": true, "email_on_new_mitre_techniques": false, "in_app_all": true}'
|
|
|
|
def _column_names(table: str) -> set[str]:
|
|
bind = op.get_bind()
|
|
insp = sa.inspect(bind)
|
|
return {c["name"] for c in insp.get_columns(table)}
|
|
|
|
def upgrade() -> None:
|
|
user_cols = _column_names("users")
|
|
if "notification_preferences" not in user_cols:
|
|
op.add_column(
|
|
"users",
|
|
sa.Column("notification_preferences", postgresql.JSONB, nullable=True, server_default=_DEFAULT_PREFS),
|
|
)
|
|
if "jira_account_id" not in user_cols:
|
|
op.add_column(
|
|
"users",
|
|
sa.Column("jira_account_id", sa.String(100), nullable=True),
|
|
)
|
|
|
|
def downgrade() -> None:
|
|
user_cols = _column_names("users")
|
|
if "jira_account_id" in user_cols:
|
|
op.drop_column("users", "jira_account_id")
|
|
if "notification_preferences" in user_cols:
|
|
op.drop_column("users", "notification_preferences")
|