-
-
Notifications
You must be signed in to change notification settings - Fork 606
/
payload.py
154 lines (120 loc) · 4.86 KB
/
payload.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
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
# Copyright 2014 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).
from hashlib import sha1
from pants.util.strutil import ensure_binary
class PayloadFieldAlreadyDefinedError(Exception):
pass
class PayloadFrozenError(Exception):
pass
class Payload:
"""A mapping from field names to PayloadField instances.
A Target will add PayloadFields to its Payload until instantiation is finished, at which point
freeze() will be called and make the Payload immutable.
:API: public
"""
def __init__(self):
self._fields = {}
self._frozen = False
self._fingerprint_memo_map = {}
@property
def fields(self):
return list(self._fields.items())
def as_dict(self):
"""Return the Payload object as a dict."""
return {k: self.get_field_value(k) for k in self._fields}
def freeze(self):
"""Permanently make this Payload instance immutable.
No more fields can be added after calling freeze().
:API: public
"""
self._frozen = True
def get_field(self, key, default=None):
"""An alternative to attribute access for duck typing Payload instances.
Has the same semantics as dict.get, and in fact just delegates to the underlying field mapping.
:API: public
"""
return self._fields.get(key, default)
def get_field_value(self, key, default=None):
"""Retrieves the value in the payload field if the field exists, otherwise returns the
default.
:API: public
"""
if key in self._fields:
payload_field = self._fields[key]
if payload_field:
return payload_field.value
return default
def add_fields(self, field_dict):
"""Add a mapping of field names to PayloadField instances.
:API: public
"""
for key, field in field_dict.items():
self.add_field(key, field)
def add_field(self, key, field):
"""Add a field to the Payload.
:API: public
:param string key: The key for the field. Fields can be accessed using attribute access as
well as `get_field` using `key`.
:param PayloadField field: A PayloadField instance. None is an allowable value for `field`,
in which case it will be skipped during hashing.
"""
if key in self._fields:
raise PayloadFieldAlreadyDefinedError(
"Key {key} is already set on this payload. The existing field was {existing_field}."
" Tried to set new field {field}.".format(
key=key, existing_field=self._fields[key], field=field
)
)
elif self._frozen:
raise PayloadFrozenError(
"Payload is frozen, field with key {key} cannot be added to it.".format(key=key)
)
else:
self._fields[key] = field
self._fingerprint_memo = None
def fingerprint(self, field_keys=None):
"""A memoizing fingerprint that rolls together the fingerprints of underlying PayloadFields.
If no fields were hashed (or all fields opted out of being hashed by returning `None`), then
`fingerprint()` also returns `None`.
:param iterable<string> field_keys: A subset of fields to use for the fingerprint. Defaults
to all fields.
"""
field_keys = frozenset(field_keys or self._fields.keys())
if field_keys not in self._fingerprint_memo_map:
self._fingerprint_memo_map[field_keys] = self._compute_fingerprint(field_keys)
return self._fingerprint_memo_map[field_keys]
def _compute_fingerprint(self, field_keys):
hasher = sha1()
empty_hash = True
for key in sorted(field_keys):
field = self._fields[key]
if field is not None:
fp = field.fingerprint()
if fp is not None:
empty_hash = False
fp = ensure_binary(fp)
key = ensure_binary(key)
key_sha1 = sha1(key).hexdigest().encode()
hasher.update(key_sha1)
hasher.update(fp)
if empty_hash:
return None
else:
return hasher.hexdigest()
def mark_dirty(self):
"""Invalidates memoized fingerprints for this payload.
Exposed for testing.
:API: public
"""
self._fingerprint_memo_map = {}
for field in self._fields.values():
field.mark_dirty()
def __getattr__(self, attr):
try:
field = self._fields[attr]
except KeyError:
raise AttributeError(attr)
if field is not None:
return field.value
else:
return None