Jul 8 - Implement QR code per facility

2026-07-08 13:06:52 -04:00
parent d0e72af188
commit ff3c76c0d8
11 changed files with 581 additions and 4 deletions
@@ -0,0 +1,74 @@
+"""phase34 — facilities.public_token for public QR landing pages
+
+Adds a unique, unguessable token per facility. The customer-facing QR code
+encodes /f/<public_token>, which serves an occupant-friendly summary + a
+"report a problem" form with no login required. Existing facilities are
+backfilled with a generated token.
+
+Uses INFORMATION_SCHEMA column-existence check — safe to re-run.
+"""
+
+revision      = 'phase34_facility_qr'
+down_revision = 'phase33_contract_recipients'
+branch_labels = None
+depends_on    = None
+
+import secrets
+from alembic import op
+import sqlalchemy as sa
+
+
+def _column_exists(conn, table, column):
+    result = 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})
+    return result.scalar() > 0
+
+
+def upgrade():
+    bind = op.get_bind()
+
+    if not _column_exists(bind, 'facilities', 'public_token'):
+        op.execute(sa.text(
+            "ALTER TABLE facilities ADD COLUMN public_token VARCHAR(48) NULL"
+        ))
+
+    # Backfill a unique token for every existing facility that lacks one.
+    rows = bind.execute(sa.text(
+        "SELECT id FROM facilities WHERE public_token IS NULL OR public_token = ''"
+    )).fetchall()
+    for (fid,) in rows:
+        # token_urlsafe(24) → ~32 URL-safe chars; well within VARCHAR(48).
+        token = secrets.token_urlsafe(24)
+        bind.execute(
+            sa.text("UPDATE facilities SET public_token = :tok WHERE id = :id"),
+            {"tok": token, "id": fid},
+        )
+
+    # Enforce uniqueness now that every row has a value.
+    # (Separate from the ADD COLUMN so the backfill can complete first.)
+    existing_idx = bind.execute(sa.text(
+        "SELECT COUNT(*) FROM INFORMATION_SCHEMA.STATISTICS "
+        "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'facilities' "
+        "AND INDEX_NAME = 'uq_facility_public_token'"
+    )).scalar()
+    if not existing_idx:
+        op.execute(sa.text(
+            "CREATE UNIQUE INDEX uq_facility_public_token "
+            "ON facilities (public_token)"
+        ))
+
+
+def downgrade():
+    bind = op.get_bind()
+    existing_idx = bind.execute(sa.text(
+        "SELECT COUNT(*) FROM INFORMATION_SCHEMA.STATISTICS "
+        "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'facilities' "
+        "AND INDEX_NAME = 'uq_facility_public_token'"
+    )).scalar()
+    if existing_idx:
+        op.execute(sa.text("DROP INDEX uq_facility_public_token ON facilities"))
+    if _column_exists(bind, 'facilities', 'public_token'):
+        op.execute(sa.text("ALTER TABLE facilities DROP COLUMN public_token"))