hat.event.common

16class BackendClosedError(Exception):
17    """Backend closed"""

20class Backend(aio.Resource):
21    """Backend ABC"""
22
23    @abc.abstractmethod
24    async def get_last_event_id(self,
25                                server_id: int
26                                ) -> EventId:
27        """Get last registered event id associated with server id"""
28
29    @abc.abstractmethod
30    async def register(self,
31                       events: Collection[Event]
32                       ) -> Collection[Event] | None:
33        """Register events"""
34
35    @abc.abstractmethod
36    async def query(self,
37                    params: QueryParams
38                    ) -> QueryResult:
39        """Query events"""
40
41    @abc.abstractmethod
42    async def flush(self):
43        """Flush internal buffers and permanently persist events"""

67class BackendInfo(typing.NamedTuple):
68    """Backend info
69
70    Backend is implemented as python module which is dynamically imported.
71    It is expected that this module contains `info` which is instance of
72    `BackendInfo`.
73
74    If backend defines JSON schema repository and JSON schema id, JSON schema
75    repository will be used for additional validation of backend configuration
76    with JSON schema id.
77
78    """
79    create: CreateBackend
80    json_schema_id: str | None = None
81    json_schema_repo: json.SchemaRepository | None = None

84def import_backend_info(py_module_str: str) -> BackendInfo:
85    """Import backend info"""
86    py_module = importlib.import_module(py_module_str)
87    info = py_module.info
88
89    if not isinstance(info, BackendInfo):
90        raise Exception('invalid backend implementation')
91
92    return info

13class EventTypeCollection(abc.ABC, typing.Generic[T]):
14
15    @abc.abstractmethod
16    def __init__(self):
17        pass
18
19    @abc.abstractmethod
20    def add(self, subscription: Subscription, value: T):
21        pass
22
23    @abc.abstractmethod
24    def remove(self, value: T):
25        pass
26
27    @abc.abstractmethod
28    def get(self, event_type: EventType) -> Iterable[T]:
29        pass

 5class ListEventTypeCollection(common.EventTypeCollection):
 6
 7    def __init__(self):
 8        self._values = {}
 9
10    def add(self, subscription, value):
11        self._values[value] = (self._values[value].union(subscription)
12                               if value in self._values else subscription)
13
14    def remove(self, value):
15        self._values.pop(value, None)
16
17    def get(self, event_type):
18        for value, subscription in self._values.items():
19            if subscription.matches(event_type):
20                yield value

 8class TreeEventTypeCollection(common.EventTypeCollection):
 9
10    def __init__(self):
11        self._root = _create_node()
12        self._value_nodes = collections.defaultdict(collections.deque)
13
14    def add(self, subscription, value):
15        for query_type in subscription.get_query_types():
16            node = self._root
17            rest = query_type
18
19            while rest:
20                head, rest = rest[0], rest[1:]
21                if head == '*' and rest:
22                    raise ValueError('invalid query type')
23
24                node = node.children[head]
25
26            if value in node.values:
27                return
28
29            node.values.add(value)
30            self._value_nodes[value].append(node)
31
32    def remove(self, value):
33        for node in self._value_nodes.pop(value, []):
34            node.values.remove(value)
35
36    def get(self, event_type):
37        return set(_get(self._root, event_type))

40class Timestamp(typing.NamedTuple):
41    s: int
42    """seconds since 1970-01-01 (can be negative)"""
43    us: int
44    """microseconds added to timestamp seconds in range [0, 1e6)"""
45
46    def add(self, s: float) -> 'Timestamp':
47        """Create new timestamp by adding seconds to existing timestamp"""
48        us = self.us + round((s - int(s)) * 1e6)
49        s = self.s + int(s)
50        return Timestamp(s=s + us // int(1e6),
51                         us=us % int(1e6))

61class Status(enum.Enum):
62    STANDBY = 'standby'
63    OPERATIONAL = 'operational'

66class Order(enum.Enum):
67    DESCENDING = 'descending'
68    ASCENDING = 'ascending'

71class OrderBy(enum.Enum):
72    TIMESTAMP = 'timestamp'
73    SOURCE_TIMESTAMP = 'sourceTimestamp'

76class EventId(typing.NamedTuple):
77    server: ServerId
78    session: SessionId
79    instance: InstanceId

82class EventPayloadBinary(typing.NamedTuple):
83    type: str
84    data: util.Bytes

87class EventPayloadJson(typing.NamedTuple):
88    data: json.Data

 94class Event(typing.NamedTuple):
 95    """Event
 96
 97    Operators `>` and `<` test for natural order where it is assumed that
 98    first operand is registered before second operand.
 99
100    """
101
102    id: EventId
103    type: EventType
104    timestamp: Timestamp
105    source_timestamp: Timestamp | None
106    payload: EventPayload | None
107
108    def __lt__(self, other):
109        if not isinstance(other, Event):
110            return NotImplemented
111
112        if self.id == other.id:
113            return False
114
115        if self.id.server == other.id.server:
116            return self.id < other.id
117
118        if self.timestamp != other.timestamp:
119            return self.timestamp < other.timestamp
120
121        return True
122
123    def __gt__(self, other):
124        if not isinstance(other, Event):
125            return NotImplemented
126
127        if self.id == other.id:
128            return False
129
130        if self.id.server == other.id.server:
131            return self.id > other.id
132
133        if self.timestamp != other.timestamp:
134            return self.timestamp > other.timestamp
135
136        return False

139class RegisterEvent(typing.NamedTuple):
140    type: EventType
141    source_timestamp: Timestamp | None
142    payload: EventPayload | None

145class QueryLatestParams(typing.NamedTuple):
146    event_types: Collection[EventType] | None = None

149class QueryTimeseriesParams(typing.NamedTuple):
150    event_types: Collection[EventType] | None = None
151    t_from: Timestamp | None = None
152    t_to: Timestamp | None = None
153    source_t_from: Timestamp | None = None
154    source_t_to: Timestamp | None = None
155    order: Order = Order.DESCENDING
156    order_by: OrderBy = OrderBy.TIMESTAMP
157    max_results: int | None = None
158    last_event_id: EventId | None = None

161class QueryServerParams(typing.NamedTuple):
162    server_id: ServerId
163    persisted: bool = False
164    max_results: int | None = None
165    last_event_id: EventId | None = None

173class QueryResult(typing.NamedTuple):
174    events: Collection[Event]
175    more_follows: bool

26def timestamp_to_bytes(t: Timestamp) -> util.Bytes:
27    """Convert timestamp to 12 byte representation
28
29    Bytes [0, 8] are big endian unsigned `Timestamp.s` + 2^63 and
30    bytes [9, 12] are big endian unsigned `Timestamp.us`.
31
32    """
33    return struct.pack(">QI", t.s + (1 << 63), t.us)

36def timestamp_from_bytes(data: util.Bytes) -> Timestamp:
37    """Create new timestamp from 12 byte representation
38
39    Bytes representation is same as defined for `timestamp_to_bytes` function.
40
41    """
42    s, us = struct.unpack(">QI", data)
43    return Timestamp(s - (1 << 63), us)

46def timestamp_to_float(t: Timestamp) -> float:
47    """Convert timestamp to floating number of seconds since 1970-01-01 UTC
48
49    For precise serialization see `timestamp_to_bytes`/`timestamp_from_bytes`.
50
51    """
52    return t.s + t.us * 1E-6

55def timestamp_from_float(ts: float) -> Timestamp:
56    """Create timestamp from floating number of seconds since 1970-01-01 UTC
57
58    For precise serialization see `timestamp_to_bytes`/`timestamp_from_bytes`.
59
60    """
61    s = int(ts)
62    if ts < 0:
63        s = s - 1
64
65    us = round((ts - s) * 1E6)
66
67    if us == 1_000_000:
68        return Timestamp(s + 1, 0)
69
70    else:
71        return Timestamp(s, us)

74def timestamp_to_datetime(t: Timestamp) -> datetime.datetime:
75    """Convert timestamp to datetime (representing utc time)
76
77    For precise serialization see `timestamp_to_bytes`/`timestamp_from_bytes`.
78
79    """
80    try:
81        dt_from_s = datetime.datetime.fromtimestamp(t.s, datetime.timezone.utc)
82
83    except OSError:
84        dt_from_s = (
85            datetime.datetime(1970, 1, 1, tzinfo=datetime.timezone.utc) +
86            datetime.timedelta(seconds=t.s))
87
88    return datetime.datetime(
89        year=dt_from_s.year,
90        month=dt_from_s.month,
91        day=dt_from_s.day,
92        hour=dt_from_s.hour,
93        minute=dt_from_s.minute,
94        second=dt_from_s.second,
95        microsecond=t.us,
96        tzinfo=datetime.timezone.utc)

 99def timestamp_from_datetime(dt: datetime.datetime) -> Timestamp:
100    """Create new timestamp from datetime
101
102    If `tzinfo` is not set, it is assumed that provided datetime represents
103    utc time.
104
105    For precise serialization see `timestamp_to_bytes`/`timestamp_from_bytes`.
106
107    """
108    if not dt.tzinfo:
109        dt = dt.replace(tzinfo=datetime.timezone.utc)
110
111    s = int(dt.timestamp())
112
113    if dt.timestamp() < 0:
114        s = s - 1
115
116    return Timestamp(s=s, us=dt.microsecond)

119def timestamp_to_sbs(t: Timestamp) -> sbs.Data:
120    """Convert timestamp to SBS data"""
121    return {'s': t.s, 'us': t.us}

124def timestamp_from_sbs(data: sbs.Data) -> Timestamp:
125    """Create new timestamp from SBS data"""
126    return Timestamp(s=data['s'], us=data['us'])

129def status_to_sbs(status: Status) -> sbs.Data:
130    """Convert Status to SBS data"""
131    return status.value, None

134def status_from_sbs(status: sbs.Data) -> Status:
135    """Create Status based on SBS data"""
136    return Status(status[0])

139def event_to_sbs(event: Event) -> sbs.Data:
140    """Convert Event to SBS data"""
141    return {'id': _event_id_to_sbs(event.id),
142            'type': list(event.type),
143            'timestamp': timestamp_to_sbs(event.timestamp),
144            'sourceTimestamp': _optional_to_sbs(event.source_timestamp,
145                                                timestamp_to_sbs),
146            'payload': _optional_to_sbs(event.payload, event_payload_to_sbs)}

149def event_from_sbs(data: sbs.Data) -> Event:
150    """Create Event based on SBS data"""
151    return Event(id=_event_id_from_sbs(data['id']),
152                 type=tuple(data['type']),
153                 timestamp=timestamp_from_sbs(data['timestamp']),
154                 source_timestamp=_optional_from_sbs(data['sourceTimestamp'],
155                                                     timestamp_from_sbs),
156                 payload=_optional_from_sbs(data['payload'],
157                                            event_payload_from_sbs))

160def register_event_to_sbs(event: RegisterEvent) -> sbs.Data:
161    """Convert RegisterEvent to SBS data"""
162    return {'type': list(event.type),
163            'sourceTimestamp': _optional_to_sbs(event.source_timestamp,
164                                                timestamp_to_sbs),
165            'payload': _optional_to_sbs(event.payload, event_payload_to_sbs)}

168def register_event_from_sbs(data: sbs.Data) -> RegisterEvent:
169    """Create RegisterEvent based on SBS data"""
170    return RegisterEvent(
171        type=tuple(data['type']),
172        source_timestamp=_optional_from_sbs(data['sourceTimestamp'],
173                                            timestamp_from_sbs),
174        payload=_optional_from_sbs(data['payload'], event_payload_from_sbs))

177def query_params_to_sbs(params: QueryParams) -> sbs.Data:
178    """Convert QueryParams to SBS data"""
179    if isinstance(params, QueryLatestParams):
180        return 'latest', {
181            'eventTypes': _optional_to_sbs(params.event_types,
182                                           _event_types_to_sbs)}
183
184    if isinstance(params, QueryTimeseriesParams):
185        return 'timeseries', {
186            'eventTypes': _optional_to_sbs(params.event_types,
187                                           _event_types_to_sbs),
188            'tFrom': _optional_to_sbs(params.t_from, timestamp_to_sbs),
189            'tTo': _optional_to_sbs(params.t_to, timestamp_to_sbs),
190            'sourceTFrom': _optional_to_sbs(params.source_t_from,
191                                            timestamp_to_sbs),
192            'sourceTTo': _optional_to_sbs(params.source_t_to,
193                                          timestamp_to_sbs),
194            'order': (params.order.value, None),
195            'orderBy': (params.order_by.value, None),
196            'maxResults': _optional_to_sbs(params.max_results),
197            'lastEventId': _optional_to_sbs(params.last_event_id,
198                                            _event_id_to_sbs)}
199
200    if isinstance(params, QueryServerParams):
201        return 'server', {
202            'serverId': params.server_id,
203            'persisted': params.persisted,
204            'maxResults': _optional_to_sbs(params.max_results),
205            'lastEventId': _optional_to_sbs(params.last_event_id,
206                                            _event_id_to_sbs)}
207
208    raise ValueError('unsupported params type')

211def query_params_from_sbs(data: sbs.Data) -> QueryParams:
212    """Create QueryParams based on SBS data"""
213    if data[0] == 'latest':
214        return QueryLatestParams(
215            event_types=_optional_from_sbs(data[1]['eventTypes'],
216                                           _event_types_from_sbs))
217
218    if data[0] == 'timeseries':
219        return QueryTimeseriesParams(
220            event_types=_optional_from_sbs(data[1]['eventTypes'],
221                                           _event_types_from_sbs),
222            t_from=_optional_from_sbs(data[1]['tFrom'], timestamp_from_sbs),
223            t_to=_optional_from_sbs(data[1]['tTo'], timestamp_from_sbs),
224            source_t_from=_optional_from_sbs(data[1]['sourceTFrom'],
225                                             timestamp_from_sbs),
226            source_t_to=_optional_from_sbs(data[1]['sourceTTo'],
227                                           timestamp_from_sbs),
228            order=Order(data[1]['order'][0]),
229            order_by=OrderBy(data[1]['orderBy'][0]),
230            max_results=_optional_from_sbs(data[1]['maxResults']),
231            last_event_id=_optional_from_sbs(data[1]['lastEventId'],
232                                             _event_id_from_sbs))
233
234    if data[0] == 'server':
235        return QueryServerParams(
236            server_id=data[1]['serverId'],
237            persisted=data[1]['persisted'],
238            max_results=_optional_from_sbs(data[1]['maxResults']),
239            last_event_id=_optional_from_sbs(data[1]['lastEventId'],
240                                             _event_id_from_sbs))
241
242    raise ValueError('unsupported params type')

245def query_result_to_sbs(result: QueryResult) -> sbs.Data:
246    """Convert QueryResult to SBS data"""
247    return {'events': [event_to_sbs(event) for event in result.events],
248            'moreFollows': result.more_follows}

251def query_result_from_sbs(data: sbs.Data) -> QueryResult:
252    """Create QueryResult based on SBS data"""
253    return QueryResult(events=[event_from_sbs(i) for i in data['events']],
254                       more_follows=data['moreFollows'])

257def event_payload_to_sbs(payload: EventPayload) -> sbs.Data:
258    """Convert EventPayload to SBS data"""
259    if isinstance(payload, EventPayloadBinary):
260        return 'binary', {'type': payload.type,
261                          'data': payload.data}
262
263    if isinstance(payload, EventPayloadJson):
264        return 'json', json.encode(payload.data)
265
266    raise ValueError('unsupported payload type')

269def event_payload_from_sbs(data: sbs.Data) -> EventPayload:
270    """Create EventPayload based on SBS data"""
271    if data[0] == 'binary':
272        return EventPayloadBinary(type=data[1]['type'],
273                                  data=data[1]['data'])
274
275    if data[0] == 'json':
276        return EventPayloadJson(data=json.decode(data[1]))
277
278    raise ValueError('unsupported payload type')

 5def matches_query_type(event_type: EventType,
 6                       query_type: EventType
 7                       ) -> bool:
 8    """Determine if event type matches query type
 9
10    Event type is tested if it matches query type according to the following
11    rules:
12
13        * Matching is performed on subtypes in increasing order.
14        * Event type is a match only if all its subtypes are matched by
15          corresponding query subtypes.
16        * Matching is finished when all query subtypes are exhausted.
17        * Query subtype '?' matches exactly one event subtype of any value.
18          The subtype must exist.
19        * Query subtype '*' matches 0 or more event subtypes of any value. It
20          must be the last query subtype.
21        * All other values of query subtype match exactly one event subtype
22          of the same value.
23        * Query type without subtypes is matched only by event type with no
24          subtypes.
25
26    As a consequence of aforementioned matching rules, event subtypes '*' and
27    '?' cannot be directly matched and it is advisable not to use them in event
28    types.
29
30    """
31    is_variable = bool(query_type and query_type[-1] == '*')
32    if is_variable:
33        query_type = query_type[:-1]
34
35    if len(event_type) < len(query_type):
36        return False
37
38    if len(event_type) > len(query_type) and not is_variable:
39        return False
40
41    for i, j in zip(event_type, query_type):
42        if j != '?' and i != j:
43            return False
44
45    return True

18class SourceType(enum.Enum):
19    EVENTER = 1
20    MODULE = 2
21    ENGINE = 3
22    SERVER = 4

25class Source(typing.NamedTuple):
26    type: SourceType
27    id: int

30class Engine(aio.Resource):
31    """Engine ABC"""
32
33    @abc.abstractmethod
34    async def register(self,
35                       source: Source,
36                       events: Collection[RegisterEvent]
37                       ) -> Collection[Event] | None:
38        """Register events"""
39
40    @abc.abstractmethod
41    async def query(self,
42                    params: QueryParams
43                    ) -> QueryResult:
44        """Query events"""