"""
StateStorage - Persistent state storage for Qtile.
This module provides a simple key-value storage that persists across Qtile
config reloads and restarts. Data is stored as JSON in the Qtile cache directory.
Features:
- Automatic loading/saving of state
- Thread-safe operations
- Support for primitive types (str, int, float, bool, list, dict)
- Namespace support for organizing state
- Hook integration for automatic save on config reload
Usage:
from qtile_expanded.storage import StateStorage
# Create a storage instance
storage = StateStorage("my_app")
# Set values
storage.set("counter", 42)
storage.set("settings", {"theme": "dark", "notifications": True})
# Get values
counter = storage.get("counter", default=0)
settings = storage.get("settings", default={})
# Delete values
storage.delete("counter")
# Save explicitly
storage.save()
# Use as context manager (auto-saves)
with storage:
storage.set("temporary", "value")
"""
import json
import os
import threading
from pathlib import Path
from typing import Any, Optional, Union
from libqtile.log_utils import logger
[docs]
class StateStorage:
"""
Persistent key-value storage for Qtile that survives config reloads.
Data is stored as JSON in ~/.cache/qtile/qtile_expanded/{namespace}.json
Args:
namespace: Unique identifier for this storage (e.g., "my_widget")
cache_dir: Override the Qtile cache directory (default: ~/.cache/qtile)
auto_save: Whether to automatically save after each write (default: True)
Attributes:
data: The loaded state dictionary
"""
_lock = threading.Lock()
_instances: dict[str, "StateStorage"] = {}
[docs]
def __init__(
self,
namespace: str,
cache_dir: Optional[Union[str, Path]] = None,
auto_save: bool = True,
):
"""Initialize StateStorage with a namespace."""
self.namespace = namespace
self.auto_save = auto_save
# Determine cache directory
if cache_dir is None:
cache_dir = os.path.expanduser("~/.cache/qtile")
self.cache_dir = Path(cache_dir) / "qtile_expanded"
self.file_path = self.cache_dir / f"{namespace}.json"
# Ensure directory exists
self.cache_dir.mkdir(parents=True, exist_ok=True)
# Load existing data
self.data: dict[str, Any] = {}
self._load()
# Register instance for hook-based saving
StateStorage._instances[namespace] = self
def _load(self) -> None:
"""Load state from the JSON file."""
if self.file_path.exists():
try:
with open(self.file_path, "r", encoding="utf-8") as f:
self.data = json.load(f)
logger.debug(
f"StateStorage[{self.namespace}]: Loaded {len(self.data)} items"
)
except (json.JSONDecodeError, IOError) as e:
logger.warning(
f"StateStorage[{self.namespace}]: Could not load state: {e}"
)
self.data = {}
else:
logger.debug(
f"StateStorage[{self.namespace}]: No existing state file, starting fresh"
)
[docs]
def save(self) -> None:
"""Save state to the JSON file."""
with StateStorage._lock:
try:
with open(self.file_path, "w", encoding="utf-8") as f:
json.dump(self.data, f, indent=2, default=str)
logger.debug(
f"StateStorage[{self.namespace}]: Saved {len(self.data)} items"
)
except IOError as e:
logger.warning(
f"StateStorage[{self.namespace}]: Could not save state: {e}"
)
[docs]
def set(self, key: str, value: Any) -> None:
"""
Set a value in the storage.
Args:
key: The key to store the value under
value: The value to store (must be JSON-serializable)
"""
self.data[key] = value
if self.auto_save:
self.save()
[docs]
def get(self, key: str, default: Any = None) -> Any:
"""
Get a value from the storage.
Args:
key: The key to retrieve
default: Default value if key doesn't exist
Returns:
The stored value, or default if not found
"""
return self.data.get(key, default)
[docs]
def delete(self, key: str) -> bool:
"""
Delete a value from the storage.
Args:
key: The key to delete
Returns:
True if the key existed and was deleted, False otherwise
"""
if key in self.data:
del self.data[key]
if self.auto_save:
self.save()
return True
return False
[docs]
def clear(self) -> None:
"""Clear all data from the storage."""
self.data = {}
if self.auto_save:
self.save()
[docs]
def has(self, key: str) -> bool:
"""Check if a key exists in the storage."""
return key in self.data
[docs]
def keys(self) -> list[str]:
"""Get all keys in the storage."""
return list(self.data.keys())
[docs]
def items(self) -> list[tuple[str, Any]]:
"""Get all key-value pairs in the storage."""
return list(self.data.items())
[docs]
def values(self) -> list[Any]:
"""Get all values in the storage."""
return list(self.data.values())
[docs]
def increment(self, key: str, amount: int = 1) -> int:
"""
Increment a numeric value in the storage.
Args:
key: The key to increment
amount: The amount to increment by (default: 1)
Returns:
The new value
"""
current = self.get(key, 0)
if not isinstance(current, (int, float)):
current = 0
new_value = current + amount
self.set(key, new_value)
return new_value
[docs]
def decrement(self, key: str, amount: int = 1) -> int:
"""
Decrement a numeric value in the storage.
Args:
key: The key to decrement
amount: The amount to decrement by (default: 1)
Returns:
The new value
"""
return self.increment(key, -amount)
[docs]
def toggle(self, key: str, default: bool = False) -> bool:
"""
Toggle a boolean value in the storage.
Args:
key: The key to toggle
default: Default value if key doesn't exist
Returns:
The new value
"""
current = self.get(key, default)
new_value = not current
self.set(key, new_value)
return new_value
[docs]
def update(self, other: dict[str, Any]) -> None:
"""
Update the storage with multiple key-value pairs.
Args:
other: Dictionary of key-value pairs to update
"""
self.data.update(other)
if self.auto_save:
self.save()
[docs]
def __enter__(self) -> "StateStorage":
"""Context manager entry."""
return self
[docs]
def __exit__(self, exc_type, exc_val, exc_tb) -> None:
"""Context manager exit - save on exit."""
if self.auto_save:
self.save()
[docs]
def __contains__(self, key: str) -> bool:
"""Check if key exists (supports 'in' operator)."""
return self.has(key)
[docs]
def __getitem__(self, key: str) -> Any:
"""Get value using bracket notation."""
return self.data[key]
[docs]
def __setitem__(self, key: str, value: Any) -> None:
"""Set value using bracket notation."""
self.set(key, value)
[docs]
def __delitem__(self, key: str) -> None:
"""Delete value using bracket notation."""
self.delete(key)
[docs]
def __len__(self) -> int:
"""Get number of items in storage."""
return len(self.data)
def __repr__(self) -> str:
"""String representation."""
return f"StateStorage(namespace='{self.namespace}', items={len(self.data)})"
[docs]
@classmethod
def save_all(cls) -> None:
"""Save all registered StateStorage instances."""
for instance in cls._instances.values():
instance.save()
[docs]
@classmethod
def get_instance(cls, namespace: str, **kwargs) -> "StateStorage":
"""
Get or create a StateStorage instance with the given namespace.
This ensures only one instance per namespace exists.
Args:
namespace: The storage namespace
**kwargs: Additional arguments passed to StateStorage constructor
Returns:
The StateStorage instance
"""
if namespace not in cls._instances:
cls._instances[namespace] = cls(namespace, **kwargs)
return cls._instances[namespace]
[docs]
def setup_qtile_hooks(qtile) -> None:
"""
Set up Qtile hooks to automatically save state before reload/shutdown.
This function should be called from your Qtile config:
from qtile_expanded.storage import setup_qtile_hooks
def main(q):
setup_qtile_hooks(q)
# ... rest of your config
Args:
qtile: The Qtile instance (usually 'q' in config.py)
"""
from libqtile import hook
@hook.subscribe.startup_once
def on_startup():
logger.info("StateStorage: Qtile started, state loaded")
@hook.subscribe.reload
def on_reload():
logger.info("StateStorage: Config reloading, saving state...")
StateStorage.save_all()
@hook.subscribe.shutdown
def on_shutdown():
logger.info("StateStorage: Qtile shutting down, saving state...")
StateStorage.save_all()