signals

imperial_coldfront_plugin.signals

Django signals.

Signals are a rather blunt instrument but are useful for hooking into models defined by Coldfront. Here we use them to enforce some constraints on attributes and to manage LDAP group membership.

Functions

ensure_no_existing_gid(sender, instance, **kwargs)

Prevent saving of GID attribute if it is already in use.

This checks both existing allocation attributes and LDAP (if enabled).

Note that this makes all operations that create or modify GID attributes potentially slow as they involve a network call to the LDAP server. GID attributes should ideally be created in background tasks rather than in the request/response cycle.

Parameters:

Name	Type	Description	Default
sender	object	The model class.	required
instance	AllocationAttribute	The instance being saved.	required
**kwargs	object	Additional keyword arguments.	{}

Source code in imperial_coldfront_plugin/signals.py

@receiver(pre_save, sender=AllocationAttribute)
def ensure_no_existing_gid(
    sender: object, instance: AllocationAttribute, **kwargs: object
) -> None:
    """Prevent saving of GID attribute if it is already in use.

    This checks both existing allocation attributes and LDAP (if enabled).

    Note that this makes all operations that create or modify GID attributes
    potentially slow as they involve a network call to the LDAP server. GID attributes
    should ideally be created in background tasks rather than in the request/response
    cycle.

    Args:
        sender: The model class.
        instance: The instance being saved.
        **kwargs: Additional keyword arguments.
    """
    if instance.allocation_attribute_type.name != "GID":
        return
    if AllocationAttribute.objects.filter(
        allocation_attribute_type__name="GID", value=instance.value
    ):
        raise ValueError(
            f"GID {instance.value} is already assigned to another allocation."
        )
    if settings.LDAP_ENABLED and ldap_gid_in_use(instance.value):
        raise ValueError(f"GID {instance.value} is already in use in LDAP.")

ensure_unique_group_id(sender, instance, **kwargs)

Prevent saving of project group name if it is not unique.

Parameters:

Name	Type	Description	Default
sender	object	The model class.	required
instance	ProjectAttribute	The instance being saved.	required
**kwargs	object	Additional keyword arguments.	{}

Source code in imperial_coldfront_plugin/signals.py

@receiver(pre_save, sender=ProjectAttribute)
def ensure_unique_group_id(
    sender: object, instance: ProjectAttribute, **kwargs: object
) -> None:
    """Prevent saving of project group name if it is not unique.

    Args:
        sender: The model class.
        instance: The instance being saved.
        **kwargs: Additional keyword arguments.
    """
    if instance.proj_attr_type.name == "Group ID" and ProjectAttribute.objects.filter(
        proj_attr_type__name="Group ID", value=instance.value
    ):
        raise ValueError(f"A project with {instance.value} already exists.")

ensure_unique_shortname(sender, instance, **kwargs)

Prevent saving of shortname attribute if it is not unique.

Parameters:

Name	Type	Description	Default
sender	object	The model class.	required
instance	AllocationAttribute	The instance being saved.	required
**kwargs	object	Additional keyword arguments.	{}

Source code in imperial_coldfront_plugin/signals.py

@receiver(pre_save, sender=AllocationAttribute)
def ensure_unique_shortname(
    sender: object, instance: AllocationAttribute, **kwargs: object
) -> None:
    """Prevent saving of shortname attribute if it is not unique.

    Args:
      sender: The model class.
      instance: The instance being saved.
      **kwargs: Additional keyword arguments.
    """
    if (
        instance.allocation_attribute_type.name == "Shortname"
        and AllocationAttribute.objects.filter(
            allocation_attribute_type__name="Shortname", value=instance.value
        )
    ):
        raise ValueError(f"An allocation with {instance.value} already exists.")

remove_ldap_group_membership(sender, instance, **kwargs)

Remove an ldap group member if the associated AllocationUser is deleted.

This isn't expected to come up in the usual course of things as removing a user via the UI does not delete the AllocationUser object. Just covering it for completeness.

Note this signal invokes a background task to do the actual LDAP operation. This leaves the potential for the database and LDAP to get out of sync if the task fails, but avoids making the request/response cycle slow.

Parameters:

Name	Type	Description	Default
sender	object	The model class.	required
instance	AllocationUser	The instance being deleted.	required
**kwargs	object	Additional keyword arguments.	{}

Source code in imperial_coldfront_plugin/signals.py

@receiver(post_delete, sender=AllocationUser)
def remove_ldap_group_membership(
    sender: object, instance: AllocationUser, **kwargs: object
) -> None:
    """Remove an ldap group member if the associated AllocationUser is deleted.

    This isn't expected to come up in the usual course of things as removing a user via
    the UI does not delete the AllocationUser object. Just covering it for completeness.

    Note this signal invokes a background task to do the actual LDAP operation. This
    leaves the potential for the database and LDAP to get out of sync if the task
    fails, but avoids making the request/response cycle slow.

    Args:
        sender: The model class.
        instance: The instance being deleted.
        **kwargs: Additional keyword arguments.
    """
    if not settings.LDAP_ENABLED:
        return

    if (group_id := _get_shortname_from_allocation(instance.allocation)) is None:
        return

    async_task(
        ldap_remove_member_from_group,
        group_id,
        instance.user.username,
        allow_missing=True,
    )

sync_ldap_group_membership(sender, instance, **kwargs)

Add or remove members from an ldap group based on AllocationUser.status.

Note this signal invokes a background task to do the actual LDAP operation. This leaves the potential for the database and LDAP to get out of sync if the task fails, but avoids making the request/response cycle slow.

Parameters:

Name	Type	Description	Default
sender	object	The model class.	required
instance	AllocationUser	The instance being saved.	required
**kwargs	object	Additional keyword arguments.	{}

Source code in imperial_coldfront_plugin/signals.py

@receiver(post_save, sender=AllocationUser)
def sync_ldap_group_membership(
    sender: object, instance: AllocationUser, **kwargs: object
) -> None:
    """Add or remove members from an ldap group based on AllocationUser.status.

    Note this signal invokes a background task to do the actual LDAP operation. This
    leaves the potential for the database and LDAP to get out of sync if the task
    fails, but avoids making the request/response cycle slow.

    Args:
        sender: The model class.
        instance: The instance being saved.
        **kwargs: Additional keyword arguments.
    """
    if not settings.LDAP_ENABLED:
        return

    if (group_id := _get_shortname_from_allocation(instance.allocation)) is None:
        return

    if instance.status.name == "Active":
        async_task(
            ldap_add_member_to_group,
            group_id,
            instance.user.username,
            allow_already_present=True,
        )
    else:
        async_task(
            ldap_remove_member_from_group,
            group_id,
            instance.user.username,
            allow_missing=True,
        )