-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathobj_update.py
124 lines (102 loc) · 3.68 KB
/
obj_update.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
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
93
94
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
"""
`dj-obj-update` helps you update an object but only saves if something changed.
"""
import datetime as dt
import logging
from operator import itemgetter
from unittest.mock import sentinel
__all__ = ["obj_update", "obj_update_or_create"]
__version__ = "0.6.0"
DIRTY = "_is_dirty"
UNSET = sentinel.UNSET
create_logger = logging.getLogger("obj_update.create")
update_logger = logging.getLogger("obj_update.update")
def datetime_repr(value):
if isinstance(value, dt.datetime):
return value.isoformat().replace("+00:00", "Z")
return str(value).replace(" ", "T")
def set_field(obj, field_name, value):
"""Fancy setattr with debugging."""
old = getattr(obj, field_name)
field = obj._meta.get_field(field_name)
# is_relation is Django 1.8 only
if field.is_relation:
# If field_name is the `_id` field, then there is no 'pk' attr and
# old/value *is* the pk
old_repr = None if old is None else getattr(old, "pk", old)
new_repr = None if value is None else getattr(value, "pk", value)
elif field.__class__.__name__ == "DateTimeField":
old_repr = None if old is None else datetime_repr(old)
new_repr = None if value is None else datetime_repr(value)
else:
old_repr = None if old is None else str(old)
new_repr = None if value is None else str(value)
if old_repr != new_repr:
setattr(obj, field_name, value)
if not hasattr(obj, DIRTY):
setattr(obj, DIRTY, [])
getattr(obj, DIRTY).append(
dict(field_name=field_name, old_value=old_repr, new_value=new_repr)
)
def human_log_formatter(dirty_data):
return "".join(
"[{field_name} {old_value}->{new_value}]".format(**x) for x in dirty_data
)
def json_log_formatter(dirty_data):
return {
x["field_name"]: {"old": x["old_value"], "new": x["new_value"]}
for x in dirty_data
}
def obj_update(obj, data: dict, *, update_fields=UNSET, save: bool = True) -> bool:
"""
Fancy way to update `obj` with `data` dict.
Parameters
----------
obj : Django model instance
data
The data to update ``obj`` with
update_fields
Use your ``update_fields`` instead of our generated one. If you need
an auto_now or auto_now_add field to get updated, set this to ``None``
to get the default Django behavior.
save
If save=False, then don't actually save. This can be useful if you
just want to utilize the verbose logging.
DEPRECRATED in favor of the more standard ``update_fields=[]``
Returns
-------
bool
True if data changed
"""
for field_name, value in data.items():
set_field(obj, field_name, value)
dirty_data = getattr(obj, DIRTY, None)
if not dirty_data:
return False
update_logger.debug(
human_log_formatter(dirty_data),
extra={
"model": obj._meta.object_name,
"pk": obj.pk,
"changes": json_log_formatter(dirty_data),
},
)
if update_fields == UNSET:
update_fields = list(map(itemgetter("field_name"), dirty_data))
if not save:
update_fields = ()
obj.save(update_fields=update_fields)
delattr(obj, DIRTY)
return True
def obj_update_or_create(model, defaults=None, update_fields=UNSET, **kwargs):
"""
Mimic queryset.update_or_create but using obj_update.
"""
obj, created = model.objects.get_or_create(defaults=defaults, **kwargs)
if created:
create_logger.debug(
"Created %s %s", model._meta.object_name, obj.pk, extra={"pk": obj.pk}
)
else:
obj_update(obj, defaults, update_fields=update_fields)
return obj, created