leocrm/alembic/versions/0013_addresses.py

"""Add addresses table, migrate existing address fields, drop old columns.

Revision ID: 0013_addresses
Revises: 0012_soft_delete
Create Date: 2026-07-04
"""

from __future__ import annotations

from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects.postgresql import UUID as PGUUID

revision: str = "0013_addresses"
down_revision: Union[str, None] = "0012_soft_delete"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    # 1. Create addresses table
    op.create_table(
        "addresses",
        sa.Column("id", PGUUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
        sa.Column("tenant_id", PGUUID(as_uuid=True), nullable=False, index=True),
        sa.Column("entity_type", sa.String(50), nullable=False),
        sa.Column("entity_id", PGUUID(as_uuid=True), nullable=False, index=True),
        sa.Column("label", sa.String(100), nullable=False),
        sa.Column("address_type", sa.String(50), nullable=False),
        sa.Column("street", sa.String(255), nullable=True),
        sa.Column("street_number", sa.String(20), nullable=True),
        sa.Column("city", sa.String(100), nullable=True),
        sa.Column("zip", sa.String(20), nullable=True),
        sa.Column("state", sa.String(100), nullable=True),
        sa.Column("country", sa.String(2), nullable=True),
        sa.Column("is_default", sa.Boolean, nullable=False, server_default=sa.text("false")),
        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
        sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True),
    )

    op.create_index("ix_addresses_tenant_entity", "addresses", ["tenant_id", "entity_type", "entity_id"])
    op.create_index("ix_addresses_tenant_type", "addresses", ["tenant_id", "address_type"])

    # Unique constraint: one default per (tenant, entity_type, entity_id, address_type)
    # Using a partial unique index WHERE is_default = true
    op.execute(
        "CREATE UNIQUE INDEX uq_address_default_per_type "
        "ON addresses (tenant_id, entity_type, entity_id, address_type) "
        "WHERE is_default = true AND deleted_at IS NULL"
    )

    # 2. Migrate existing address data from companies and contacts
    # Insert into addresses from companies where address_street is not null
    op.execute(
        """
        INSERT INTO addresses (id, tenant_id, entity_type, entity_id, label, address_type, street, city, zip, country, state, is_default, created_at, updated_at)
        SELECT gen_random_uuid(), tenant_id, 'company', id, 'Hauptsitz', 'headquarters',
               address_street, address_city, address_zip, address_country, address_state, true,
               NOW(), NOW()
        FROM companies
        WHERE address_street IS NOT NULL AND deleted_at IS NULL
        """
    )

    # Insert into addresses from contacts where address_street is not null
    op.execute(
        """
        INSERT INTO addresses (id, tenant_id, entity_type, entity_id, label, address_type, street, city, zip, country, state, is_default, created_at, updated_at)
        SELECT gen_random_uuid(), tenant_id, 'contact', id, 'Privat', 'private',
               address_street, address_city, address_zip, address_country, address_state, true,
               NOW(), NOW()
        FROM contacts
        WHERE address_street IS NOT NULL AND deleted_at IS NULL
        """
    )

    # 3. Drop address columns from companies
    op.drop_column("companies", "address_street")
    op.drop_column("companies", "address_city")
    op.drop_column("companies", "address_zip")
    op.drop_column("companies", "address_country")
    op.drop_column("companies", "address_state")

    # 4. Drop address columns from contacts
    op.drop_column("contacts", "address_street")
    op.drop_column("contacts", "address_city")
    op.drop_column("contacts", "address_zip")
    op.drop_column("contacts", "address_country")
    op.drop_column("contacts", "address_state")


def downgrade() -> None:
    # Re-add address columns to companies
    op.add_column("companies", sa.Column("address_street", sa.String(255), nullable=True))
    op.add_column("companies", sa.Column("address_city", sa.String(100), nullable=True))
    op.add_column("companies", sa.Column("address_zip", sa.String(20), nullable=True))
    op.add_column("companies", sa.Column("address_country", sa.String(2), nullable=True))
    op.add_column("companies", sa.Column("address_state", sa.String(100), nullable=True))

    # Re-add address columns to contacts
    op.add_column("contacts", sa.Column("address_street", sa.String(255), nullable=True))
    op.add_column("contacts", sa.Column("address_city", sa.String(100), nullable=True))
    op.add_column("contacts", sa.Column("address_zip", sa.String(20), nullable=True))
    op.add_column("contacts", sa.Column("address_country", sa.String(2), nullable=True))
    op.add_column("contacts", sa.Column("address_state", sa.String(100), nullable=True))

    # Drop addresses table
    op.execute("DROP INDEX IF EXISTS uq_address_default_per_type")
    op.drop_index("ix_addresses_tenant_entity", table_name="addresses")
    op.drop_index("ix_addresses_tenant_type", table_name="addresses")
    op.drop_table("addresses")