querysets

django_spire.history.querysets

log = logging.getLogger(__name__) module-attribute

HistoryQuerySet

Bases: QuerySet

active

Source code in django_spire/history/querysets.py

def active(self) -> Self:
    return self.filter(is_active=True, is_deleted=False)

inactive

Source code in django_spire/history/querysets.py

def inactive(self) -> Self:
    return self.filter(is_active=False, is_deleted=False)

deleted

Source code in django_spire/history/querysets.py

def deleted(self) -> Self:
    return self.filter(is_deleted=True)

not_deleted

Source code in django_spire/history/querysets.py

def not_deleted(self) -> Self:
    return self.filter(is_deleted=False)

bulk_create

Source code in django_spire/history/querysets.py

def bulk_create(
    self,
    objs: Iterable[models.Model],
    *,
    batch_size: int | None = None,
    ignore_conflicts: bool = False,
    update_conflicts: bool = False,
    update_fields: Iterable[str] | None = None,
    unique_fields: Iterable[str] | None = None,
) -> list[models.Model]:
    if not self._activity_enabled():
        return super().bulk_create(
            objs,
            batch_size=batch_size,
            ignore_conflicts=ignore_conflicts,
            update_conflicts=update_conflicts,
            update_fields=update_fields,
            unique_fields=unique_fields,
        )

    with transaction.atomic(using=self.db):
        created = super().bulk_create(
            objs,
            batch_size=batch_size,
            ignore_conflicts=ignore_conflicts,
            update_conflicts=update_conflicts,
            update_fields=update_fields,
            unique_fields=unique_fields,
        )

        if update_conflicts:
            log.warning(
                'bulk_create with update_conflicts cannot distinguish created '
                'rows from updated rows; no activity records were created.'
            )
        else:
            self._add_bulk_activity(created, ActivityVerb.CREATED)

    return created

bulk_update

Source code in django_spire/history/querysets.py

def bulk_update(
    self,
    objs: Iterable[models.Model],
    fields: Iterable[str],
    *args,
    **kwargs
) -> int:
    obj_list = list(objs)

    if not self._activity_enabled():
        return super().bulk_update(obj_list, fields, *args, **kwargs)

    field_names = set(fields)
    token = _bulk_update_active.set(True)

    try:
        with transaction.atomic(using=self.db):
            matched_pks = self._activity_matched_pks(obj_list)

            previously_deleted_pks = self._activity_previously_deleted_pks(
                matched_pks,
                field_names,
            )

            updated_count = super().bulk_update(obj_list, fields, *args, **kwargs)
            matched_objs = self._activity_matched_objs(obj_list, matched_pks)

            if 'is_deleted' in field_names:
                deleted_objs = [
                    obj
                    for obj in matched_objs
                    if obj.is_deleted and obj.pk not in previously_deleted_pks
                ]
            else:
                deleted_objs = []

            deleted_obj_pks = {obj.pk for obj in deleted_objs}
            updated_objs = [obj for obj in matched_objs if obj.pk not in deleted_obj_pks]

            if updated_objs:
                self._add_bulk_activity(updated_objs, ActivityVerb.UPDATED)

            if deleted_objs:
                self._add_bulk_activity(deleted_objs, ActivityVerb.DELETED)
    finally:
        _bulk_update_active.reset(token)

    return updated_count

delete

Source code in django_spire/history/querysets.py

def delete(self, *args, **kwargs) -> tuple[int, dict[str, int]]:
    user = get_current_user()

    if user is None:
        return super().delete(*args, **kwargs)

    token = start_delete_activity_collection()

    try:
        with transaction.atomic(using=self.db):
            result = super().delete(*args, **kwargs)
            entries = get_delete_activity_entries() or []

            if entries:
                self._add_bulk_delete_activity(entries, user)
    finally:
        reset_delete_activity_collection(token)

    return result

update

Source code in django_spire/history/querysets.py

def update(self, **kwargs) -> int:
    if _bulk_update_active.get() or not kwargs:
        return super().update(**kwargs)
    if not self._activity_enabled():
        return super().update(**kwargs)
    if self.query.is_sliced:
        return super().update(**kwargs)

    with transaction.atomic(using=self.db):
        pks = self._activity_update_target_pks()
        not_deleted_pks = self._activity_not_deleted_pks(pks, kwargs)
        updated_count = super().update(**kwargs)

        if updated_count < len(pks):
            log.warning(
                'update() modified %d rows but snapshotted %d rows for '
                'activity records; the audit trail may include rows the '
                'update did not modify.',
                updated_count,
                len(pks),
            )

        if pks:
            instances = list(self.model._base_manager.using(self.db).filter(pk__in=pks))

            deleted_instances = [
                obj
                for obj in instances
                if getattr(obj, 'is_deleted', False) and obj.pk in not_deleted_pks
            ]

            deleted_pks = {obj.pk for obj in deleted_instances}
            updated_instances = [obj for obj in instances if obj.pk not in deleted_pks]

            if updated_instances:
                self._add_bulk_activity(updated_instances, ActivityVerb.UPDATED)

            if deleted_instances:
                self._add_bulk_activity(deleted_instances, ActivityVerb.DELETED)

    return updated_count