Coverage for src/pythia/event_stream/log.py: 54%

1"""Memory-backed event stream storage.""" 

2 

3from __future__ import annotations 

4 

5from pathlib import Path 

6 

7from pythia.applications.annotation import _DumpLogger 

8from pythia.applications.annotation import _make_logger 

9from pythia.event_stream.base import Backend as Base 

10 

11 

12class Backend(Base): 

13 """Simple event stream client to dump incoming data using logs.""" 

14 

15 _logger: _DumpLogger | None = None 

16 

17 @property 

18 def logger(self) -> _DumpLogger: 

19 """Internal logger lazy-loader. 

20 

21 Returns: 

22 Initialized logger. 

23 

24 """ 

25 if self._logger is None: 

26 self.connect() 

27 return self._logger # type: ignore 

28 

29 def connect(self) -> None: 

30 """Fetch stream-specific deque from global container.""" 

31 if self.stream not in ("stdout", "stderr"): 

32 self.stream = Path(self.stream) 

33 self._logger = _make_logger(type(self).__qualname__, self.stream) 

34 

35 def post(self, data) -> None: 

36 """Append an element into the deque. 

37 

38 Args: 

39 data: the data to append. Can be any python object. 

40 

41 """ 

42 self.logger.json(data)