68 lines
2.3 KiB
Python
68 lines
2.3 KiB
Python
"""phase42 — areas.qr_token for per-area public QR landing pages
|
|
|
|
Adds a unique, unguessable token per area. Each area's QR encodes
|
|
/f/area/<qr_token>, a login-free summary scoped to that area plus a
|
|
"report a problem" form.
|
|
|
|
Ported from the single-tenant chain (phase39_area_public_token) and adapted:
|
|
the MT column is named `qr_token` (matching `facilities.qr_token` from
|
|
phase38_facility_qr), not `public_token`, and is VARCHAR(64) to match the
|
|
facility column and MT's `secrets.token_urlsafe(32)` generator.
|
|
|
|
Unlike the ST original, existing areas are NOT backfilled: MT mints tokens
|
|
lazily via `Area.ensure_qr_token()` on first use, exactly as
|
|
`Facility.ensure_qr_token()` already does. A backfill would mint tokens for
|
|
areas nobody ever prints a code for.
|
|
|
|
Uses INFORMATION_SCHEMA checks — safe to re-run on every tenant DB.
|
|
"""
|
|
|
|
revision = 'phase42_area_qr_token'
|
|
down_revision = 'phase41_auditor_role'
|
|
branch_labels = None
|
|
depends_on = None
|
|
|
|
from alembic import op
|
|
import sqlalchemy as sa
|
|
|
|
|
|
def _column_exists(conn, table, column):
|
|
return conn.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}).scalar() > 0
|
|
|
|
|
|
def _index_exists(conn, table, index):
|
|
return conn.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}).scalar() > 0
|
|
|
|
|
|
def upgrade():
|
|
bind = op.get_bind()
|
|
|
|
if not _column_exists(bind, 'areas', 'qr_token'):
|
|
op.execute(sa.text(
|
|
"ALTER TABLE areas ADD COLUMN qr_token VARCHAR(64) NULL"
|
|
))
|
|
|
|
# Unique index tolerates multiple NULLs in MySQL, so it can be created
|
|
# immediately — no backfill needed before enforcing uniqueness.
|
|
if not _index_exists(bind, 'areas', 'uq_area_qr_token'):
|
|
op.execute(sa.text(
|
|
"CREATE UNIQUE INDEX uq_area_qr_token ON areas (qr_token)"
|
|
))
|
|
|
|
|
|
def downgrade():
|
|
bind = op.get_bind()
|
|
|
|
if _index_exists(bind, 'areas', 'uq_area_qr_token'):
|
|
op.execute(sa.text("DROP INDEX uq_area_qr_token ON areas"))
|
|
if _column_exists(bind, 'areas', 'qr_token'):
|
|
op.execute(sa.text("ALTER TABLE areas DROP COLUMN qr_token"))
|