hat.event.backends.lmdb.convert.v09

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))

73class OrderBy(enum.Enum):
74    TIMESTAMP = 'timestamp'
75    SOURCE_TIMESTAMP = 'sourceTimestamp'

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

84class EventPayloadBinary(typing.NamedTuple):
85    type: str
86    data: util.Bytes

89class EventPayloadJson(typing.NamedTuple):
90    data: json.Data

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

31class DbType(enum.Enum):
32    SYSTEM_SETTINGS = 0
33    SYSTEM_LAST_EVENT_ID = 1
34    SYSTEM_LAST_TIMESTAMP = 2
35    REF = 3
36    LATEST_DATA = 4
37    LATEST_TYPE = 5
38    TIMESERIES_DATA = 6
39    TIMESERIES_PARTITION = 7
40    TIMESERIES_COUNT = 8

75class SettingsId(enum.Enum):
76    VERSION = 0
77    IDENTIFIER = 1

80class LatestEventRef(typing.NamedTuple):
81    key: EventTypeRef

84class TimeseriesEventRef(typing.NamedTuple):
85    key: TimeseriesKey

147def ext_create_env(path: Path,
148                   max_size: int = default_max_size,
149                   readonly: bool = False
150                   ) -> lmdb.Environment:
151    return lmdb.Environment(str(path),
152                            map_size=max_size,
153                            subdir=False,
154                            max_dbs=len(DbType),
155                            readonly=readonly)

158def ext_open_db(env: lmdb.Environment,
159                db_type: DbType,
160                create: bool = True
161                ) -> lmdb._Database:
162    return env.open_db(db_type.name.encode('utf-8'), create=create)

30def read(dbs: dict[DbType, lmdb._Database],
31         txn: lmdb.Transaction,
32         db_type: DbType
33         ) -> typing.Iterable[tuple[DbKey, DbValue]]:
34    db_def = db_defs[db_type]
35    with txn.cursor(dbs[db_type]) as cursor:
36        for encoded_key, encoded_value in cursor:
37            key = db_def.decode_key(encoded_key)
38            value = db_def.decode_value(encoded_value)
39            yield key, value

42def write(dbs: dict[DbType, lmdb._Database],
43          txn: lmdb.Transaction,
44          db_type: DbType,
45          key: DbKey,
46          value: DbValue):
47    db_def = db_defs[db_type]
48    encoded_key = db_def.encode_key(key)
49    encoded_value = db_def.encode_value(value)
50
51    txn.put(encoded_key, encoded_value, db=dbs[db_type])