Skip to content

Common module

Common classes/utils included in SDK public API.

Caching

Cache

Bases: ABC

Interface for cache instances used across SDK. Can be used to create custom cache implementations.

Source code in hiero_did_sdk_python/utils/cache.py 
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
class Cache[K, V](ABC):
    """Interface for cache instances used across SDK. Can be used to create custom cache implementations."""

    def __init__(self):
        self._locks = {}

    def _get_lock(self, key: K) -> Lock:
        if key not in self._locks:
            self._locks[key] = Lock()

        return self._locks[key]

    @final
    def get(self, key: K) -> V | None:
        """Get cached data by key

        Args:
            key: Cached data key

        Returns:
            object: Cached data
        """

        lock = self._get_lock(key)

        with lock:
            return self.data_get(key)

    @final
    def set(self, key: K, value: V, ttl: seconds | None = None) -> None:
        """Set cached data with key

        Args:
            key: Data key
            value: Data to cache
            ttl: Data retention duration in seconds.
        """
        ttl = ttl or DEFAULT_TTL

        lock = self._get_lock(key)

        with lock:
            self.data_set(key, value, ttl)

    @final
    def remove(self, key: K) -> None:
        """Remove cached data by key.

        Args:
            key: Cached data key
        """
        lock = self._get_lock(key)

        with lock:
            self.data_remove(key)

    @final
    def size(self):
        """Get cached records count."""
        return self.data_size()

    @final
    def flush(self):
        """Clear cached data."""
        return self.data_flush()

    @abstractmethod
    def data_get(self, key: K) -> V | None:
        pass

    @abstractmethod
    def data_set(self, key: K, value: V, ttl):
        pass

    @abstractmethod
    def data_remove(self, key: K):
        pass

    @abstractmethod
    def data_size(self) -> int:
        pass

    @abstractmethod
    def data_flush(self):
        pass

flush()

Clear cached data.

Source code in hiero_did_sdk_python/utils/cache.py 
86
87
88
89
@final
def flush(self):
    """Clear cached data."""
    return self.data_flush()

get(key)

Get cached data by key

Parameters:

Name Type Description Default
key K

Cached data key

 required

Returns:

Name Type Description
object V | None

Cached data
Source code in hiero_did_sdk_python/utils/cache.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
@final
def get(self, key: K) -> V | None:
    """Get cached data by key

    Args:
        key: Cached data key

    Returns:
        object: Cached data
    """

    lock = self._get_lock(key)

    with lock:
        return self.data_get(key)

remove(key)

Remove cached data by key.

Parameters:

Name Type Description Default
key K

Cached data key

 required
Source code in hiero_did_sdk_python/utils/cache.py 
69
70
71
72
73
74
75
76
77
78
79
@final
def remove(self, key: K) -> None:
    """Remove cached data by key.

    Args:
        key: Cached data key
    """
    lock = self._get_lock(key)

    with lock:
        self.data_remove(key)

set(key, value, ttl=None)

Set cached data with key

Parameters:

Name Type Description Default
key K

Data key

 required
value V

Data to cache

 required
ttl seconds | None

Data retention duration in seconds.

 None
Source code in hiero_did_sdk_python/utils/cache.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
@final
def set(self, key: K, value: V, ttl: seconds | None = None) -> None:
    """Set cached data with key

    Args:
        key: Data key
        value: Data to cache
        ttl: Data retention duration in seconds.
    """
    ttl = ttl or DEFAULT_TTL

    lock = self._get_lock(key)

    with lock:
        self.data_set(key, value, ttl)

size()

Get cached records count.

Source code in hiero_did_sdk_python/utils/cache.py 
81
82
83
84
@final
def size(self):
    """Get cached records count."""
    return self.data_size()

MemoryCache

Bases: Cache[K, V]

In-memory cache implementation. Includes built-in data retention logic.

Source code in hiero_did_sdk_python/utils/cache.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
class MemoryCache[K, V](Cache[K, V]):
    """In-memory cache implementation. Includes built-in data retention logic."""

    def __init__(self):
        super().__init__()
        self._mem: dict[K, TimestampedRecord[V]] = {}

    # Cache clearing logic goes to child class because different classes can use different strategies
    # For example, Redis would use its built-in key TTL mechanic
    def _remove_expired_cached_items(self):
        for key in self._mem.copy():
            value = self._mem.get(key, None)

            # Assure value is still there, in multithreaded environment
            if value is not None:
                expires_timestamp = value.timestamp
                now = time.time()

                if now > expires_timestamp:
                    self.data_remove(key)

    @override
    def data_get(self, key: K) -> V | None:
        self._remove_expired_cached_items()

        record = self._mem.get(key, None)

        if record is None:
            return None

        return record.data

    @override
    def data_set(self, key: K, value: V, ttl: seconds):
        self._remove_expired_cached_items()

        expires_timestamp = time.time() + ttl
        self._mem[key] = TimestampedRecord(value, expires_timestamp)

    @override
    def data_size(self):
        return len(self._mem)

    @override
    def data_remove(self, key):
        # del wouldn't be thread safe
        self._mem.pop(key, None)

    @override
    def data_flush(self):
        self._mem = {}

TimestampedRecord dataclass

Helper model for timestamped records.

Attributes:

Name Type Description
data T

Record data (generic)
timestamp float

Timestamp of record creation
Source code in hiero_did_sdk_python/utils/cache.py 
12
13
14
15
16
17
18
19
20
21
22
@dataclass
class TimestampedRecord[T]:
    """Helper model for timestamped records.

    Attributes:
        data: Record data (generic)
        timestamp: Timestamp of record creation
    """

    data: T
    timestamp: float = field(default_factory=time.time)

Helper classes and utils

Serializable

Source code in hiero_did_sdk_python/utils/serializable.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
class Serializable:
    @classmethod
    def from_json(
        cls,
        json_str: str,
    ) -> Self:
        """Parse a JSON string into an object instance.

        Args:
            json_str: JSON string

        Returns:
            An instance representation of this JSON

        """
        try:
            return cls.from_json_payload(json.loads(json_str))
        except ValueError as e:
            raise Exception(f"{cls.__name__} JSON parsing failed: Invalid JSON structure") from e

    @classmethod
    @abstractmethod
    def from_json_payload(cls, payload: dict) -> Self:
        """Create object instance from parsed JSON payload.

        Args:
            payload: parsed JSON dictionary

        Returns:
            Object instance

        """

    def to_json(self) -> str:
        """Create JSON string of object payload.

        Returns:
            A JSON representation of this message

        """
        return json.dumps(self.get_json_payload())

    @abstractmethod
    def get_json_payload(self) -> dict:
        """Get object payload for JSON representation"""

from_json(json_str) classmethod

Parse a JSON string into an object instance.

Parameters:

Name Type Description Default
json_str str

JSON string

 required

Returns:

Type Description
Self

An instance representation of this JSON
Source code in hiero_did_sdk_python/utils/serializable.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
@classmethod
def from_json(
    cls,
    json_str: str,
) -> Self:
    """Parse a JSON string into an object instance.

    Args:
        json_str: JSON string

    Returns:
        An instance representation of this JSON

    """
    try:
        return cls.from_json_payload(json.loads(json_str))
    except ValueError as e:
        raise Exception(f"{cls.__name__} JSON parsing failed: Invalid JSON structure") from e

from_json_payload(payload) abstractmethod classmethod

Create object instance from parsed JSON payload.

Parameters:

Name Type Description Default
payload dict

parsed JSON dictionary

 required

Returns:

Type Description
Self

Object instance
Source code in hiero_did_sdk_python/utils/serializable.py 
26
27
28
29
30
31
32
33
34
35
36
37
@classmethod
@abstractmethod
def from_json_payload(cls, payload: dict) -> Self:
    """Create object instance from parsed JSON payload.

    Args:
        payload: parsed JSON dictionary

    Returns:
        Object instance

    """

get_json_payload() abstractmethod

Get object payload for JSON representation

Source code in hiero_did_sdk_python/utils/serializable.py 
48
49
50
@abstractmethod
def get_json_payload(self) -> dict:
    """Get object payload for JSON representation"""

to_json()

Create JSON string of object payload.

Returns:

Type Description
str

A JSON representation of this message
Source code in hiero_did_sdk_python/utils/serializable.py 
39
40
41
42
43
44
45
46
def to_json(self) -> str:
    """Create JSON string of object payload.

    Returns:
        A JSON representation of this message

    """
    return json.dumps(self.get_json_payload())