115 lines
5.3 KiB
Python
115 lines
5.3 KiB
Python
"""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")
|