Coverage for rdflib_ocdm / counter_handler / redis_counter_handler.py: 100%

1#!/usr/bin/python 

2 

3# SPDX-FileCopyrightText: 2023-2025 Arcangelo Massari <arcangelo.massari@unibo.it> 

4# 

5# SPDX-License-Identifier: ISC 

6 

7import redis 

8 

9 

10class RedisCounterHandler: 

11 def __init__(self, host, port, db: int, password=None): 

12 self.host = host 

13 self.port = port 

14 self.db = db 

15 self.password = password 

16 self.connection = None 

17 

18 def connect(self): 

19 self.connection = redis.Redis(host=self.host, port=self.port, db=self.db, password=self.password) 

20 

21 def disconnect(self): 

22 if self.connection: 

23 self.connection.close() 

24 

25 def set_counter(self, new_value: int, entity_name: str) -> None: 

26 entity_name = str(entity_name) 

27 if new_value < 0: 

28 raise ValueError("new_value must be a non negative integer!") 

29 self.connection.set(entity_name, new_value) 

30 

31 def read_counter(self, entity_name: str) -> int: 

32 entity_name = str(entity_name) 

33 result = self.connection.get(entity_name) 

34 if result: 

35 return int(result.decode('utf-8')) 

36 else: 

37 return 0 

38 

39 def increment_counter(self, entity_name: str) -> int: 

40 entity_name = str(entity_name) 

41 cur_count = self.read_counter(entity_name) 

42 count = cur_count + 1 

43 self.set_counter(count, entity_name) 

44 return count 

45 

46 def flush(self): 

47 self.connection.flushdb()