Utilities

django-guardian helper functions.

Unstable API

Functions defined within this module should be considered as django-guardian’s internal functionality. They are not guaranteed to be stable, and their APIs may change in any future releases.

get_anonymous_user()

Get the django-guardian equivalent of the anonymous user.

It returns a User model instance (not AnonymousUser) depending on ANONYMOUS_USER_NAME configuration.

This function can be cached to avoid repetitive database queries based on the GUARDIAN_ANONYMOUS_USER_CACHE_TTL setting: - 0 (default): No caching, each call performs a fresh database query - Positive number: Cache for that many seconds - -1: Cache indefinitely (not recommended)

See Also

See the configuration docs that explain that the Guardian anonymous user is not equivalent to Django's AnonymousUser.

• Guardian Configuration
• ANONYMOUS_USER_NAME configuration
• ANONYMOUS_USER_CACHE_TTL configuration

Source code in guardian/utils.py

def get_anonymous_user() -> Any:
    """Get the django-guardian equivalent of the anonymous user.

    It returns a `User` model instance (not `AnonymousUser`) depending on
    `ANONYMOUS_USER_NAME` configuration.

    This function can be cached to avoid repetitive database queries based on the
    `GUARDIAN_ANONYMOUS_USER_CACHE_TTL` setting:
    - 0 (default): No caching, each call performs a fresh database query
    - Positive number: Cache for that many seconds
    - -1: Cache indefinitely (not recommended)

    See Also:
        See the configuration docs that explain that the Guardian anonymous user is
        not equivalent to Django's AnonymousUser.

        - [Guardian Configuration](https://django-guardian.readthedocs.io/en/stable/configuration.html)
        - [ANONYMOUS_USER_NAME configuration](https://django-guardian.readthedocs.io/en/stable/configuration.html#anonymous-user-nam)
        - [ANONYMOUS_USER_CACHE_TTL configuration](https://django-guardian.readthedocs.io/en/stable/configuration.html#anonymous-user-cache-ttl)
    """
    if guardian_settings.ANONYMOUS_USER_CACHE_TTL > 0 or guardian_settings.ANONYMOUS_USER_CACHE_TTL == -1:
        return _get_anonymous_user_cached()
    else:
        return _get_anonymous_user_uncached()

get_identity(identity)

Get a tuple with the identity of the given input.

Returns a tuple with one of the members set to None depending on whether the input is a Group instance or a User instance. Also accepts AnonymousUser instance but would return User instead. It is convenient and needed for authorization backend to support anonymous users.

Returns:

Type	Description
tuple[Union[Any, None], Union[Any, None]]	Either (user_obj, None) or (None, group_obj) depending on the input type.

Parameters:

Name	Type	Description	Default
identity	User | Group	Instance of User or Group to get identity from.	required

Raises:

Type	Description
NotUserNorGroup	If the function cannot return proper identity instance

Examples:

>>> from django.contrib.auth.models import User, Group
>>> user = User.objects.create(username='joe')
>>> get_identity(user)
(<User: joe>, None)

>>> group = Group.objects.create(name='users')
>>> get_identity(group)
(None, <Group: users>)

>>> anon = AnonymousUser()
>>> get_identity(anon)
(<User: AnonymousUser>, None)

>>> get_identity("not instance")
...
NotUserNorGroup: User/AnonymousUser or Group instance is required (got )

Source code in guardian/utils.py

def get_identity(identity: Model) -> tuple[Union[Any, None], Union[Any, None]]:
    """Get a tuple with the identity of the given input.

    Returns a tuple with one of the members set to `None` depending on whether the input is
    a `Group` instance or a `User` instance.
    Also accepts AnonymousUser instance but would return `User` instead.
    It is convenient and needed for authorization backend to support anonymous users.

    Returns:
         Either (user_obj, None) or (None, group_obj) depending on the input type.

    Parameters:
        identity (User | Group): Instance of `User` or `Group` to get identity from.

    Raises:
        NotUserNorGroup: If the function cannot return proper identity instance

    Examples:
        ```shell
        >>> from django.contrib.auth.models import User, Group
        >>> user = User.objects.create(username='joe')
        >>> get_identity(user)
        (<User: joe>, None)

        >>> group = Group.objects.create(name='users')
        >>> get_identity(group)
        (None, <Group: users>)

        >>> anon = AnonymousUser()
        >>> get_identity(anon)
        (<User: AnonymousUser>, None)

        >>> get_identity("not instance")
        ...
        NotUserNorGroup: User/AnonymousUser or Group instance is required (got )
        ```
    """
    if isinstance(identity, AnonymousUser):
        identity = get_anonymous_user()

    group_model = get_group_obj_perms_model().group.field.related_model  # type: ignore[attr-defined]

    # get identity from queryset model type
    if isinstance(identity, QuerySet):
        identity_model_type = identity.model
        if issubclass(identity_model_type, get_user_model()):
            return identity, None
        elif issubclass(identity_model_type, group_model):
            return None, identity

    # get identity from the first element in the list
    if isinstance(identity, list):
        if not identity:
            return None, None
        if isinstance(identity[0], get_user_model()):
            return identity, None
        if isinstance(identity[0], group_model):
            return None, identity

    if isinstance(identity, get_user_model()):
        return identity, None
    if isinstance(identity, group_model):
        return None, identity

    raise NotUserNorGroup("User/AnonymousUser or Group instance is required (got %s)" % identity)

clean_orphan_obj_perms(batch_size=None, max_batches=None, max_duration_secs=None, skip_batches=0)

Removes orphan object permissions using queryset slice-based batching, batch skipping, batch limit, and time-based interruption.

Source code in guardian/utils.py

def clean_orphan_obj_perms(
    batch_size: Optional[int] = None,
    max_batches: Optional[int] = None,
    max_duration_secs: Optional[int] = None,
    skip_batches: int = 0,
) -> int:
    """
    Removes orphan object permissions using queryset slice-based batching,
    batch skipping, batch limit, and time-based interruption.
    """

    UserObjectPermission = get_user_obj_perms_model()
    GroupObjectPermission = get_group_obj_perms_model()

    deleted = 0
    scanned = 0
    processed_batches = 0
    batch_count = 0
    start_time = time.monotonic()

    if batch_size is None:
        all_objs = list(
            chain(
                UserObjectPermission.objects.order_by("pk"),
                GroupObjectPermission.objects.order_by("pk"),
            )
        )

        for obj in all_objs:
            if max_duration_secs is not None and (time.monotonic() - start_time >= max_duration_secs):
                logger.info(f"Time limit of {max_duration_secs}s reached.")
                break

            scanned += 1
            if obj.content_object is None:
                logger.debug("Removing %s (pk=%d)", obj, obj.pk)
                obj.delete()
                deleted += 1
        processed_batches = 1
    else:
        total_user = UserObjectPermission.objects.count()
        total_group = GroupObjectPermission.objects.count()
        total_records = total_user + total_group

        total_batches_possible = ceil(total_records / batch_size)

        remaining_batches = total_batches_possible - skip_batches
        if max_batches is not None:
            remaining_batches = min(remaining_batches, max_batches)

        logger.info(
            f"Starting orphan object permissions cleanup with batch_size={batch_size}, "
            f"max_batches={max_batches}, max_duration_secs={max_duration_secs}, skip_batches={skip_batches}"
        )

        # Process User permissions first
        user_processed = 0
        user_total = UserObjectPermission.objects.count()

        # Skip user batches if needed
        user_skip_records = min(skip_batches * batch_size, user_total)
        user_remaining = user_total - user_skip_records

        while user_remaining > 0 and remaining_batches > 0:
            if max_duration_secs is not None and (time.monotonic() - start_time >= max_duration_secs):
                logger.info(f"Time limit of {max_duration_secs}s reached.")
                break

            gc.collect()

            current_batch_size = min(batch_size, user_remaining)
            user_batch = list(
                UserObjectPermission.objects.order_by("pk")[
                    user_skip_records + user_processed : user_skip_records + user_processed + current_batch_size
                ]
            )

            if not user_batch:
                break

            scanned += len(user_batch)
            user_processed += len(user_batch)
            user_remaining -= len(user_batch)

            orphan_pks = [obj.pk for obj in user_batch if obj.content_object is None]

            if orphan_pks:
                logger.info(
                    f"!!! Found {len(orphan_pks)} orphan user permissions in batch {processed_batches + 1}. !!!"
                )
                UserObjectPermission.objects.filter(pk__in=orphan_pks).delete()
                deleted += len(orphan_pks)

            processed_batches += 1
            batch_count += 1
            remaining_batches -= 1

            logger.info(f"Processed user batch {processed_batches}, scanned {scanned} objects.")

        # Process Group permissions
        if remaining_batches > 0:
            group_processed = 0
            group_total = GroupObjectPermission.objects.count()

            # Calculate skip for group permissions
            total_user_batches = ceil(user_total / batch_size) if user_total > 0 else 0
            group_skip_batches = max(0, skip_batches - total_user_batches)
            group_skip_records = min(group_skip_batches * batch_size, group_total)
            group_remaining = group_total - group_skip_records

            while group_remaining > 0 and remaining_batches > 0:
                if max_duration_secs is not None and (time.monotonic() - start_time >= max_duration_secs):
                    logger.info(f"Time limit of {max_duration_secs}s reached.")
                    break

                gc.collect()

                current_batch_size = min(batch_size, group_remaining)
                group_batch = list(
                    GroupObjectPermission.objects.order_by("pk")[
                        group_skip_records + group_processed : group_skip_records + group_processed + current_batch_size
                    ]
                )

                if not group_batch:
                    break

                scanned += len(group_batch)
                group_processed += len(group_batch)
                group_remaining -= len(group_batch)

                orphan_pks = [obj.pk for obj in group_batch if obj.content_object is None]

                if orphan_pks:
                    logger.info(
                        f"!!! Found {len(orphan_pks)} orphan group permissions in batch {processed_batches + 1}. !!!"
                    )
                    GroupObjectPermission.objects.filter(pk__in=orphan_pks).delete()
                    deleted += len(orphan_pks)

                processed_batches += 1
                batch_count += 1
                remaining_batches -= 1

                logger.info(f"Processed group batch {processed_batches}, scanned {scanned} objects.")

    logger.info(
        f"Finished orphan object permissions cleanup. "
        f"Scanned: {scanned} | Removed: {deleted} | "
        f"Batches processed: {processed_batches}"
    )

    if batch_size:
        suggestion = (
            f"To resume cleanup, call:\n"
            f"clean_orphan_obj_perms(batch_size={batch_size}, "
            f"skip_batches={skip_batches + processed_batches}, "
        )
        if max_batches is not None:
            suggestion += f"max_batches={max_batches - batch_count}, "
        if max_duration_secs is not None:
            suggestion += f"max_duration_secs={max_duration_secs}, "
        suggestion = suggestion.rstrip(", ") + ")"
        logger.info(suggestion)

    return deleted