Push Notifications that work with just about every platform!
—
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
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')Install with Tessl CLI
npx tessl i tessl/pypi-apprise