State management and persistent storage for notification services supporting various storage modes and data persistence. PersistentStore provides namespace-based storage with support for different storage backends including memory-only and file-based persistence.
Manages persistent storage with namespace support and configurable storage modes for caching and state management.
class PersistentStore:
def __init__(self, path=None, namespace='default', mode=None):
"""
Initialize persistent storage.
Parameters:
- path (str, optional): File path for persistent storage (None for memory-only)
- namespace (str): Storage namespace for data organization
- mode (PersistentStoreMode, optional): Storage mode (AUTO, FLUSH, MEMORY)
"""
def read(self, key=None, compress=True, expires=False):
"""
Read data from storage.
Parameters:
- key (str, optional): Specific key to read (None for all data)
- compress (bool): Whether to decompress data
- expires (bool): Whether to check expiration
Returns:
dict|any: Stored data
"""
def write(self, data, key=None, compress=True):
"""
Write data to storage.
Parameters:
- data (any): Data to store
- key (str, optional): Storage key (None for root level)
- compress (bool): Whether to compress data
Returns:
bool: True if data was written successfully
"""
def get(self, key, expires=False):
"""
Get cached value by key.
Parameters:
- key (str): Cache key
- expires (bool): Check expiration
Returns:
any: Cached value or None if not found
"""
def set(self, key, value, expires=False):
"""
Set cached value.
Parameters:
- key (str): Cache key
- value (any): Value to cache
- expires (bool): Set expiration
Returns:
bool: True if value was set successfully
"""
def clear(self, *args):
"""
Clear stored data.
Parameters:
- *args: Keys to clear (empty for all data)
Returns:
bool: True if data was cleared successfully
"""
def flush(self, key=None, **kwargs):
"""
Flush data to persistent storage.
Parameters:
- key (str, optional): Specific key to flush
- **kwargs: Additional flush options
Returns:
bool: True if flush was successful
"""
def prune(self):
"""
Remove expired entries from storage.
Returns:
int: Number of entries removed
"""
def size(self, **kwargs):
"""
Get storage size information.
Parameters:
- **kwargs: Size calculation options
Returns:
int: Storage size in bytes
"""
@property
def mode(self):
"""
Current storage mode.
Returns:
PersistentStoreMode: Active storage mode
"""
@property
def path(self):
"""
Storage file path.
Returns:
str: Path to storage file or None for memory-only storage
"""
@property
def namespace(self):
"""
Storage namespace.
Returns:
str: Current namespace
"""from apprise import PersistentStore, PersistentStoreMode
# Create file-based storage
store = PersistentStore(
path='/tmp/apprise_storage.db',
namespace='notifications',
mode=PersistentStoreMode.AUTO
)
# Store data
store.write({'last_notification': '2023-01-01'}, key='status')
# Read data
status = store.read(key='status')
print(status) # {'last_notification': '2023-01-01'}
# Cache operations
store.set('temp_token', 'abc123')
token = store.get('temp_token')
# Clear specific data
store.clear('temp_token')
# Flush to disk
store.flush()# Create memory-only storage
memory_store = PersistentStore(
namespace='temp',
mode=PersistentStoreMode.MEMORY
)
# Data exists only during runtime
memory_store.set('session_id', 'xyz789')
session = memory_store.get('session_id')# Different namespaces for different purposes
user_store = PersistentStore(namespace='users')
config_store = PersistentStore(namespace='config')
# Data is isolated by namespace
user_store.write({'user_id': 123}, key='current_user')
config_store.write({'theme': 'dark'}, key='ui_settings')