63 lines
2.0 KiB
Python
63 lines
2.0 KiB
Python
"""phase15 — add indexes on audit_logs action and entity_type
|
|
|
|
The audit log filter UI allows filtering by action and entity_type.
|
|
Without indexes, every filter invocation performs a full table scan.
|
|
As the log grows toward the 180/365-day purge threshold this degrades
|
|
noticeably. This migration adds individual indexes on both columns.
|
|
|
|
A composite index on (action, entity_type, created_at) would be ideal
|
|
for the combined-filter case, but individual indexes are added here to
|
|
keep the migration additive and safe for re-run. created_at already
|
|
has an index from the model definition.
|
|
|
|
Existence checks use information_schema so the migration is safe to
|
|
re-run on any MySQL version (compatible back to 5.7).
|
|
|
|
Revision ID: phase15_audit_log_indexes
|
|
Revises: phase14_facility_created_at
|
|
"""
|
|
|
|
revision = 'phase15_audit_log_indexes'
|
|
down_revision = 'phase14_facility_created_at'
|
|
branch_labels = None
|
|
depends_on = None
|
|
|
|
from alembic import op
|
|
from sqlalchemy import text
|
|
|
|
|
|
def _index_exists(conn, table: str, index_name: str) -> bool:
|
|
"""Return True if the named index already exists on the given table."""
|
|
result = conn.execute(text(
|
|
"SELECT COUNT(*) FROM information_schema.statistics "
|
|
"WHERE table_schema = DATABASE() "
|
|
" AND table_name = :table "
|
|
" AND index_name = :index"
|
|
), {'table': table, 'index': index_name})
|
|
return result.scalar() > 0
|
|
|
|
|
|
# (table, index_name, column)
|
|
INDEXES = [
|
|
('audit_logs', 'ix_audit_logs_action', 'action'),
|
|
('audit_logs', 'ix_audit_logs_entity_type', 'entity_type'),
|
|
]
|
|
|
|
|
|
def upgrade():
|
|
conn = op.get_bind()
|
|
for table, index_name, column in INDEXES:
|
|
if not _index_exists(conn, table, index_name):
|
|
op.execute(text(
|
|
f'CREATE INDEX {index_name} ON {table} ({column})'
|
|
))
|
|
|
|
|
|
def downgrade():
|
|
conn = op.get_bind()
|
|
for table, index_name, _column in INDEXES:
|
|
if _index_exists(conn, table, index_name):
|
|
op.execute(text(
|
|
f'DROP INDEX {index_name} ON {table}'
|
|
))
|