74 lines
2.5 KiB
Python
74 lines
2.5 KiB
Python
"""Prevent duplicate active file-folder paths.
|
|
|
|
Revision ID: 6a7b8c9d0e1f
|
|
Revises: 5f6a7b8c9d0e
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections import defaultdict
|
|
import sqlalchemy as sa
|
|
from alembic import op
|
|
from sqlalchemy import inspect
|
|
|
|
revision = "6a7b8c9d0e1f"
|
|
down_revision = "5f6a7b8c9d0e"
|
|
branch_labels = None
|
|
depends_on = None
|
|
|
|
USER_INDEX = "uq_file_folders_active_user_path"
|
|
GROUP_INDEX = "uq_file_folders_active_group_path"
|
|
|
|
|
|
def _deduplicate_active_folders() -> None:
|
|
bind = op.get_bind()
|
|
rows = bind.execute(sa.text("""
|
|
SELECT id, tenant_id, owner_type, owner_user_id, owner_group_id, path, created_at
|
|
FROM file_folders
|
|
WHERE deleted_at IS NULL
|
|
ORDER BY created_at ASC, id ASC
|
|
""")).mappings().all()
|
|
grouped: dict[tuple[object, ...], list[str]] = defaultdict(list)
|
|
for row in rows:
|
|
owner_id = row["owner_user_id"] if row["owner_type"] == "user" else row["owner_group_id"]
|
|
grouped[(row["tenant_id"], row["owner_type"], owner_id, row["path"])].append(row["id"])
|
|
duplicate_ids = [folder_id for ids in grouped.values() for folder_id in ids[1:]]
|
|
if duplicate_ids:
|
|
bind.execute(
|
|
sa.text("UPDATE file_folders SET deleted_at = CURRENT_TIMESTAMP WHERE id IN :ids").bindparams(
|
|
sa.bindparam("ids", expanding=True)
|
|
),
|
|
{"ids": duplicate_ids},
|
|
)
|
|
|
|
|
|
def upgrade() -> None:
|
|
_deduplicate_active_folders()
|
|
existing = {item["name"] for item in inspect(op.get_bind()).get_indexes("file_folders")}
|
|
if USER_INDEX not in existing:
|
|
op.create_index(
|
|
USER_INDEX,
|
|
"file_folders",
|
|
["tenant_id", "owner_user_id", "path"],
|
|
unique=True,
|
|
sqlite_where=sa.text("owner_type = 'user' AND deleted_at IS NULL"),
|
|
postgresql_where=sa.text("owner_type = 'user' AND deleted_at IS NULL"),
|
|
)
|
|
if GROUP_INDEX not in existing:
|
|
op.create_index(
|
|
GROUP_INDEX,
|
|
"file_folders",
|
|
["tenant_id", "owner_group_id", "path"],
|
|
unique=True,
|
|
sqlite_where=sa.text("owner_type = 'group' AND deleted_at IS NULL"),
|
|
postgresql_where=sa.text("owner_type = 'group' AND deleted_at IS NULL"),
|
|
)
|
|
|
|
|
|
def downgrade() -> None:
|
|
existing = {item["name"] for item in inspect(op.get_bind()).get_indexes("file_folders")}
|
|
if GROUP_INDEX in existing:
|
|
op.drop_index(GROUP_INDEX, table_name="file_folders")
|
|
if USER_INDEX in existing:
|
|
op.drop_index(USER_INDEX, table_name="file_folders")
|