ab591d30c4
Some checks failed
Aegis CI / lint-and-test (push) Has been cancelled
PostureSnapshot model, Alembic migration (b039exec), schemas, service aggregating all phases (coverage/risk/operations/knowledge/MTTD), and router at /api/v1/dashboard with executive view, KPIs, coverage-by-tactic, posture-history, posture-snapshot, and activity-feed endpoints. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
78 lines
2.5 KiB
Python
78 lines
2.5 KiB
Python
"""Phase 13: Executive Dashboard — posture_snapshots table.
|
|
|
|
Revision ID: b039exec
|
|
Revises: b038risk
|
|
Create Date: 2026-05-20
|
|
"""
|
|
|
|
from alembic import op
|
|
import sqlalchemy as sa
|
|
|
|
revision = "b039exec"
|
|
down_revision = "b038risk"
|
|
branch_labels = None
|
|
depends_on = None
|
|
|
|
|
|
def upgrade() -> None:
|
|
conn = op.get_bind()
|
|
conn.execute(sa.text("""
|
|
CREATE TABLE IF NOT EXISTS posture_snapshots (
|
|
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
|
snapshot_date DATE NOT NULL,
|
|
|
|
-- Coverage
|
|
total_techniques INTEGER NOT NULL DEFAULT 0,
|
|
validated_count INTEGER NOT NULL DEFAULT 0,
|
|
partial_count INTEGER NOT NULL DEFAULT 0,
|
|
not_covered_count INTEGER NOT NULL DEFAULT 0,
|
|
coverage_pct FLOAT NOT NULL DEFAULT 0.0,
|
|
|
|
-- Risk
|
|
avg_risk_score FLOAT NOT NULL DEFAULT 0.0,
|
|
critical_count INTEGER NOT NULL DEFAULT 0,
|
|
high_count INTEGER NOT NULL DEFAULT 0,
|
|
medium_count INTEGER NOT NULL DEFAULT 0,
|
|
low_count INTEGER NOT NULL DEFAULT 0,
|
|
|
|
-- Operations
|
|
open_queue_items INTEGER NOT NULL DEFAULT 0,
|
|
orphan_techniques INTEGER NOT NULL DEFAULT 0,
|
|
|
|
-- Knowledge
|
|
playbook_count INTEGER NOT NULL DEFAULT 0,
|
|
lesson_count INTEGER NOT NULL DEFAULT 0,
|
|
|
|
-- MTTD
|
|
mttd_avg_seconds FLOAT,
|
|
executions_30d INTEGER NOT NULL DEFAULT 0,
|
|
detection_rate_30d FLOAT,
|
|
|
|
-- Meta
|
|
created_by UUID REFERENCES users(id) ON DELETE SET NULL,
|
|
created_at TIMESTAMP WITHOUT TIME ZONE DEFAULT now(),
|
|
extra JSONB
|
|
)
|
|
"""))
|
|
|
|
# Unique constraint: one snapshot per calendar day
|
|
conn.execute(sa.text("""
|
|
DO $$ BEGIN
|
|
ALTER TABLE posture_snapshots
|
|
ADD CONSTRAINT uq_posture_snapshot_date UNIQUE (snapshot_date);
|
|
EXCEPTION WHEN duplicate_table THEN NULL;
|
|
WHEN duplicate_object THEN NULL;
|
|
END $$
|
|
"""))
|
|
|
|
# Index for date-range trend queries
|
|
conn.execute(sa.text("""
|
|
CREATE INDEX IF NOT EXISTS ix_posture_snapshots_date
|
|
ON posture_snapshots (snapshot_date)
|
|
"""))
|
|
|
|
|
|
def downgrade() -> None:
|
|
conn = op.get_bind()
|
|
conn.execute(sa.text("DROP TABLE IF EXISTS posture_snapshots CASCADE"))
|