"""Baseline schema for the current MVP state.""" from typing import Sequence from alembic import op import sqlalchemy as sa # revision identifiers, used by Alembic. revision: str = "20260607_0001" down_revision: str | None = None branch_labels: Sequence[str] | None = None depends_on: Sequence[str] | None = None def _has_table(inspector: sa.Inspector, table_name: str) -> bool: return table_name in set(inspector.get_table_names()) def _has_column(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) if not _has_table(inspector, "users"): op.create_table( "users", sa.Column("id", sa.Integer(), nullable=False), sa.Column("username", sa.String(length=64), nullable=False), sa.Column("password_hash", sa.String(length=255), nullable=False), sa.Column("role", sa.String(length=32), nullable=False), sa.Column("is_active", sa.Boolean(), nullable=False), sa.Column("token_version", sa.Integer(), nullable=False), sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), sa.PrimaryKeyConstraint("id"), sa.UniqueConstraint("username"), ) op.create_index("ix_users_id", "users", ["id"], unique=False) op.create_index("ix_users_username", "users", ["username"], unique=True) elif not _has_column(inspector, "users", "token_version"): # Keep this baseline revision compatible with pre-migration dev databases # that were created before token_version existed. op.add_column( "users", sa.Column( "token_version", sa.Integer(), nullable=False, server_default=sa.text("0"), ), ) if not _has_table(inspector, "servers"): op.create_table( "servers", sa.Column("id", sa.Integer(), nullable=False), sa.Column("name", sa.String(length=128), nullable=False), sa.Column("host", sa.String(length=255), nullable=False), sa.Column("port", sa.Integer(), nullable=False), sa.Column("ssh_username", sa.String(length=128), nullable=True), sa.Column("connection_type", sa.String(length=32), nullable=False), sa.Column("environment", sa.String(length=64), nullable=False), sa.Column("is_active", sa.Boolean(), nullable=False), sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), sa.PrimaryKeyConstraint("id"), sa.UniqueConstraint("name"), ) op.create_index("ix_servers_id", "servers", ["id"], unique=False) op.create_index("ix_servers_name", "servers", ["name"], unique=True) if not _has_table(inspector, "managed_services"): op.create_table( "managed_services", sa.Column("id", sa.Integer(), nullable=False), sa.Column("server_id", sa.Integer(), nullable=False), sa.Column("name", sa.String(length=128), nullable=False), sa.Column("service_type", sa.String(length=32), nullable=False), sa.Column("control_enabled", sa.Boolean(), nullable=False), sa.Column("log_source", sa.Text(), nullable=True), sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), sa.ForeignKeyConstraint(["server_id"], ["servers.id"]), sa.PrimaryKeyConstraint("id"), ) op.create_index("ix_managed_services_id", "managed_services", ["id"], unique=False) op.create_index("ix_managed_services_name", "managed_services", ["name"], unique=False) op.create_index( "ix_managed_services_server_id", "managed_services", ["server_id"], unique=False, ) if not _has_table(inspector, "telegram_links"): op.create_table( "telegram_links", sa.Column("id", sa.Integer(), nullable=False), sa.Column("user_id", sa.Integer(), nullable=False), sa.Column("telegram_user_id", sa.String(length=64), nullable=True), sa.Column("chat_id", sa.String(length=64), nullable=True), sa.Column("link_code", sa.String(length=32), nullable=True), sa.Column("link_code_expires_at", sa.DateTime(timezone=True), nullable=True), sa.Column("linked_at", sa.DateTime(timezone=True), nullable=True), sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), sa.ForeignKeyConstraint(["user_id"], ["users.id"]), sa.PrimaryKeyConstraint("id"), sa.UniqueConstraint("link_code"), sa.UniqueConstraint("telegram_user_id"), sa.UniqueConstraint("user_id"), ) op.create_index("ix_telegram_links_id", "telegram_links", ["id"], unique=False) op.create_index("ix_telegram_links_user_id", "telegram_links", ["user_id"], unique=True) if not _has_table(inspector, "actions"): op.create_table( "actions", sa.Column("id", sa.Integer(), nullable=False), sa.Column("server_id", sa.Integer(), nullable=False), sa.Column("service_id", sa.Integer(), nullable=True), sa.Column("target_type", sa.String(length=32), nullable=False), sa.Column("target_name", sa.String(length=128), nullable=False), sa.Column("action", sa.String(length=32), nullable=False), sa.Column("status", sa.String(length=32), nullable=False), sa.Column("source", sa.String(length=32), nullable=False), sa.Column("requested_by", sa.Integer(), nullable=False), sa.Column("result_message", sa.Text(), nullable=True), sa.Column("requested_at", sa.DateTime(timezone=True), nullable=False), sa.Column("completed_at", sa.DateTime(timezone=True), nullable=True), sa.ForeignKeyConstraint(["requested_by"], ["users.id"]), sa.ForeignKeyConstraint(["server_id"], ["servers.id"]), sa.ForeignKeyConstraint(["service_id"], ["managed_services.id"]), sa.PrimaryKeyConstraint("id"), ) op.create_index("ix_actions_id", "actions", ["id"], unique=False) op.create_index("ix_actions_requested_by", "actions", ["requested_by"], unique=False) op.create_index("ix_actions_server_id", "actions", ["server_id"], unique=False) if not _has_table(inspector, "audit_logs"): op.create_table( "audit_logs", sa.Column("id", sa.Integer(), nullable=False), sa.Column("actor_id", sa.Integer(), nullable=False), sa.Column("actor_source", sa.String(length=32), nullable=False), sa.Column("target_type", sa.String(length=32), nullable=False), sa.Column("target_id", sa.String(length=128), nullable=False), sa.Column("action", sa.String(length=64), nullable=False), sa.Column("result", sa.String(length=32), nullable=False), sa.Column("details", sa.JSON(), nullable=False), sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), sa.ForeignKeyConstraint(["actor_id"], ["users.id"]), sa.PrimaryKeyConstraint("id"), ) op.create_index("ix_audit_logs_actor_id", "audit_logs", ["actor_id"], unique=False) op.create_index("ix_audit_logs_id", "audit_logs", ["id"], unique=False) def downgrade() -> None: raise NotImplementedError( "Automatic downgrade is not supported for the baseline migration." )