16909028829

Committed 12 Aug 2025 12:35PM UTC coverage: 29.831% (+2.7%) from 27.099%

Build # 16909028829

Build Type

Pull #11

github

Committed by

cwehmeier

Commit Message

KIT-4207 updated version in setup

Pull Request Pull Request #11: KIT-4207

Run Details

24 of 100 new or added lines in 5 files covered. (24.0%)

5 existing lines in 2 files now uncovered.

159 of 533 relevant lines covered (29.83%)

0.3 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

93.02

/sag_py_logging_logstash/memory_cache.py

# -*- coding: utf-8 -*-
#
# This software may be modified and distributed under the terms
# of the MIT license.  See the LICENSE file for details.

import uuid
from datetime import datetime, timedelta

from sag_py_logging_logstash.cache import Cache
from sag_py_logging_logstash.constants import constants


class MemoryCache(Cache):
    """Backend implementation for python-logstash-async. Keeps messages in a local, in-memory cache
    while attempting to publish them to logstash. Does not persist through process restarts. Also,
    does not write to disk.

    :param cache: Usually just an empty dictionary
    :param event_ttl: Optional parameter used to expire events in the cache after a time
    """

    # ----------------------------------------------------------------------
    def __init__(self, safe_logger, cache, event_ttl=None):
        self._cache = cache
        self._event_ttl = event_ttl
        self._safe_logger = safe_logger

    # ----------------------------------------------------------------------
    def add_event(self, event):
        event_id = uuid.uuid4()
        self._cache[event_id] = {
            "event_text": event,
            "pending_delete": False,
            "entry_date": datetime.now(),
            "id": event_id,
        }

    # ----------------------------------------------------------------------
    def get_queued_events(self):
        events = []
        event_count = 0
        for event in self._cache.values():
            if not event["pending_delete"]:
                events.append(event)
                event["pending_delete"] = True

                event_count += 1
                if event_count >= constants.QUEUED_EVENTS_BATCH_SIZE:
                    break
        return events

    # ----------------------------------------------------------------------
    def requeue_queued_events(self, events):
        for event in events:
            event_to_queue = self._cache.get(event["id"], None)
            # If they gave us an event which is not in the cache,
            # there is really nothing for us to do. Right now
            # this use-case does not raise an error. Instead, we
            # just log the message.
            if event_to_queue:
                event_to_queue["pending_delete"] = False
            else:
                self._safe_logger.log(
                    "warning",
                    "Could not requeue event with id {}. It does not appear to be in the cache.".format(event["id"]),
                )

    # ----------------------------------------------------------------------
    def delete_queued_events(self):
        ids_to_delete = [event["id"] for event in self._cache.values() if event["pending_delete"]]
        self._delete_events(ids_to_delete)

    # ----------------------------------------------------------------------
    def expire_events(self):
        if self._event_ttl is None:
            return

        delete_time = datetime.now() - timedelta(seconds=self._event_ttl)
        ids_to_delete = [event["id"] for event in self._cache.values() if event["entry_date"] < delete_time]
        self._delete_events(ids_to_delete)

    # ----------------------------------------------------------------------
    def _delete_events(self, ids_to_delete):
        for event_id in ids_to_delete:
            # If the event is not in the cache, is there anything
            # that we can do. This currently doesn't throw an error.
            event = self._cache.pop(event_id, None)
            if not event:
                self._safe_logger.log(
                    "warning",
                    "Could not delete event with id {}. It does not appear to be in the cache.".format(event_id),
                )

1	# -- coding: utf-8 --
2	#
3	# This software may be modified and distributed under the terms
4	# of the MIT license. See the LICENSE file for details.
5
6	import uuid	1✔
7	from datetime import datetime, timedelta	1✔
8
9	from sag_py_logging_logstash.cache import Cache	1✔
10	from sag_py_logging_logstash.constants import constants	1✔
11
12
13	class MemoryCache(Cache):	1✔
14	"""Backend implementation for python-logstash-async. Keeps messages in a local, in-memory cache
15	while attempting to publish them to logstash. Does not persist through process restarts. Also,
16	does not write to disk.
17
18	:param cache: Usually just an empty dictionary
19	:param event_ttl: Optional parameter used to expire events in the cache after a time
20	"""
21
22	# ----------------------------------------------------------------------
23	def __init__(self, safe_logger, cache, event_ttl=None):	1✔
24	self._cache = cache	1✔
25	self._event_ttl = event_ttl	1✔
26	self._safe_logger = safe_logger	1✔
27
28	# ----------------------------------------------------------------------
29	def add_event(self, event):	1✔
30	event_id = uuid.uuid4()	1✔
31	self._cache[event_id] = {	1✔
32	"event_text": event,
33	"pending_delete": False,
34	"entry_date": datetime.now(),
35	"id": event_id,
36	}
37
38	# ----------------------------------------------------------------------
39	def get_queued_events(self):	1✔
40	events = []	1✔
41	event_count = 0	1✔
42	for event in self._cache.values():	1✔
43	if not event["pending_delete"]:	1✔
44	events.append(event)	1✔
45	event["pending_delete"] = True	1✔
46
47	event_count += 1	1✔
48	if event_count >= constants.QUEUED_EVENTS_BATCH_SIZE:	1✔
49	break	1✔
50	return events	1✔
51
52	# ----------------------------------------------------------------------
53	def requeue_queued_events(self, events):	1✔
54	for event in events:	1✔
55	event_to_queue = self._cache.get(event["id"], None)	1✔
56	# If they gave us an event which is not in the cache,
57	# there is really nothing for us to do. Right now
58	# this use-case does not raise an error. Instead, we
59	# just log the message.
60	if event_to_queue:	1✔
61	event_to_queue["pending_delete"] = False	1✔
62	else:
NEW 63	self._safe_logger.log(	×
64	"warning",
65	"Could not requeue event with id {}. It does not appear to be in the cache.".format(event["id"]),
66	)
67
68	# ----------------------------------------------------------------------
69	def delete_queued_events(self):	1✔
70	ids_to_delete = [event["id"] for event in self._cache.values() if event["pending_delete"]]	1✔
71	self._delete_events(ids_to_delete)	1✔
72
73	# ----------------------------------------------------------------------
74	def expire_events(self):	1✔
75	if self._event_ttl is None:	1✔
76	return	×
77
78	delete_time = datetime.now() - timedelta(seconds=self._event_ttl)	1✔
79	ids_to_delete = [event["id"] for event in self._cache.values() if event["entry_date"] < delete_time]	1✔
80	self._delete_events(ids_to_delete)	1✔
81
82	# ----------------------------------------------------------------------
83	def _delete_events(self, ids_to_delete):	1✔
84	for event_id in ids_to_delete:	1✔
85	# If the event is not in the cache, is there anything
86	# that we can do. This currently doesn't throw an error.
87	event = self._cache.pop(event_id, None)	1✔
88	if not event:	1✔
NEW 89	self._safe_logger.log(	×
90	"warning",
91	"Could not delete event with id {}. It does not appear to be in the cache.".format(event_id),
92	)

SamhammerAG / sag_py_logging_logstash / 16909028829

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous