"""Initial schema for CRM domain objects.""" from __future__ import annotations from alembic import op import sqlalchemy as sa from sqlalchemy.dialects import postgresql # revision identifiers, used by Alembic. revision: str = "20251122_0001" down_revision: str | None = None branch_labels: tuple[str, ...] | None = None depends_on: tuple[str, ...] | None = None def upgrade() -> None: organization_role = sa.Enum( "owner", "admin", "manager", "member", name="organization_role" ) deal_status = sa.Enum("new", "in_progress", "won", "lost", name="deal_status") deal_stage = sa.Enum("qualification", "proposal", "negotiation", "closed", name="deal_stage") activity_type = sa.Enum( "comment", "status_changed", "task_created", "system", name="activity_type" ) bind = op.get_bind() organization_role.create(bind, checkfirst=True) deal_status.create(bind, checkfirst=True) deal_stage.create(bind, checkfirst=True) activity_type.create(bind, checkfirst=True) op.create_table( "organizations", sa.Column("id", sa.Integer(), nullable=False), sa.Column("name", sa.String(length=255), nullable=False), sa.Column( "created_at", sa.DateTime(timezone=True), server_default=sa.text("timezone('utc', now())"), nullable=False, ), sa.PrimaryKeyConstraint("id"), sa.UniqueConstraint("name", name="uq_organizations_name"), ) op.create_index("ix_organizations_id", "organizations", ["id"], unique=False) op.create_table( "users", sa.Column("id", sa.Integer(), nullable=False), sa.Column("email", sa.String(length=320), nullable=False), sa.Column("hashed_password", sa.String(length=255), nullable=False), sa.Column("name", sa.String(length=255), nullable=False), sa.Column("is_active", sa.Boolean(), server_default=sa.true(), nullable=False), sa.Column( "created_at", sa.DateTime(timezone=True), server_default=sa.text("timezone('utc', now())"), nullable=False, ), sa.Column( "updated_at", sa.DateTime(timezone=True), server_default=sa.text("timezone('utc', now())"), nullable=False, ), sa.PrimaryKeyConstraint("id"), sa.UniqueConstraint("email", name="uq_users_email"), ) op.create_index("ix_users_id", "users", ["id"], unique=False) op.create_index("ix_users_email", "users", ["email"], unique=False) op.create_table( "organization_members", sa.Column("id", sa.Integer(), nullable=False), sa.Column("organization_id", sa.Integer(), nullable=False), sa.Column("user_id", sa.Integer(), nullable=False), sa.Column( "role", organization_role, nullable=False, server_default="member", ), sa.Column( "created_at", sa.DateTime(timezone=True), server_default=sa.text("timezone('utc', now())"), nullable=False, ), sa.ForeignKeyConstraint([ "organization_id" ], ["organizations.id"], ondelete="CASCADE"), sa.ForeignKeyConstraint(["user_id"], ["users.id"], ondelete="CASCADE"), sa.PrimaryKeyConstraint("id"), sa.UniqueConstraint( "organization_id", "user_id", name="uq_organization_member" ), ) op.create_table( "contacts", sa.Column("id", sa.Integer(), nullable=False), sa.Column("organization_id", sa.Integer(), nullable=False), sa.Column("owner_id", sa.Integer(), nullable=False), sa.Column("name", sa.String(length=255), nullable=False), sa.Column("email", sa.String(length=320), nullable=True), sa.Column("phone", sa.String(length=64), nullable=True), sa.Column( "created_at", sa.DateTime(timezone=True), server_default=sa.text("timezone('utc', now())"), nullable=False, ), sa.ForeignKeyConstraint([ "organization_id" ], ["organizations.id"], ondelete="CASCADE"), sa.ForeignKeyConstraint(["owner_id"], ["users.id"], ondelete="RESTRICT"), sa.PrimaryKeyConstraint("id"), ) op.create_table( "deals", sa.Column("id", sa.Integer(), nullable=False), sa.Column("organization_id", sa.Integer(), nullable=False), sa.Column("contact_id", sa.Integer(), nullable=False), sa.Column("owner_id", sa.Integer(), nullable=False), sa.Column("title", sa.String(length=255), nullable=False), sa.Column("amount", sa.Numeric(12, 2), nullable=True), sa.Column("currency", sa.String(length=8), nullable=True), sa.Column("status", deal_status, nullable=False, server_default="new"), sa.Column( "stage", deal_stage, nullable=False, server_default="qualification" ), sa.Column( "created_at", sa.DateTime(timezone=True), server_default=sa.text("timezone('utc', now())"), nullable=False, ), sa.Column( "updated_at", sa.DateTime(timezone=True), server_default=sa.text("timezone('utc', now())"), nullable=False, ), sa.ForeignKeyConstraint([ "organization_id" ], ["organizations.id"], ondelete="CASCADE"), sa.ForeignKeyConstraint(["contact_id"], ["contacts.id"], ondelete="RESTRICT"), sa.ForeignKeyConstraint(["owner_id"], ["users.id"], ondelete="RESTRICT"), sa.PrimaryKeyConstraint("id"), ) op.create_table( "tasks", sa.Column("id", sa.Integer(), nullable=False), sa.Column("deal_id", sa.Integer(), nullable=False), sa.Column("title", sa.Text(), nullable=False), sa.Column("description", sa.Text(), nullable=True), sa.Column( "due_date", sa.DateTime(timezone=True), nullable=True, ), sa.Column( "is_done", sa.Boolean(), nullable=False, server_default=sa.false(), ), sa.Column( "created_at", sa.DateTime(timezone=True), server_default=sa.text("timezone('utc', now())"), nullable=False, ), sa.ForeignKeyConstraint(["deal_id"], ["deals.id"], ondelete="CASCADE"), sa.PrimaryKeyConstraint("id"), ) op.create_table( "activities", sa.Column("id", sa.Integer(), nullable=False), sa.Column("deal_id", sa.Integer(), nullable=False), sa.Column("author_id", sa.Integer(), nullable=True), sa.Column("type", activity_type, nullable=False), sa.Column( "payload", postgresql.JSONB(astext_type=sa.Text()), server_default=sa.text("'{}'::jsonb"), nullable=False, ), sa.Column( "created_at", sa.DateTime(timezone=True), server_default=sa.text("timezone('utc', now())"), nullable=False, ), sa.ForeignKeyConstraint(["author_id"], ["users.id"], ondelete="SET NULL"), sa.ForeignKeyConstraint(["deal_id"], ["deals.id"], ondelete="CASCADE"), sa.PrimaryKeyConstraint("id"), ) def downgrade() -> None: op.drop_table("activities") op.drop_table("tasks") op.drop_table("deals") op.drop_table("contacts") op.drop_table("organization_members") op.drop_index("ix_users_email", table_name="users") op.drop_index("ix_users_id", table_name="users") op.drop_table("users") op.drop_index("ix_organizations_id", table_name="organizations") op.drop_table("organizations") organization_role = sa.Enum( "owner", "admin", "manager", "member", name="organization_role" ) deal_status = sa.Enum("new", "in_progress", "won", "lost", name="deal_status") deal_stage = sa.Enum("qualification", "proposal", "negotiation", "closed", name="deal_stage") activity_type = sa.Enum( "comment", "status_changed", "task_created", "system", name="activity_type" ) bind = op.get_bind() activity_type.drop(bind, checkfirst=True) deal_stage.drop(bind, checkfirst=True) deal_status.drop(bind, checkfirst=True) organization_role.drop(bind, checkfirst=True)