59 lines
2.1 KiB
Python
59 lines
2.1 KiB
Python
"""phase38 — facility QR tokens
|
|
|
|
Adds facilities.qr_token VARCHAR(64) NULL + unique index. The token backs the
|
|
public, login-less facility QR scan page (GET /f/<token>) which shows recent
|
|
inspection scores, open-issue counts, and a 30-day score trend for that
|
|
facility — counts + scores only, no free text, no names, no photos.
|
|
|
|
NULL for all existing rows; Facility.ensure_qr_token() generates lazily when
|
|
staff first open the QR card / bulk sheet.
|
|
|
|
Idempotent: INFORMATION_SCHEMA column + index existence checks — safe to
|
|
re-run across every tenant DB (CLAUDE.md rule 14). Revision id kept ≤ 32
|
|
chars (alembic_version.version_num is VARCHAR(32)).
|
|
"""
|
|
|
|
import sqlalchemy as sa
|
|
from alembic import op
|
|
|
|
revision = 'phase38_facility_qr'
|
|
down_revision = 'phase37_contract_recipients'
|
|
branch_labels = None
|
|
depends_on = None
|
|
|
|
|
|
def _column_exists(bind, table: str, column: str) -> bool:
|
|
result = bind.execute(sa.text(
|
|
"SELECT COUNT(*) FROM information_schema.columns "
|
|
"WHERE table_schema = DATABASE() AND table_name = :t AND column_name = :c"
|
|
), {'t': table, 'c': column})
|
|
return result.scalar() > 0
|
|
|
|
|
|
def _index_exists(bind, table: str, index: str) -> bool:
|
|
result = bind.execute(sa.text(
|
|
"SELECT COUNT(*) FROM information_schema.statistics "
|
|
"WHERE table_schema = DATABASE() AND table_name = :t AND index_name = :i"
|
|
), {'t': table, 'i': index})
|
|
return result.scalar() > 0
|
|
|
|
|
|
def upgrade():
|
|
bind = op.get_bind()
|
|
if not _column_exists(bind, 'facilities', 'qr_token'):
|
|
op.execute(sa.text(
|
|
"ALTER TABLE facilities ADD COLUMN qr_token VARCHAR(64) NULL"
|
|
))
|
|
if not _index_exists(bind, 'facilities', 'uq_facilities_qr_token'):
|
|
op.execute(sa.text(
|
|
"CREATE UNIQUE INDEX uq_facilities_qr_token ON facilities (qr_token)"
|
|
))
|
|
|
|
|
|
def downgrade():
|
|
bind = op.get_bind()
|
|
if _index_exists(bind, 'facilities', 'uq_facilities_qr_token'):
|
|
op.execute(sa.text('DROP INDEX uq_facilities_qr_token ON facilities'))
|
|
if _column_exists(bind, 'facilities', 'qr_token'):
|
|
op.execute(sa.text('ALTER TABLE facilities DROP COLUMN qr_token'))
|