Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
79 changes: 16 additions & 63 deletions alembic_osm/versions/9221408912dd_add_user_role_table.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@

import sqlalchemy as sa
from alembic import op
from sqlalchemy import inspect, text
from sqlalchemy import text

# revision identifiers, used by Alembic.
revision: str = "9221408912dd"
Expand All @@ -20,69 +20,22 @@


def upgrade() -> None:
bind = op.get_bind()
assert bind is not None
insp = inspect(bind)

# Add unique constraint on users.auth_uid (if not already present)
constraint_exists = bind.execute(
text("SELECT 1 FROM pg_constraint WHERE conname = 'auth_uid_unique'")
).scalar()
if not constraint_exists:
op.create_unique_constraint("auth_uid_unique", "users", ["auth_uid"])

# Create the workspace_role enum type (if not already present)
result = bind.execute(
text("SELECT 1 FROM pg_type WHERE typname = 'workspace_role'")
op.create_unique_constraint("auth_uid_unique", "users", ["auth_uid"])
op.create_table(
"user_workspace_roles",
sa.Column("user_auth_uid", sa.String(), nullable=False),
sa.Column("workspace_id", sa.BigInteger(), nullable=False),
sa.Column(
"role",
sa.Enum("lead", "validator", "contributor", name="workspace_role"),
nullable=False,
),
sa.ForeignKeyConstraint(["user_auth_uid"], ["users.auth_uid"]),
sa.PrimaryKeyConstraint("user_auth_uid", "workspace_id"),
)
if not result.scalar():
workspace_role = sa.Enum(
"lead", "validator", "contributor", name="workspace_role"
)
workspace_role.create(bind)

# Create the user_workspace_roles table (if not already present)
if not insp.has_table("user_workspace_roles"):
op.create_table(
"user_workspace_roles",
sa.Column("user_auth_uid", sa.String(), nullable=False),
sa.Column("workspace_id", sa.BigInteger(), nullable=False),
sa.Column(
"role",
sa.Enum(
"lead",
"validator",
"contributor",
name="workspace_role",
create_type=False,
),
nullable=False,
),
sa.ForeignKeyConstraint(["user_auth_uid"], ["users.auth_uid"]),
sa.PrimaryKeyConstraint("user_auth_uid", "workspace_id"),
)


def downgrade() -> None:
bind = op.get_bind()
assert bind is not None
insp = inspect(bind)

if insp.has_table("user_workspace_roles"):
op.drop_table("user_workspace_roles")

# Drop the enum type
result = bind.execute(
text("SELECT 1 FROM pg_type WHERE typname = 'workspace_role'")
)
if result.scalar():
workspace_role = sa.Enum(
"lead", "validator", "contributor", name="workspace_role"
)
workspace_role.drop(bind)

constraint_exists = bind.execute(
text("SELECT 1 FROM pg_constraint WHERE conname = 'auth_uid_unique'")
).scalar()
if constraint_exists:
op.drop_constraint("auth_uid_unique", "users", type_="unique")
op.drop_table("user_workspace_roles")
op.execute(text("DROP TYPE workspace_role"))
op.drop_constraint("auth_uid_unique", "users", type_="unique")
Loading