Coverage for lib/datamodel/event.py: 93%
87 statements
« prev ^ index » next coverage.py v7.10.1, created at 2025-07-28 07:25 +0000
1#!/usr/bin/env python3
2# -*- coding: utf-8 -*-
4# Hermes : Change Data Capture (CDC) tool from any source(s) to any target
5# Copyright (C) 2023, 2024 INSA Strasbourg
6#
7# This file is part of Hermes.
8#
9# Hermes is free software: you can redistribute it and/or modify
10# it under the terms of the GNU General Public License as published by
11# the Free Software Foundation, either version 3 of the License, or
12# (at your option) any later version.
13#
14# Hermes is distributed in the hope that it will be useful,
15# but WITHOUT ANY WARRANTY; without even the implied warranty of
16# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
17# GNU General Public License for more details.
18#
19# You should have received a copy of the GNU General Public License
20# along with Hermes. If not, see <https://www.gnu.org/licenses/>.
23from typing import Any
25from lib.datamodel.dataobject import DataObject
26from lib.datamodel.diffobject import DiffObject
27from lib.datamodel.serialization import JSONSerializable
29from datetime import datetime
32class Event(JSONSerializable):
33 """Serializable Event message"""
35 EVTYPES = ["initsync", "added", "modified", "removed", "dataschema"]
37 LONG_STRING_LIMIT: int | None = 256
38 """If a string attribute should be logged and its len is greater than this value,
39 it will be marked as a LONG_STRING and its content will be truncated.
40 Can be set to None to disable this feature.
41 """
43 def __init__(
44 self,
45 evcategory: str | None = None,
46 eventtype: str | None = None,
47 obj: DataObject | None = None,
48 objattrs: dict[str, Any] | None = None,
49 from_json_dict: dict[str, Any] | None = None,
50 ):
51 """Create Event message
52 - of specified category (base/initsync)
53 - of specified type (init-start, init-stop, added, modified, removed)
54 - with facultative Dataobject instance obj that will be used to store it type
55 and pkey
56 - with specified objattrs that should contains a dict with useful attributes in
57 current context defined by evcategory and eventtype
58 or a from_json_dict to deserialize an Event instance
59 """
61 if objattrs is None and from_json_dict is None:
62 err = (
63 "Cannot instantiate object from nothing: you must specify one data"
64 " source"
65 )
66 __hermes__.logger.critical(err)
67 raise AttributeError(err)
69 if objattrs is not None and from_json_dict is not None:
70 err = "Cannot instantiate object from multiple data sources at once"
71 __hermes__.logger.critical(err)
72 raise AttributeError(err)
74 __jsondataattrs = [
75 "evcategory",
76 "eventtype",
77 "objtype",
78 "objpkey",
79 "objattrs",
80 "step",
81 "isPartiallyProcessed",
82 ]
83 super().__init__(jsondataattr=__jsondataattrs)
84 self.offset: int | None = None
85 self.timestamp: datetime = datetime(year=1, month=1, day=1)
86 self.step: int = 0
87 self.isPartiallyProcessed: bool = False
88 if from_json_dict is not None:
89 for attr in __jsondataattrs:
90 if attr in from_json_dict:
91 setattr(self, attr, from_json_dict[attr])
92 if attr == "objpkey" and type(self.objpkey) is list:
93 self.objpkey = tuple(self.objpkey)
94 else:
95 if attr == "isPartiallyProcessed":
96 # "isPartiallyProcessed" was added in v1.0.0-alpha.2,
97 # As fallback when missing, set to True if step > 0, False
98 # otherwise
99 if from_json_dict.get("step", 0) != 0:
100 self.isPartiallyProcessed = True
101 # As obj instance isn't available, use pkey as default repr
102 self.objrepr = str(self.objpkey)
103 else:
104 self.evcategory: str | None = evcategory
105 self.eventtype: str | None = eventtype
106 if obj is None:
107 self.objtype: str | None = None
108 self.objpkey: Any = None
109 self.objrepr: str | None = None
110 else:
111 self.objtype = obj.getType()
112 self.objpkey = obj.getPKey()
113 self.objrepr = repr(obj)
114 self.objattrs: dict[str, Any] | None = objattrs
116 def __repr__(self) -> str:
117 """Returns a printable representation of current Event"""
118 category = f"{self.evcategory}_" if self.evcategory != "base" else ""
120 if self.objtype is None:
121 s = f"<Event({category}{self.eventtype})>"
122 else:
123 s = f"<Event({category}{self.objtype}_{self.eventtype}[{self.objrepr}])>"
124 return s
126 def toString(self, secretattrs: set[str]) -> str:
127 """Returns a printable string of current Event"""
128 category = f"{self.evcategory}_" if self.evcategory != "base" else ""
129 objattrs = self.objattrsToString(self.objattrs, secretattrs)
131 if self.objtype is None:
132 s = f"<Event({category}{self.eventtype}, {objattrs})>"
133 else:
134 s = (
135 f"<Event({category}{self.objtype}_{self.eventtype}[{self.objrepr}],"
136 f" {objattrs})>"
137 )
138 return s
140 @staticmethod
141 def objattrsToString(objattrs: dict[str, any], secretattrs: set[str]) -> str:
142 """Returns a printable string of current objattrs dict, with specified
143 secret attributes filtered"""
144 res = {}
145 for k, v in objattrs.items():
146 if type(v) is dict:
147 res[k] = Event.objattrsToString(v, secretattrs)
148 continue
150 if k in secretattrs:
151 res[k] = f"<SECRET_VALUE({type(v)})>"
152 elif type(v) is bytes:
153 res[k] = f"<BINARY_DATA({len(v)})>"
154 elif (
155 type(v) is str
156 and Event.LONG_STRING_LIMIT is not None
157 and len(v) > Event.LONG_STRING_LIMIT
158 ):
159 res[k] = f"<LONG_STR({len(v)}, '{v[:Event.LONG_STRING_LIMIT]}...')>"
160 else:
161 res[k] = v
162 return res
164 @staticmethod
165 def fromDiffItem(
166 diffitem: DiffObject | DataObject, eventCategory: str, changeType: str
167 ) -> tuple["Event", DataObject]:
168 """Convert the specified diffitem (item from DiffObject of two DataObjectList)
169 to Event with specified eventCategory and changeType.
170 Return the event, and the 'new' DataObject from diffitem"""
171 obj: DataObject
172 objattrs: dict[str, Any]
174 match changeType:
175 case "modified":
176 # changetype is "modified", so diffitem is a DiffObject
177 obj = diffitem.objnew
178 objattrs = diffitem.dict
179 case "added":
180 # changetype isn't "modified", so diffitem is a DataObject
181 obj = diffitem
182 objattrs = diffitem.toEvent()
183 case "removed":
184 # changetype isn't "modified", so diffitem is a DataObject
185 obj = diffitem
186 objattrs = {}
187 case "_":
188 raise AttributeError(
189 f"Invalid {changeType=} specified: valid values are"
190 " ['added', 'modified', 'removed']"
191 )
193 return (
194 Event(
195 evcategory=eventCategory,
196 eventtype=changeType,
197 obj=obj,
198 objattrs=objattrs,
199 ),
200 obj,
201 )