leocrm/alembic/versions/0027_unify_company_to_contact.py

"""Unify entity_type 'company' to 'contact' across all plugins.

Revision ID: 0027
Revises: 0026_mail_salt_security
Create Date: 2026-07-23

Changes:
- UPDATE entity_links SET entity_type='contact' WHERE entity_type='company'
- UPDATE tag_assignments SET entity_type='contact' WHERE entity_type='company'
- UPDATE calendar_entry_links SET entity_type='contact' WHERE entity_type='company'
- UPDATE addresses SET entity_type='contact' WHERE entity_type='company'
- ALTER TABLE mails RENAME COLUMN company_id TO contact_id (if exists)
"""
from alembic import op
import sqlalchemy as sa


revision = "0027_unify_company_to_contact"
down_revision = "0026_mail_salt_security"


def upgrade():
    # Update entity_links: company -> contact
    op.execute(
        "UPDATE entity_links SET entity_type = 'contact' WHERE entity_type = 'company'"
    )
    # Update tag_assignments: company -> contact
    op.execute(
        "UPDATE tag_assignments SET entity_type = 'contact' WHERE entity_type = 'company'"
    )
    # Update calendar_entry_links: company -> contact
    op.execute(
        "UPDATE calendar_entry_links SET entity_type = 'contact' WHERE entity_type = 'company'"
    )
    # Update addresses: company -> contact
    op.execute(
        "UPDATE addresses SET entity_type = 'contact' WHERE entity_type = 'company'"
    )
    # Rename company_id to contact_id in mails (if column exists)
    conn = op.get_bind()
    has_company_id = conn.execute(
        sa.text("SELECT 1 FROM information_schema.columns WHERE table_name='mails' AND column_name='company_id'")
    ).fetchone()
    has_contact_id = conn.execute(
        sa.text("SELECT 1 FROM information_schema.columns WHERE table_name='mails' AND column_name='contact_id'")
    ).fetchone()
    if has_company_id and has_contact_id:
        # Both columns exist: drop company_id (contact_id already present)
        op.drop_column("mails", "company_id")
    elif has_company_id:
        op.alter_column("mails", "company_id", new_column_name="contact_id")


def downgrade():
    # Revert entity_links: contact -> company (only for rows that were originally company)
    op.execute(
        "UPDATE entity_links SET entity_type = 'company' WHERE entity_type = 'contact'"
    )
    # Revert tag_assignments: contact -> company
    op.execute(
        "UPDATE tag_assignments SET entity_type = 'company' WHERE entity_type = 'contact'"
    )
    # Revert calendar_entry_links: contact -> company
    op.execute(
        "UPDATE calendar_entry_links SET entity_type = 'company' WHERE entity_type = 'contact'"
    )
    # Revert addresses: contact -> company
    op.execute(
        "UPDATE addresses SET entity_type = 'company' WHERE entity_type = 'contact'"
    )
    # Rename contact_id back to company_id in mails (only if contact_id exists and company_id doesn't)
    conn = op.get_bind()
    has_contact_id = conn.execute(
        sa.text("SELECT 1 FROM information_schema.columns WHERE table_name='mails' AND column_name='contact_id'")
    ).fetchone()
    has_company_id = conn.execute(
        sa.text("SELECT 1 FROM information_schema.columns WHERE table_name='mails' AND column_name='company_id'")
    ).fetchone()
    if has_contact_id and not has_company_id:
        op.alter_column("mails", "contact_id", new_column_name="company_id")