populate_denormalized_fields.py

from django.db.models import OuterRef, Subquery

from dcim import models

# Map each denormalized field to its "source" field on the parent device
device_component_fields = {
    '_site': 'site',
    '_location': 'location',
    '_rack': 'rack',
}

# All device components
denormalized_fields = {
    models.ConsolePort: device_component_fields,
    models.ConsoleServerPort: device_component_fields,
    models.DeviceBay: device_component_fields,
    models.FrontPort: device_component_fields,
    models.Interface: device_component_fields,
    models.InventoryItem: device_component_fields,
    models.PowerOutlet: device_component_fields,
    models.PowerPort: device_component_fields,
    models.RearPort: device_component_fields,
}

def populate_denormalized_fields(*models_to_update):
    """
    Repopulate the denormalized fields for the given models.

    Parameters:
        models_to_update: The model classes to be updated. If not specified, all models will be updated.
    """
    if not models_to_update:
        models_to_update = denormalized_fields.keys()
    for model in models_to_update:
        field_mappings = denormalized_fields.get(model)
        for field, source_field in field_mappings.items():
            # Filter for objects with the denormalized field set to null. Note: This doesn't necessarily indicate an
            # error. For instance, the _location field of an Interface is expected to be null if its parent Device
            # is not assigned to a Location.
            qs = model.objects.filter(**{f'{field}__isnull': True})
            if null_count := qs.count():
                print(f'{model.__name__}.{field}: Updating {null_count} null values')
                # Employ a Subquery to efficiently update all records
                qs.update(**{
                    field: Subquery(
                        models.Device.objects.filter(pk=OuterRef('device_id')).values(f'{source_field}_id')[:1]
                    )
                })
            else:
                print(f'{model.__name__}.{field}: No null values found; skipping')