-
-
Notifications
You must be signed in to change notification settings - Fork 45
/
models.py
146 lines (124 loc) · 4.63 KB
/
models.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
import json
import logging
import numbers
import typing
from datetime import datetime, timedelta, timezone
from typing import (
Any,
Dict,
Optional,
Union,
)
import iso8601
logger = logging.getLogger(__name__)
Number = Union[int, float]
Id = Optional[Union[int, str]]
ConvertibleTimestamp = Union[datetime, str]
Duration = Union[timedelta, Number]
Data = Dict[str, Any]
def _timestamp_parse(ts_in: ConvertibleTimestamp) -> datetime:
"""
Takes something representing a timestamp and
returns a timestamp in the representation we want.
"""
ts = iso8601.parse_date(ts_in) if isinstance(ts_in, str) else ts_in
# Set resolution to milliseconds instead of microseconds
# (Fixes incompability with software based on unix time, for example mongodb)
ts = ts.replace(microsecond=int(ts.microsecond / 1000) * 1000)
# Add timezone if not set
if not ts.tzinfo:
# Needed? All timestamps should be iso8601 so ought to always contain timezone.
# Yes, because it is optional in iso8601
logger.warning(f"timestamp without timezone found, using UTC: {ts}")
ts = ts.replace(tzinfo=timezone.utc)
return ts
class Event(dict):
"""
Used to represents an event.
"""
def __init__(
self,
id: Optional[Id] = None,
timestamp: Optional[ConvertibleTimestamp] = None,
duration: Duration = 0,
data: Data = dict(),
) -> None:
self.id = id
if timestamp is None:
logger.warning(
"Event initializer did not receive a timestamp argument, "
"using now as timestamp"
)
# FIXME: The typing.cast here was required for mypy to shut up, weird...
self.timestamp = datetime.now(typing.cast(timezone, timezone.utc))
else:
# The conversion needs to be explicit here for mypy to pick it up
# (lacks support for properties)
self.timestamp = _timestamp_parse(timestamp)
self.duration = duration # type: ignore
self.data = data
def __eq__(self, other: object) -> bool:
if isinstance(other, Event):
return (
self.timestamp == other.timestamp
and self.duration == other.duration
and self.data == other.data
)
else:
raise TypeError(
"operator not supported between instances of '{}' and '{}'".format(
type(self), type(other)
)
)
def __lt__(self, other: object) -> bool:
if isinstance(other, Event):
return self.timestamp < other.timestamp
else:
raise TypeError(
"operator not supported between instances of '{}' and '{}'".format(
type(self), type(other)
)
)
def to_json_dict(self) -> dict:
"""Useful when sending data over the wire.
Any mongodb interop should not use do this as it accepts datetimes."""
json_data = self.copy()
json_data["timestamp"] = self.timestamp.astimezone(timezone.utc).isoformat()
json_data["duration"] = self.duration.total_seconds()
return json_data
def to_json_str(self) -> str:
data = self.to_json_dict()
return json.dumps(data)
def _hasprop(self, propname: str) -> bool:
"""Badly named, but basically checks if the underlying
dict has a prop, and if it is a non-empty list"""
return propname in self and self[propname] is not None
@property
def id(self) -> Id:
return self["id"] if self._hasprop("id") else None
@id.setter
def id(self, id: Id) -> None:
self["id"] = id
@property
def data(self) -> dict:
return self["data"] if self._hasprop("data") else {}
@data.setter
def data(self, data: dict) -> None:
self["data"] = data
@property
def timestamp(self) -> datetime:
return self["timestamp"]
@timestamp.setter
def timestamp(self, timestamp: ConvertibleTimestamp) -> None:
self["timestamp"] = _timestamp_parse(timestamp).astimezone(timezone.utc)
@property
def duration(self) -> timedelta:
return self["duration"] if self._hasprop("duration") else timedelta(0)
@duration.setter
def duration(self, duration: Duration) -> None:
if isinstance(duration, timedelta):
self["duration"] = duration
elif isinstance(duration, numbers.Real):
self["duration"] = timedelta(seconds=duration) # type: ignore
else:
raise TypeError(f"Couldn't parse duration of invalid type {type(duration)}")