Cheng Zhou
89 lines
3.9 KiB
Python
89 lines
3.9 KiB
Python
"""add etmf nodes
|
|
|
|
Revision ID: 20260527_09
|
|
Revises: 20260527_08_precautions
|
|
Create Date: 2026-05-27 20:30:00.000000
|
|
|
|
"""
|
|
|
|
from typing import Sequence, Union
|
|
|
|
from alembic import op
|
|
import sqlalchemy as sa
|
|
from sqlalchemy.dialects import postgresql
|
|
|
|
|
|
revision: str = "20260527_09"
|
|
down_revision: Union[str, None] = "20260527_08_precautions"
|
|
branch_labels: Union[str, Sequence[str], None] = None
|
|
depends_on: Union[str, Sequence[str], None] = None
|
|
|
|
|
|
def _table_exists(inspector: sa.Inspector, table_name: str) -> bool:
|
|
return table_name in inspector.get_table_names()
|
|
|
|
|
|
def _column_exists(inspector: sa.Inspector, table_name: str, column_name: str) -> bool:
|
|
return column_name in {column["name"] for column in inspector.get_columns(table_name)}
|
|
|
|
|
|
def upgrade() -> None:
|
|
bind = op.get_bind()
|
|
inspector = sa.inspect(bind)
|
|
uuid_type = postgresql.UUID(as_uuid=True)
|
|
scope_type = postgresql.ENUM(
|
|
"GLOBAL",
|
|
"SITE",
|
|
"DERIVED",
|
|
name="document_scope_type",
|
|
create_type=False,
|
|
)
|
|
|
|
if not _table_exists(inspector, "etmf_nodes"):
|
|
op.create_table(
|
|
"etmf_nodes",
|
|
sa.Column("id", uuid_type, nullable=False),
|
|
sa.Column("study_id", uuid_type, nullable=False),
|
|
sa.Column("parent_id", uuid_type, nullable=True),
|
|
sa.Column("code", sa.String(length=50), nullable=False),
|
|
sa.Column("name", sa.String(length=200), nullable=False),
|
|
sa.Column("description", sa.Text(), nullable=True),
|
|
sa.Column("scope_type", scope_type, server_default="GLOBAL", nullable=False),
|
|
sa.Column("required", sa.Boolean(), server_default=sa.text("false"), nullable=False),
|
|
sa.Column("expected_doc_type", sa.String(length=50), nullable=True),
|
|
sa.Column("sort_order", sa.Integer(), server_default="0", nullable=False),
|
|
sa.Column("is_active", sa.Boolean(), server_default=sa.text("true"), nullable=False),
|
|
sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
|
|
sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
|
|
sa.ForeignKeyConstraint(["parent_id"], ["etmf_nodes.id"]),
|
|
sa.ForeignKeyConstraint(["study_id"], ["studies.id"]),
|
|
sa.PrimaryKeyConstraint("id"),
|
|
sa.UniqueConstraint("study_id", "parent_id", "code", name="uq_etmf_nodes_study_parent_code"),
|
|
)
|
|
op.create_index("ix_etmf_nodes_study_id", "etmf_nodes", ["study_id"])
|
|
op.create_index("ix_etmf_nodes_parent_id", "etmf_nodes", ["parent_id"])
|
|
op.create_index("ix_etmf_nodes_scope_type", "etmf_nodes", ["scope_type"])
|
|
op.create_index("ix_etmf_nodes_is_active", "etmf_nodes", ["is_active"])
|
|
|
|
if _table_exists(inspector, "documents") and not _column_exists(inspector, "documents", "etmf_node_id"):
|
|
op.add_column("documents", sa.Column("etmf_node_id", uuid_type, nullable=True))
|
|
op.create_index("ix_documents_etmf_node_id", "documents", ["etmf_node_id"])
|
|
op.create_foreign_key("fk_documents_etmf_node_id", "documents", "etmf_nodes", ["etmf_node_id"], ["id"])
|
|
|
|
|
|
def downgrade() -> None:
|
|
bind = op.get_bind()
|
|
inspector = sa.inspect(bind)
|
|
|
|
if _table_exists(inspector, "documents") and _column_exists(inspector, "documents", "etmf_node_id"):
|
|
op.drop_constraint("fk_documents_etmf_node_id", "documents", type_="foreignkey")
|
|
op.drop_index("ix_documents_etmf_node_id", table_name="documents")
|
|
op.drop_column("documents", "etmf_node_id")
|
|
|
|
if _table_exists(inspector, "etmf_nodes"):
|
|
op.drop_index("ix_etmf_nodes_is_active", table_name="etmf_nodes")
|
|
op.drop_index("ix_etmf_nodes_scope_type", table_name="etmf_nodes")
|
|
op.drop_index("ix_etmf_nodes_parent_id", table_name="etmf_nodes")
|
|
op.drop_index("ix_etmf_nodes_study_id", table_name="etmf_nodes")
|
|
op.drop_table("etmf_nodes")
|