59 lines
1.9 KiB
Python
59 lines
1.9 KiB
Python
|
"""invites: DB-backed access whitelist / request queue
|
||
|
|
|
||
|
|
Revision ID: 0008_invites
|
||
|
|
Revises: 0007_deep_requested
|
||
|
|
Create Date: 2026-06-12
|
||
|
|
"""
|
||
|
|
from typing import Sequence, Union
|
||
|
|
|
||
|
|
from alembic import op
|
||
|
|
import sqlalchemy as sa
|
||
|
|
|
||
|
|
from app.config import settings
|
||
|
|
|
||
|
|
revision: str = "0008_invites"
|
||
|
|
down_revision: Union[str, None] = "0007_deep_requested"
|
||
|
|
branch_labels: Union[str, Sequence[str], None] = None
|
||
|
|
depends_on: Union[str, Sequence[str], None] = None
|
||
|
|
|
||
|
|
|
||
|
|
def upgrade() -> None:
|
||
|
|
op.create_table(
|
||
|
|
"invites",
|
||
|
|
sa.Column("id", sa.Integer(), primary_key=True),
|
||
|
|
sa.Column("email", sa.String(length=320), nullable=False),
|
||
|
|
sa.Column(
|
||
|
|
"status", sa.String(length=16), nullable=False, server_default="pending"
|
||
|
|
),
|
||
|
|
sa.Column("note", sa.Text(), nullable=True),
|
||
|
|
sa.Column(
|
||
|
|
"requested_at",
|
||
|
|
sa.DateTime(timezone=True),
|
||
|
|
server_default=sa.func.now(),
|
||
|
|
nullable=False,
|
||
|
|
),
|
||
|
|
sa.Column("decided_at", sa.DateTime(timezone=True), nullable=True),
|
||
|
|
sa.Column("decided_by", sa.String(length=320), nullable=True),
|
||
|
|
)
|
||
|
|
op.create_index("ix_invites_email", "invites", ["email"], unique=True)
|
||
|
|
|
||
|
|
# Seed the current env-based whitelist as already-approved invites so existing access
|
||
|
|
# is preserved and shows up in the admin UI. Idempotent on re-run via the unique email.
|
||
|
|
seed = sorted(settings.allowed_email_set | settings.admin_email_set)
|
||
|
|
if seed:
|
||
|
|
invites = sa.table(
|
||
|
|
"invites",
|
||
|
|
sa.column("email", sa.String),
|
||
|
|
sa.column("status", sa.String),
|
||
|
|
sa.column("decided_by", sa.String),
|
||
|
|
)
|
||
|
|
op.bulk_insert(
|
||
|
|
invites,
|
||
|
|
[{"email": e, "status": "approved", "decided_by": "seed"} for e in seed],
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def downgrade() -> None:
|
||
|
|
op.drop_index("ix_invites_email", table_name="invites")
|
||
|
|
op.drop_table("invites")
|