68 lines
2.3 KiB
Python
68 lines
2.3 KiB
Python
"""phase21 — composite performance indexes
|
|
|
|
Adds composite (multi-column) indexes on the highest-traffic query patterns.
|
|
Phase 12 already covers single-column indexes; these target the multi-column
|
|
WHERE clauses that appear on every Reports, Inspections list, and Issues list
|
|
page load.
|
|
|
|
inspections (facility_id, inspection_date)
|
|
— facility-scoped date-range queries on every list page and report
|
|
|
|
inspections (inspector_id, inspection_date)
|
|
— inspector-scoped date-range queries on the Performance page and API stats
|
|
|
|
inspections (status, inspection_date)
|
|
— "completed inspections in date range" pattern used by all score aggregations
|
|
|
|
issues (facility_id, status)
|
|
— "open issues at this facility" pattern used by reports and dashboard
|
|
|
|
All existence checks use INFORMATION_SCHEMA.STATISTICS — safe to re-run on
|
|
any MySQL version (compatible back to 5.7).
|
|
"""
|
|
|
|
import sqlalchemy as sa
|
|
from alembic import op
|
|
|
|
revision = 'phase21_performance_indexes'
|
|
down_revision = 'phase21_template_active'
|
|
branch_labels = None
|
|
depends_on = None
|
|
|
|
|
|
def _index_exists(bind, table: str, index_name: str) -> bool:
|
|
result = bind.execute(sa.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, columns)
|
|
INDEXES = [
|
|
('inspections', 'ix_inspections_facility_date', 'facility_id, inspection_date'),
|
|
('inspections', 'ix_inspections_inspector_date', 'inspector_id, inspection_date'),
|
|
('inspections', 'ix_inspections_status_date', 'status, inspection_date'),
|
|
('issues', 'ix_issues_facility_status', 'facility_id, status'),
|
|
]
|
|
|
|
|
|
def upgrade():
|
|
bind = op.get_bind()
|
|
for table, index_name, columns in INDEXES:
|
|
if not _index_exists(bind, table, index_name):
|
|
op.execute(sa.text(
|
|
f'CREATE INDEX {index_name} ON {table} ({columns})'
|
|
))
|
|
|
|
|
|
def downgrade():
|
|
bind = op.get_bind()
|
|
for table, index_name, _columns in INDEXES:
|
|
if _index_exists(bind, table, index_name):
|
|
op.execute(sa.text(
|
|
f'DROP INDEX {index_name} ON {table}'
|
|
))
|