API Reference
This section provides detailed API documentation for True Storage’s core modules and components.
Core Components
Storage backends including Hot, Cold, and Mixed storage implementations.
Session management and persistence functionality.
Environment variable handling and configuration management.
Database integration and persistence layer.
Module Reference
Storage Module
Storage package providing flexible data storage solutions.
This package provides a comprehensive set of storage implementations for different use cases, from fast in-memory caching to persistent disk storage.
- Classes:
HotStorage: Fast in-memory storage with LRU caching. ColdStorage: Persistent disk storage with compression. MixedStorage: Hybrid storage combining hot and cold storage benefits.
- Key Features:
Multiple storage implementations for different needs
Thread-safe operations across all storage types
Configurable policies and optimization strategies
Performance monitoring and metrics
Event-driven architecture
Automatic resource management
- class true_storage.storage.HotStorage(storage_id='hot_storage', max_size=100, expiration_time=300, policy=StoragePolicy.STRICT)[source]
Bases:
BaseStorageManagerHot storage implementation with LRU cache and advanced features.
- __init__(storage_id='hot_storage', max_size=100, expiration_time=300, policy=StoragePolicy.STRICT)[source]
- class true_storage.storage.ColdStorage(storage_id='cold_storage', storage_directory='cold_storage', compression_level=6, policy=StoragePolicy.STRICT)[source]
Bases:
BaseStorageManagerCold storage implementation with compression and advanced features.
- __init__(storage_id='cold_storage', storage_directory='cold_storage', compression_level=6, policy=StoragePolicy.STRICT)[source]
- class true_storage.storage.MixedStorage(storage_id='mixed_storage', max_size=100, expiration_time=300, storage_directory='cold_storage', compression_level=6, policy=StoragePolicy.STRICT)[source]
Bases:
BaseStorageManagerMixed storage implementation combining hot and cold storage.
- __init__(storage_id='mixed_storage', max_size=100, expiration_time=300, storage_directory='cold_storage', compression_level=6, policy=StoragePolicy.STRICT)[source]
- optimize_hot_storage()[source]
Optimize hot storage by removing least accessed items.
- Return type:
Session Module
Session management module for in-memory data storage with persistence capabilities.
This module provides a robust implementation of a session store with features like expiration, cleanup, and persistence. It offers thread-safe operations and LRU (Least Recently Used) eviction strategy.
- Classes:
SessionStatus: Enumeration of possible session states (ACTIVE, EXPIRED, LOCKED). SessionMetadata: Data class containing metadata for session entries. SessionStoreConfig: Configuration class for SessionStore settings. SessionStore: Main class implementing the session storage functionality.
- Functions:
None
- Types:
None
- Exceptions:
StorageError: Raised when storage operations fail
- Key Features:
Thread-safe operations with lock mechanism
Automatic session expiration and cleanup
LRU (Least Recently Used) eviction strategy
Session persistence to disk with atomic writes
Session locking for exclusive access
Dict-like interface with familiar operations
Configurable logging
Background cleanup and backup processes
- class true_storage.session.SessionStatus(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]
Bases:
EnumSession status enumeration.
- ACTIVE = 'active'
- EXPIRED = 'expired'
- LOCKED = 'locked'
- class true_storage.session.SessionMetadata(created_at=<factory>, last_accessed=<factory>, access_count=0, status=SessionStatus.ACTIVE, lock_expiry=None)[source]
Bases:
objectMetadata for session entries.
-
created_at:
float
-
last_accessed:
float
-
access_count:
int= 0
-
status:
SessionStatus= 'active'
- __init__(created_at=<factory>, last_accessed=<factory>, access_count=0, status=SessionStatus.ACTIVE, lock_expiry=None)
-
created_at:
- class true_storage.session.SessionStoreConfig(max_size=1000, expiration_time=3600, cleanup_interval=60, persistence_path=None, backup_interval=300, max_lock_time=30, enable_logging=True, log_level=20)[source]
Bases:
objectConfiguration for SessionStore.
-
max_size:
int= 1000
-
expiration_time:
int= 3600
-
cleanup_interval:
int= 60
-
backup_interval:
int= 300
-
max_lock_time:
int= 30
-
enable_logging:
bool= True
-
log_level:
int= 20
- __init__(max_size=1000, expiration_time=3600, cleanup_interval=60, persistence_path=None, backup_interval=300, max_lock_time=30, enable_logging=True, log_level=20)
-
max_size:
- class true_storage.session.SessionStore(config=None)[source]
Bases:
objectA robust and thread-safe in-memory session store with expiration, LRU eviction, and persistence.
- __init__(config=None)[source]
- _start_background_threads()[source]
Initialize and start background threads.
- set(key, value, expiration=None)[source]
Set a session key to a value with optional custom expiration.
- Return type:
- get(key, default=None)[source]
Retrieve a session value by key with metadata update.
- Return type:
- _backup_sessions()[source]
Periodically backup sessions to disk if persistence is enabled.
- Return type:
- __del__()[source]
Ensure all threads are stopped and final backup is performed.
- delete(key)[source]
Delete a session key. Returns True if the key was deleted, False if not found.
- Return type:
- items()[source]
Return an iterator over the session items (key, value).
- _cleanup_expired_sessions()[source]
Background thread method to clean up expired sessions periodically.
- Return type:
Environment Module
Advanced environment configuration and management.
This module provides comprehensive control over environment variables.
- Classes:
Environment: Main class for managing environment configurations. MODES: Enum defining operational modes for environment management.
- Functions:
to_settings: Converts Environment instance to pydantic_settings v2 BaseSettings.
- Types:
EnvPath: Union type for environment file paths.
- Exceptions:
EnvError: Base exception for environment errors. ValidationError: Exception raised for environment validation errors. ModeError: Exception raised for mode-related errors.
- Key Features:
Multiple environment sources (env files, JSON, config files)
Environment validation and type checking
Secure secret management
Environment inheritance and layering
Variable interpolation
Environment snapshots and rollback
Mode-specific environment variables
- class true_storage.env.Environment(env_data='.env', validator=None, parent=None, interpolate=True, mode=<MODES.DEV: dev>, *extenal_envs)[source]
Bases:
objectAdvanced environment configuration and management system.
This class provides a comprehensive environment variable management system with features like mode-specific variables, secure storage, and variable validation.
- Variables:
_mode (MODES) – Current environment mode.
_instance (Environment) – Singleton instance of the environment.
_lock (threading.Lock) – Thread lock for singleton pattern.
_mode_vars (Dict[MODES, Set[str]]) – Mode-specific variable tracking.
_secure_mode_mappings (Dict[str, Set[MODES]]) – Secure storage of mode mappings.
- Parameters:
env_data (EnvPath, optional) – Source of environment data. Defaults to “.env”.
validator (EnvValidatorProtocol, optional) – Validator for environment variables.
parent (Environment, optional) – Parent environment for inheritance.
interpolate (bool, optional) – Enable variable interpolation. Defaults to True.
- static __new__(cls, *args, **kwargs)[source]
Implement thread-safe singleton pattern.
- __init__(env_data='.env', validator=None, parent=None, interpolate=True, mode=<MODES.DEV: dev>, *extenal_envs)[source]
- property envpath
- property externalenvs
- property mode_mappings: Dict[str, Set[MODES]]
Get a secure copy of the mode-to-variable mappings.
- Returns:
A mapping of variable names to their allowed modes.
- Return type:
Dict[str, Set[MODES]]
- property mode: MODES
Get current environment mode.
- property parent: Environment | None
Get parent environment.
- property snapshots: List[EnvSnapshot]
Get list of environment snapshots.
- mark(mode)[source]
Decorator for mode-specific function execution.
- Parameters:
mode (MODES) – Mode to execute the function in.
- Returns:
Decorated function that executes in specified mode.
- Return type:
Callable
Example
>>> env = Environment() >>> @env.mark(MODES.TEST) ... def test_function(): ... return env.get('TEST_CONFIG') # Only accessible in test mode
- with_mode(mode)[source]
Context manager for temporary mode switching.
- Parameters:
mode (MODES) – Mode to temporarily switch to.
- Yields:
Environment – Self with temporarily changed mode.
- Return type:
ModeContext
Example
>>> env = Environment() >>> with env.with_mode(MODES.PROD): ... secret = env.get('API_KEY') # Access production-only variable
- _is_mode_var(key, mode=None)[source]
Check if a variable belongs to a specific mode.
- Return type:
- mark_as_mode_var(key, mode)[source]
Mark a variable as belonging to a specific mode.
- Return type:
- _get_mode_key(key, mode=None)[source]
Generate a mode-specific key for an environment variable.
- static _get_base_key(key)[source]
Extract the base key from a mode-specific key.
- is_allowed_in_mode(key, mode)[source]
Check if a variable is allowed in a specific mode.
- static create_snapshot()[source]
Create a snapshot of the current environment state.
- Returns:
Snapshot containing current variable values.
- Return type:
EnvSnapshot
Example
>>> env = Environment() >>> snapshot = env.create_snapshot() >>> env.set({'DEBUG': 'false'}) >>> env.rollback(snapshot) # Restore previous state
- static rollback(snapshot)[source]
Rollback environment to a previous snapshot.
- Parameters:
snapshot (EnvSnapshot) – Snapshot to restore from.
- Return type:
Example
>>> env = Environment() >>> snapshot = env.create_snapshot() >>> env.set({'DEBUG': 'false'}) >>> env.rollback(snapshot) # Restore DEBUG to previous value
- get(key, default=None, mode=None)[source]
Retrieve an environment variable with mode support.
- Parameters:
key (str) – The variable name to retrieve.
mode (MODES) – To specify a mode or it will go for current mode.
default (Any, optional) – Default value if variable not found. Defaults to None.
- Returns:
The value of the environment variable.
- Return type:
- Raises:
ModeError – If the variable is not accessible in the current mode.
- set(items, system_env=False, modes=None)[source]
Set environment variables with mode-specific access control.
- Return type:
- _validate_and_set_value(key, value, system_env, modes)[source]
Validate and set a single environment variable.
- Return type:
- _set_value_in_environments(key, value, system_env, modes)[source]
Set the value in appropriate environments based on modes.
- Return type:
- delete(key, modes=None)[source]
Delete an environment variable from specified modes.
- Parameters:
key (str) – The variable name to delete.
modes (List[MODES], optional) – List of modes to delete from. If None, deletes from all modes.
- Return type:
Example
>>> env = Environment() >>> env.delete('DEBUG', modes=[MODES.PROD]) # Remove from production only >>> env.delete('API_KEY') # Remove from all modes
- static _normalize_modes(modes)[source]
Normalize the modes list to handle None case.
- Parameters:
modes (List[MODES], optional) – List of modes to normalize.
- Returns:
All available modes if input is None, otherwise the input list.
- Return type:
List[MODES]
- _delete_from_env(key, modes)[source]
Delete the variable from specified modes in the environment.
- _delete_common_variable(key)[source]
Delete a common (non-mode-specific) variable.
- _delete_mode_specific_variable(key, mode)[source]
Delete a mode-specific variable.
- _update_secure_mappings(key, modes)[source]
Update the secure mode mappings after variable deletion.
- format_debug(batch_size=10)[source]
Format environment data for debugging in batches.
- Return type:
- _format_batch(title, items, batch_size, start=0)[source]
Format a batch of items with a title.
- Return type:
- __getitem__(key)[source]
Get environment variable using dictionary-style access with mode support.
- Return type:
- __setitem__(key, value)[source]
Set environment variable using dictionary-style access with mode support.
- Return type:
- __delitem__(key)[source]
Delete environment variable using dictionary-style access with mode support.
- Return type:
- __contains__(key)[source]
Check if environment variable exists using ‘in’ operator with mode support.
- Return type:
- __iter__()[source]
Iterate over environment variables.
- filter(search_value, search_in='key', exclude_secrets=True, mode_specific=True)[source]
Filter environment variables with mode and secret support.
- filter_with_predicate(predicate, exclude_secrets=True, mode_specific=True)[source]
Filter environment variables with a predicate function.
- classmethod from_json(json_path, **kwargs)[source]
Create an Environment instance from a JSON file.
- Return type:
Environment
- classmethod from_dict(env_dict, **kwargs)[source]
Create an Environment instance from a dictionary.
- Return type:
Environment
- classmethod from_config(config_path, **kwargs)[source]
Create an Environment instance from a configuration file.
- Return type:
Environment
- validate_external_envs(extenal_envs)[source]
- write_env(env_path=None, flush=False)[source]
Write environment variables to a file, organized by mode.
This method writes the current environment variables to a file, organizing them by mode. It determines the output path, organizes the variables, formats them into sections, and then writes them to the specified file.
Notes
If env_path is None it will override the existing env file!
- Parameters:
- Return type:
- Returns:
None
- Raises:
- class true_storage.env.MODES(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]
-
Environment modes for configuration management.
This enum defines different operational modes for environment variable management, each with specific behaviors and access patterns.
- Variables:
- ALL = 'all'
- DEV = 'dev'
- TEST = 'test'
- STAGE = 'stage'
- PROD = 'prod'
- __init__(value)[source]
- classmethod _generate_next_value_(name, start, count, last_values)[source]
Generate the next value for enum members.
- property is_development: bool
Check if the current mode is a development mode.
- property is_production: bool
Check if the current mode is production mode.
- property is_all: bool
Check if the current mode is ALL mode.
- property prefix: str
Get the prefix for environment variables in this mode.
- property suffix: str
Get the suffix for environment variables in this mode.
- classmethod with_stages(**new_stages)[source]
Create a new MODES enum with additional custom stages.
- Parameters:
**new_stages (
str) – Keyword arguments of new stage names and their values- Return type:
- Returns:
A new MODES enum class with additional stages
- Raises:
ValueError – If a stage name conflicts with existing stages
- classmethod get_stage(name)[source]
Get a stage by name.
- Parameters:
name (
str) – Name of the stage to get- Return type:
MODES- Returns:
Either a MODES enum member or a CustomStage instance
- Raises:
ValueError – If stage doesn’t exist
- class true_storage.env.EnvValidator(schema)[source]
Bases:
objectEnvironment validator for type checking and validation.
- __init__(schema)[source]
- exception true_storage.env.EnvError[source]
Bases:
ExceptionException raised for environment errors.
- exception true_storage.env.ValidationError[source]
Bases:
EnvErrorException raised for environment validation errors.
- exception true_storage.env.ModeError[source]
Bases:
EnvErrorException raised for mode-related errors.
- true_storage.env.to_settings(env_instance, settings_class)[source]
Convert an Environment instance to a pydantic_settings v2 BaseSettings instance.
This allows optional pydantic compatibility without modifying the core Environment class.
- Parameters:
env_instance (Environment) – The Environment instance to convert
settings_class (Type[BaseSettings]) – The pydantic_settings BaseSettings class to convert to
- Returns:
An instance of the provided BaseSettings class
- Return type:
BaseSettings
Example
>>> from pydantic_settings import BaseSettings, SettingsConfigDict # Recommended V2 >>> from typing import Optional >>> >>> class MySettings(BaseSettings): ... app_name: str ... port: int ... debug: bool = False ... api_key: Optional[str] = None ... ... model_config = SettingsConfigDict( ... env_file='.env', ... env_prefix='', ... case_sensitive=False ... ) ... >>> env = Environment() >>> settings = to_settings(env, MySettings)
Environment Methods
- set(items: Dict[str, Any], system_env: bool = False, modes: Optional[List[MODES]] = None)
Set one or more environment variables.
- param items:
Dictionary of key-value pairs to set
- param system_env:
Whether to also set in system environment
- param modes:
List of modes where variables are accessible
- raises ValidationError:
If validation fails for any value
- raises ModeError:
If current mode not in allowed modes
- get(key: str, default: Any = None) -> Any
Get an environment variable value.
- param key:
Variable name to get
- param default:
Default value if not found
- returns:
Variable value or default
- raises ModeError:
If variable not accessible in current mode
- filter(prefix: str) -> Dict[str, str]
Filter variables by prefix.
- param prefix:
Prefix to filter by
- returns:
Dictionary of matching variables
Database Module
Database storage implementations.
This package provides a comprehensive set of storage implementations for different use cases.
- Classes:
FileSystemStorage: Local file system storage. RedisStorage: Redis-based storage. SQLiteStorage: SQLite database storage.
- Key Features:
Multiple storage implementations for different needs
Thread-safe operations across all storage types
Configurable policies and optimization strategies
Performance monitoring and metrics
Event-driven architecture
Automatic resource management
- class true_storage.database.SQLiteStorage(db_path=None)[source]
Bases:
BaseStorageSQLite-based storage implementation.
- __init__(db_path=None)[source]
Initialize SQLite storage.
- _get_connection()[source]
Get a SQLite connection, creating it if needed.
- _init_db()[source]
Initialize the database schema.
- close()[source]
Close the database connection.
- clone()[source]
Create a copy of the storage instance.
- Return type:
BaseStorage
- __del__()[source]
Ensure connection is closed on deletion.
- class true_storage.database.RedisStorage(host='localhost', port=6379, db=0, prefix='checkpoint:')[source]
Bases:
BaseStorageRedis-based storage implementation.
- __init__(host='localhost', port=6379, db=0, prefix='checkpoint:')[source]
- clone()[source]
Create a copy of the storage instance.
- Return type:
- class true_storage.database.FileSystemStorage(base_dir=None)[source]
Bases:
BaseStorageFile system-based storage implementation.
- __init__(base_dir=None)[source]
Initialize the file system storage.
- _get_path(key)[source]
Get a full path for a key.
- clone()[source]
Create a copy of the storage instance.
- Return type:
BaseStorage
- store(key, value)[source]
Store a value with the given key.
- retrieve(key)[source]
Retrieve a value by key.
- exists(key)[source]
Check if a key exists.
Note
For storage-related errors, see StorageError in the Exceptions section.
Utility Functions
Exceptions
Exceptions for the true_storage package.
- exception true_storage.exceptions.StorageError[source]
Bases:
ExceptionBase exception for storage-related errors.
- exception true_storage.exceptions.StorageConnectionError[source]
Bases:
StorageErrorRaised when a storage connection fails.
Note
The true_storage.exceptions.StorageError is the base exception class for all storage-related errors.
It is imported and used by both filesystem and SQLite storage modules.
See Also
Module Reference - Detailed module documentation
Examples and Tutorials - Usage examples
Troubleshooting Guide - Troubleshooting guide