1+ """Collection utilities: adaptive buffers and expiring key-value stores.
2+
3+ Provides AdaptiveBuffer for rate-aware batching and ExpiringDict for
4+ in-memory caches with optional TTL and persistence.
5+ """
16import json
27import logging
38import os
611
712logger = logging .getLogger (__name__ )
813
14+
915class AdaptiveBuffer :
10- """Buffer that adapts its flush size based on item arrival rate."""
16+ """Buffer that adapts its flush size based on item arrival rate.
17+
18+ When items arrive quickly (intervals <= fast_interval), flushes at max_size
19+ for efficiency. When items arrive slowly (intervals >= slow_interval),
20+ flushes at min_size for responsiveness. Uses EMA of inter-arrival intervals
21+ to interpolate target size between these extremes.
22+ """
1123
1224 def __init__ (
1325 self ,
@@ -18,6 +30,7 @@ def __init__(
1830 smoothing = 0.2 ,
1931 time_func = time .time ,
2032 ):
33+ """Initialize the buffer with size and timing parameters."""
2134 if min_size < 1 :
2235 raise ValueError ("min_size must be >= 1" )
2336 if max_size < min_size :
@@ -42,6 +55,7 @@ def __init__(
4255 self ._target_size = min_size
4356
4457 def append (self , item ):
58+ """Add an item; returns flushed batch if target size reached, else None."""
4559 now = self .time_func ()
4660 if self ._last_append_time is not None :
4761 interval = max (0.0 , now - self ._last_append_time )
@@ -55,6 +69,7 @@ def append(self, item):
5569 return None
5670
5771 def extend (self , items ):
72+ """Add multiple items; returns list of any flushed batches."""
5873 flushed = []
5974 for item in items :
6075 batch = self .append (item )
@@ -63,6 +78,7 @@ def extend(self, items):
6378 return flushed
6479
6580 def flush (self ):
81+ """Force flush and return all buffered items, or None if empty."""
6682 if not self ._buffer :
6783 return None
6884 items = self ._buffer
@@ -73,20 +89,23 @@ def __len__(self):
7389 return len (self ._buffer )
7490
7591 def _update_interval (self , interval ):
92+ """Update EMA of inter-arrival interval using smoothing factor."""
7693 if self ._ema_interval is None :
7794 self ._ema_interval = interval
7895 else :
7996 alpha = self .smoothing
8097 self ._ema_interval = (alpha * interval ) + ((1 - alpha ) * self ._ema_interval )
8198
8299 def _compute_target_size (self ):
100+ """Compute target flush size from EMA interval (min_size to max_size)."""
83101 if self ._ema_interval is None :
84102 return self .min_size
85103 if self ._ema_interval <= self .fast_interval :
86104 return self .max_size
87105 if self ._ema_interval >= self .slow_interval :
88106 return self .min_size
89107
108+ # Linear interpolation: faster arrivals -> larger target
90109 ratio = (self .slow_interval - self ._ema_interval ) / (
91110 self .slow_interval - self .fast_interval
92111 )
@@ -95,7 +114,14 @@ def _compute_target_size(self):
95114
96115
97116class ExpiringDict (UserDict ):
117+ """Dict-like store with per-key TTL and optional JSON persistence.
118+
119+ Keys expire after their TTL (seconds). If filename is set, the dict is
120+ saved to disk on every mutation and loaded on init.
121+ """
122+
98123 def __init__ (self , filename = None , default_ttl = None ):
124+ """Initialize with optional persistence path and default TTL in seconds."""
99125 super ().__init__ ()
100126 self .default_ttl = default_ttl
101127 self .filename = filename
@@ -105,6 +131,7 @@ def __init__(self, filename=None, default_ttl=None):
105131 self ._load ()
106132
107133 def __setitem__ (self , key , value , ttl = None ):
134+ """Set key to value; ttl overrides default_ttl if provided."""
108135 self .data [key ] = value
109136
110137 if ttl is None :
@@ -117,6 +144,7 @@ def __setitem__(self, key, value, ttl=None):
117144 self ._save () # Save when a key is set
118145
119146 def __delitem__ (self , key ):
147+ """Delete key and its expiry entry."""
120148 super ().__delitem__ (key )
121149 if key in self .expiry :
122150 del self .expiry [key ]
@@ -125,6 +153,7 @@ def __delitem__(self, key):
125153 self ._save () # Save when a key is deleted
126154
127155 def __getitem__ (self , key ):
156+ """Get value; raises KeyError if missing or expired."""
128157 self ._clean_expired ()
129158 if key in self .expiry and time .time () > self .expiry [key ]:
130159 del self .data [key ]
@@ -135,6 +164,7 @@ def __getitem__(self, key):
135164 return self .data [key ]
136165
137166 def set_with_ttl (self , key , value , ttl ):
167+ """Convenience method to set a key with explicit TTL."""
138168 self .__setitem__ (key , value , ttl )
139169
140170 def clear (self ):
@@ -165,7 +195,7 @@ def popitem(self):
165195 return key , value
166196
167197 def _clean_expired (self ):
168- """Remove all expired keys"""
198+ """Remove all expired keys from data and expiry maps. """
169199 now = time .time ()
170200 expired = [k for k , exp in self .expiry .items () if now > exp ]
171201 if expired : # Only save if something was expired
@@ -197,6 +227,7 @@ def __contains__(self, key):
197227 return key in self .data
198228
199229 def _save (self ):
230+ """Persist data and expiry to JSON file via atomic write."""
200231 if self .filename :
201232 # Use .tmp file and atomic rename for safety
202233 tmp_filename = str (self .filename ) + '.tmp'
@@ -214,6 +245,7 @@ def _save(self):
214245 raise
215246
216247 def _load (self ):
248+ """Load data and expiry from JSON file; on failure, start empty."""
217249 try :
218250 if os .path .exists (self .filename ):
219251 with open (self .filename , 'r' ) as f :
0 commit comments