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

1#!/usr/bin/python 

2 

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

4# 

5# SPDX-License-Identifier: ISC 

6 

7from __future__ import annotations 

8 

9import redis 

10 

11 

12class RedisCounterHandler: 

13 def __init__(self, host: str, port: int, db: int, password: str | None = None): 

14 self.host = host 

15 self.port = port 

16 self.db = db 

17 self.password = password 

18 self.connection: redis.Redis | None = None # type: ignore[type-arg] 

19 

20 def connect(self) -> None: 

21 self.connection = redis.Redis( 

22 host=self.host, port=self.port, db=self.db, password=self.password 

23 ) 

24 

25 def disconnect(self) -> None: 

26 if self.connection: 

27 self.connection.close() 

28 

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

30 entity_name = str(entity_name) 

31 if new_value < 0: 

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

33 assert self.connection is not None 

34 self.connection.set(entity_name, new_value) 

35 

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

37 entity_name = str(entity_name) 

38 assert self.connection is not None 

39 result: bytes | None = self.connection.get(entity_name) # type: ignore[assignment] 

40 if result: 

41 return int(result.decode("utf-8")) 

42 else: 

43 return 0 

44 

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

46 entity_name = str(entity_name) 

47 cur_count = self.read_counter(entity_name) 

48 count = cur_count + 1 

49 self.set_counter(count, entity_name) 

50 return count 

51 

52 def flush(self) -> None: 

53 assert self.connection is not None 

54 self.connection.flushdb()