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
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
|
import datetime
import importlib
import json
try:
from zoneinfo import ZoneInfo
except ImportError:
from backports.zoneinfo import ZoneInfo
from .lazy import LazyUUIDTaskSet, LazyUUIDTask
DATE_FORMAT = '%Y%m%dT%H%M%SZ'
class SerializingObject(object):
"""
Common ancestor for TaskResource & TaskWarriorFilter, since they both
need to serialize arguments.
Serializing method should hold the following contract:
- any empty value (meaning removal of the attribute)
is deserialized into a empty string
- None denotes a empty value for any attribute
Deserializing method should hold the following contract:
- None denotes a empty value for any attribute (however,
this is here as a safeguard, TaskWarrior currently does
not export empty-valued attributes) if the attribute
is not iterable (e.g. list or set), in which case
a empty iterable should be used.
Normalizing methods should hold the following contract:
- They are used to validate and normalize the user input.
Any attribute value that comes from the user (during Task
initialization, assignign values to Task attributes, or
filtering by user-provided values of attributes) is first
validated and normalized using the normalize_{key} method.
- If validation or normalization fails, normalizer is expected
to raise ValueError.
"""
def __init__(self, backend):
self.backend = backend
def _deserialize(self, key, value):
hydrate_func = getattr(self, 'deserialize_{0}'.format(key),
lambda x: x if x != '' else None)
return hydrate_func(value)
def _serialize(self, key, value):
dehydrate_func = getattr(self, 'serialize_{0}'.format(key),
lambda x: x if x is not None else '')
return dehydrate_func(value)
def _normalize(self, key, value):
"""
Use normalize_<key> methods to normalize user input. Any user
input will be normalized at the moment it is used as filter,
or entered as a value of Task attribute.
"""
# None value should not be converted by normalizer
if value is None:
return None
normalize_func = getattr(self, 'normalize_{0}'.format(key),
lambda x: x)
return normalize_func(value)
def timestamp_serializer(self, date):
if not date:
return ''
# Any serialized timestamp should be localized, we need to
# convert to UTC before converting to string (DATE_FORMAT uses UTC)
date = date.astimezone(ZoneInfo('UTC'))
return date.strftime(DATE_FORMAT)
def timestamp_deserializer(self, date_str):
if not date_str:
return None
# Return timestamp localized in the local zone
naive_timestamp = datetime.datetime.strptime(date_str, DATE_FORMAT)
localized_timestamp = naive_timestamp.replace(tzinfo=ZoneInfo('UTC'))
return localized_timestamp.astimezone()
def serialize_entry(self, value):
return self.timestamp_serializer(value)
def deserialize_entry(self, value):
return self.timestamp_deserializer(value)
def normalize_entry(self, value):
return self.datetime_normalizer(value)
def serialize_modified(self, value):
return self.timestamp_serializer(value)
def deserialize_modified(self, value):
return self.timestamp_deserializer(value)
def normalize_modified(self, value):
return self.datetime_normalizer(value)
def serialize_start(self, value):
return self.timestamp_serializer(value)
def deserialize_start(self, value):
return self.timestamp_deserializer(value)
def normalize_start(self, value):
return self.datetime_normalizer(value)
def serialize_end(self, value):
return self.timestamp_serializer(value)
def deserialize_end(self, value):
return self.timestamp_deserializer(value)
def normalize_end(self, value):
return self.datetime_normalizer(value)
def serialize_due(self, value):
return self.timestamp_serializer(value)
def deserialize_due(self, value):
return self.timestamp_deserializer(value)
def normalize_due(self, value):
return self.datetime_normalizer(value)
def serialize_scheduled(self, value):
return self.timestamp_serializer(value)
def deserialize_scheduled(self, value):
return self.timestamp_deserializer(value)
def normalize_scheduled(self, value):
return self.datetime_normalizer(value)
def serialize_until(self, value):
return self.timestamp_serializer(value)
def deserialize_until(self, value):
return self.timestamp_deserializer(value)
def normalize_until(self, value):
return self.datetime_normalizer(value)
def serialize_wait(self, value):
return self.timestamp_serializer(value)
def deserialize_wait(self, value):
return self.timestamp_deserializer(value)
def normalize_wait(self, value):
return self.datetime_normalizer(value)
def serialize_annotations(self, value):
value = value if value is not None else []
# This may seem weird, but it's correct, we want to export
# a list of dicts as serialized value
serialized_annotations = [json.loads(annotation.export_data())
for annotation in value]
return serialized_annotations if serialized_annotations else ''
def deserialize_annotations(self, data):
task_module = importlib.import_module('tasklib.task')
TaskAnnotation = getattr(task_module, 'TaskAnnotation')
return [TaskAnnotation(self, d) for d in data] if data else []
def serialize_tags(self, tags):
return ','.join(tags) if tags else ''
def deserialize_tags(self, tags):
if isinstance(tags, str):
return set(tags.split(',')) if tags else set()
return set(tags or [])
def serialize_parent(self, parent):
return parent['uuid'] if parent else ''
def deserialize_parent(self, uuid):
return LazyUUIDTask(self.backend, uuid) if uuid else None
def serialize_depends(self, value):
# Return the list of uuids
value = value if value is not None else set()
if isinstance(value, LazyUUIDTaskSet):
return ','.join(value._uuids)
else:
return ','.join(task['uuid'] for task in value)
def deserialize_depends(self, raw_uuids):
raw_uuids = raw_uuids or [] # Convert None to empty list
if not raw_uuids:
return set()
# TW 2.4.4 encodes list of dependencies as a single string
if type(raw_uuids) is not list:
uuids = raw_uuids.split(',')
# TW 2.4.5 and later exports them as a list, no conversion needed
else:
uuids = raw_uuids
return LazyUUIDTaskSet(self.backend, uuids)
def datetime_normalizer(self, value):
"""
Normalizes date/datetime value (considered to come from user input)
to localized datetime value. Following conversions happen:
naive date -> localized datetime with the same date, and time=midnight
naive datetime -> localized datetime with the same value
localized datetime -> localized datetime (no conversion)
"""
if (
isinstance(value, datetime.date)
and not isinstance(value, datetime.datetime)
):
# Convert to local midnight
value_full = datetime.datetime.combine(value, datetime.time.min)
localized = value_full.astimezone()
elif isinstance(value, datetime.datetime):
if value.tzinfo is None:
# Convert to localized datetime object
localized = value.astimezone()
else:
# If the value is already localized, there is no need to change
# time zone at this point. Also None is a valid value too.
localized = value
elif isinstance(value, str):
localized = self.backend.convert_datetime_string(value)
else:
raise ValueError("Provided value could not be converted to "
"datetime, its type is not supported: {}"
.format(type(value)))
return localized
def normalize_uuid(self, value):
# Enforce sane UUID
if not isinstance(value, str) or value == '':
raise ValueError("UUID must be a valid non-empty string, "
"not: {}".format(value))
return value
|
