Skip to content

utils

django_spire.history.activity.utils

BULK_ACTIVITY_COUNT_MAX = 100000 module-attribute

log = logging.getLogger(__name__) module-attribute

actor_name

Source code in django_spire/history/activity/utils.py 
28
29
def actor_name(user: User) -> str:
    return user.get_full_name() or user.get_username()

build_activity_information

Source code in django_spire/history/activity/utils.py 
32
33
def build_activity_information(instance: ActivityMixin, user: User, verb: str) -> str:
    return f'{actor_name(user)} {verb} {instance._meta.verbose_name} "{instance}".'

add_activity

Source code in django_spire/history/activity/utils.py 
36
37
38
def add_activity(instance: ActivityMixin, user: User, verb: str) -> Activity:
    information = build_activity_information(instance, user, verb)
    return instance.add_activity(user=user, verb=verb, information=information)

add_bulk_activity

Source code in django_spire/history/activity/utils.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def add_bulk_activity(
    instances: Iterable[ActivityMixin],
    user: User,
    verb: str,
    using: str | None = None
) -> list[Activity]:
    bounded_instances = list(itertools.islice(iter(instances), BULK_ACTIVITY_COUNT_MAX + 1))

    if len(bounded_instances) > BULK_ACTIVITY_COUNT_MAX:
        log.warning(
            'add_bulk_activity truncated "%s" activity records to the %d row cap; '
            'the audit trail for this operation is incomplete.',
            verb,
            BULK_ACTIVITY_COUNT_MAX,
        )

        bounded_instances = bounded_instances[:BULK_ACTIVITY_COUNT_MAX]

    instance_list = [
        instance
        for instance in bounded_instances
        if instance.pk is not None
    ]

    skipped_count = len(bounded_instances) - len(instance_list)

    if skipped_count > 0:
        log.warning(
            'add_bulk_activity skipped %d "%s" activity records for instances without '
            'primary keys (unsaved objects, bulk_create with ignore_conflicts, or a '
            'database backend that cannot return inserted rows).',
            skipped_count,
            verb,
        )

    if not instance_list:
        return []

    content_type = ContentType.objects.get_for_model(instance_list[0])

    activities = [
        Activity(
            content_type=content_type,
            object_id=instance.pk,
            user=user,
            verb=verb,
            information=build_activity_information(instance, user, verb)
        )
        for instance in instance_list
    ]

    return _activity_manager(using).bulk_create(activities)

add_bulk_delete_activity

Source code in django_spire/history/activity/utils.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def add_bulk_delete_activity(
    entries: Iterable[DeleteActivityEntry],
    user: User,
    using: str | None = None
) -> list[Activity]:
    entry_list = list(itertools.islice(iter(entries), BULK_ACTIVITY_COUNT_MAX + 1))

    if len(entry_list) > BULK_ACTIVITY_COUNT_MAX:
        log.warning(
            'add_bulk_delete_activity truncated "deleted" activity records to the '
            '%d row cap; the audit trail for this operation is incomplete.',
            BULK_ACTIVITY_COUNT_MAX,
        )

        entry_list = entry_list[:BULK_ACTIVITY_COUNT_MAX]

    if not entry_list:
        return []

    activities = [
        Activity(
            content_type=ContentType.objects.get_for_model(model),
            object_id=object_id,
            user=user,
            verb=ActivityVerb.DELETED,
            information=information
        )
        for model, object_id, information in entry_list
    ]

    return _activity_manager(using).bulk_create(activities)

add_form_activity

Source code in django_spire/history/activity/utils.py 
128
129
130
def add_form_activity(model_object: ActivityMixin, pk: int | bool, user: User) -> None:
    verb = ActivityVerb.UPDATED if pk else ActivityVerb.CREATED
    add_activity(model_object, user, verb)