API Reference

This section provides detailed API documentation for True Storage’s core modules and components.

Core Components

Storage

Storage backends including Hot, Cold, and Mixed storage implementations.

Storage Module
Session

Session management and persistence functionality.

Session Module
Environment

Environment variable handling and configuration management.

Environment Module
Database

Database integration and persistence layer.

Database Module

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: BaseStorageManager

Hot storage implementation with LRU cache and advanced features.

__init__(storage_id='hot_storage', max_size=100, expiration_time=300, policy=StoragePolicy.STRICT)[source]
_is_expired(key)[source]

Check if a key has expired.

Return type:

bool

_remove_expired_items()[source]

Remove expired items from storage.

Return type:

None

store(key, value)[source]

Store a value in hot storage.

Return type:

None

retrieve(key)[source]

Retrieve a value from hot storage.

Return type:

Optional[Any]

delete(key)[source]

Delete a value from hot storage.

Return type:

None

clear()[source]

Clear all items from hot storage.

Return type:

None

get_size()[source]

Get the current size of hot storage.

Return type:

int

get_keys()[source]

Get all non-expired keys in hot storage.

Return type:

list[str]

get_stats()[source]

Get storage statistics.

Return type:

Dict[str, Any]

class true_storage.storage.ColdStorage(storage_id='cold_storage', storage_directory='cold_storage', compression_level=6, policy=StoragePolicy.STRICT)[source]

Bases: BaseStorageManager

Cold storage implementation with compression and advanced features.

__init__(storage_id='cold_storage', storage_directory='cold_storage', compression_level=6, policy=StoragePolicy.STRICT)[source]
_load_metadata()[source]

Load metadata from JSON file.

Return type:

None

_save_metadata()[source]

Save metadata to JSON file.

Return type:

None

store(key, value)[source]

Store a value in cold storage.

Return type:

None

retrieve(key)[source]

Retrieve a value from cold storage.

Return type:

Optional[Any]

delete(key)[source]

Delete a value from cold storage.

Return type:

None

clear()[source]

Clear all items from cold storage.

Return type:

None

get_size()[source]

Get the total size of cold storage in bytes.

Return type:

int

get_keys()[source]

Get all keys in cold storage.

Return type:

list[str]

get_stats()[source]

Get storage statistics.

Return type:

Dict[str, Any]

get_item_metadata(key)[source]

Get metadata for a specific item.

Return type:

Optional[Dict]

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: BaseStorageManager

Mixed 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]
store(key, value)[source]

Store a value in mixed storage.

Return type:

None

retrieve(key)[source]

Retrieve a value from mixed storage.

Return type:

Optional[Any]

delete(key)[source]

Delete a value from mixed storage.

Return type:

None

clear()[source]

Clear all items from mixed storage.

Return type:

None

get_hot_stats()[source]

Get hot storage statistics.

Return type:

Dict[str, Any]

get_cold_stats()[source]

Get cold storage statistics.

Return type:

Dict[str, Any]

get_stats()[source]

Get combined storage statistics.

Return type:

Dict[str, Any]

get_keys()[source]

Get all keys from both hot and cold storage.

Return type:

list[str]

optimize_hot_storage()[source]

Optimize hot storage by removing least accessed items.

Return type:

None

warm_up_hot_storage(keys)[source]

Pre-load specified keys into hot storage.

Return type:

None

class true_storage.storage.StoragePolicy(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

Storage policies for data management.

STRICT = 'strict'
LENIENT = 'lenient'
LAZY = 'lazy'

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: Enum

Session 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: object

Metadata for session entries.

created_at: float
last_accessed: float
access_count: int = 0
status: SessionStatus = 'active'
lock_expiry: Optional[float] = None
__init__(created_at=<factory>, last_accessed=<factory>, access_count=0, status=SessionStatus.ACTIVE, lock_expiry=None)
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: object

Configuration for SessionStore.

max_size: int = 1000
expiration_time: int = 3600
cleanup_interval: int = 60
persistence_path: Optional[str] = None
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)
class true_storage.session.SessionStore(config=None)[source]

Bases: object

A 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.

_setup_logging()[source]

Setup logging configuration.

Return type:

None

set(key, value, expiration=None)[source]

Set a session key to a value with optional custom expiration.

Return type:

None

get(key, default=None)[source]

Retrieve a session value by key with metadata update.

Return type:

Any

lock(key, duration=None)[source]

Lock a session key for exclusive access.

Return type:

bool

unlock(key)[source]

Unlock a session key.

Return type:

bool

_evict_lru_session()[source]

Evict the least recently used session.

Return type:

None

get_metadata(key)[source]

Get metadata for a session key.

Return type:

Optional[SessionMetadata]

get_status(key)[source]

Get the status of a session key.

Return type:

Optional[SessionStatus]

_backup_sessions()[source]

Periodically backup sessions to disk if persistence is enabled.

Return type:

None

_save_to_disk()[source]

Save sessions to disk.

Return type:

None

_restore_sessions()[source]

Restore sessions from disk on initialization.

Return type:

None

stop()[source]

Stop all background threads and perform final backup.

Return type:

None

__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:

bool

clear()[source]

Clear all sessions.

Return type:

None

keys()[source]

Return an iterator over the session keys.

Return type:

Iterator[Any]

values()[source]

Return an iterator over the session values.

Return type:

Iterator[Any]

items()[source]

Return an iterator over the session items (key, value).

Return type:

Iterator[tuple[Any, Any]]

_cleanup_expired_sessions()[source]

Background thread method to clean up expired sessions periodically.

Return type:

None

__setitem__(key, value)[source]

Enable dict-like setting of items.

Return type:

None

__getitem__(key)[source]

Enable dict-like getting of items.

Return type:

Any

__delitem__(key)[source]

Enable dict-like deletion of items.

Return type:

None

__contains__(key)[source]

Enable use of ‘in’ keyword to check for key existence.

Return type:

bool

__len__()[source]

Return the number of active (non-expired) sessions.

Return type:

int

exception true_storage.session.StorageError[source]

Bases: Exception

Base exception for storage-related errors.

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: object

Advanced 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 variables: Dict[str, str]

Get all environment variables.

property secrets: Dict[str, str]

Get all secret environment variables.

property non_secrets: Dict[str, str]

Get all non-secret environment variables.

property mode: MODES

Get current environment mode.

property parent: Environment | None

Get parent environment.

property snapshots: List[EnvSnapshot]

Get list of environment snapshots.

property mode_variables: Dict[str, str]

Get all variables specific to the current mode.

Returns:

Dictionary of mode-specific variables.

Return type:

Dict[str, str]

_interpolate_value(value)[source]

Interpolate environment variables in value.

Return type:

str

load_env()[source]

Load environment variables with inheritance and interpolation.

Return type:

None

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:

bool

mark_as_mode_var(key, mode)[source]

Mark a variable as belonging to a specific mode.

Return type:

None

_get_mode_key(key, mode=None)[source]

Generate a mode-specific key for an environment variable.

Parameters:
  • key (str) – Base variable name.

  • mode (MODES, optional) – Mode to generate key for. Defaults to current mode.

Returns:

Mode-specific variable key.

Return type:

str

static _get_base_key(key)[source]

Extract the base key from a mode-specific key.

Parameters:

key (str) – Mode-specific variable key.

Returns:

Base variable name without mode prefix/suffix.

Return type:

str

is_allowed_in_mode(key, mode)[source]

Check if a variable is allowed in a specific mode.

Parameters:
  • key (str) – The variable name to check.

  • mode (MODES) – The mode to check against.

Returns:

True if the variable is accessible in the specified mode, False otherwise.

Return type:

bool

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:

None

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:

str

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:

None

_validate_and_set_value(key, value, system_env, modes)[source]

Validate and set a single environment variable.

Return type:

None

_validate_value(key, value)[source]

Validate the value if a validator is present.

Return type:

Any

_set_value_in_environments(key, value, system_env, modes)[source]

Set the value in appropriate environments based on modes.

Return type:

None

_track_mode_variables(key, modes)[source]

Track variables for each mode.

Return type:

None

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:

None

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.

Parameters:
  • key (str) – The variable name to delete.

  • modes (List[MODES]) – List of modes to delete from.

Return type:

None

_delete_common_variable(key)[source]

Delete a common (non-mode-specific) variable.

Parameters:

key (str) – The variable name to delete.

Return type:

None

_delete_mode_specific_variable(key, mode)[source]

Delete a mode-specific variable.

Parameters:
  • key (str) – The variable name to delete.

  • mode (MODES) – The mode to delete from.

Return type:

None

_update_secure_mappings(key, modes)[source]

Update the secure mode mappings after variable deletion.

Parameters:
  • key (str) – The variable name that was deleted.

  • modes (List[MODES]) – The modes it was deleted from.

Return type:

None

__str__()[source]

Get string representation of environment state.

Return type:

str

__repr__()[source]

Get detailed string representation for debugging.

Return type:

str

format_debug(batch_size=10)[source]

Format environment data for debugging in batches.

Return type:

str

_format_basic_info()[source]

Format basic environment information.

Return type:

str

_format_variables_by_mode(batch_size)[source]

Format variables for each mode.

Return type:

str

_format_secrets(batch_size)[source]

Format secret variables.

Return type:

str

_format_mode_mappings()[source]

Format mode mappings.

Return type:

str

_format_snapshots()[source]

Format snapshots information.

Return type:

str

_format_batch(title, items, batch_size, start=0)[source]

Format a batch of items with a title.

Return type:

str

__getitem__(key)[source]

Get environment variable using dictionary-style access with mode support.

Return type:

str

__setitem__(key, value)[source]

Set environment variable using dictionary-style access with mode support.

Return type:

None

__delitem__(key)[source]

Delete environment variable using dictionary-style access with mode support.

Return type:

None

__contains__(key)[source]

Check if environment variable exists using ‘in’ operator with mode support.

Return type:

bool

__iter__()[source]

Iterate over environment variables.

__len__()[source]

Get number of unique base environment variables.

Return type:

int

filter(search_value, search_in='key', exclude_secrets=True, mode_specific=True)[source]

Filter environment variables with mode and secret support.

Return type:

Dict[str, str]

filter_with_predicate(predicate, exclude_secrets=True, mode_specific=True)[source]

Filter environment variables with a predicate function.

Return type:

Dict[str, str]

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:
  • env_path (Optional[str]) – The path to write the environment file. If None, a default path will be used.

  • flush (bool) – If True, flush the file buffer immediately after writing.

Return type:

None

Returns:

None

Raises:
  • IOError – If there’s an error writing to the file.

  • OSError – if os’s (system) error

class true_storage.env.MODES(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: str, Enum

Environment modes for configuration management.

This enum defines different operational modes for environment variable management, each with specific behaviors and access patterns.

Variables:
  • ALL (str) – Special mode for variables accessible across all modes.

  • DEV (str) – Development mode for local development environment.

  • TEST (str) – Testing mode for test environment.

  • STAGE (str) – Staging mode for pre-production environment.

  • PROD (str) – Production mode for live environment.

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:

Type[Enum]

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: object

Environment validator for type checking and validation.

__init__(schema)[source]
validate(key, value)[source]

Validate a value against the schema.

Parameters:
  • key (str) – Key to validate

  • value (Any) – Value to validate

Returns:

Validated value

Return type:

Any

Raises:

ValidationError – If validation fails

exception true_storage.env.EnvError[source]

Bases: Exception

Exception raised for environment errors.

exception true_storage.env.ValidationError[source]

Bases: EnvError

Exception raised for environment validation errors.

exception true_storage.env.ModeError[source]

Bases: EnvError

Exception 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: BaseStorage

SQLite-based storage implementation.

__init__(db_path=None)[source]

Initialize SQLite storage.

Parameters:

db_path (Optional[str]) – Path to SQLite database file. If None, uses in-memory database.

_get_connection()[source]

Get a SQLite connection, creating it if needed.

_init_db()[source]

Initialize the database schema.

store(key, value)[source]

Store a value in the database.

Return type:

None

retrieve(key)[source]

Retrieve a value from the database.

Return type:

Any

delete(key)[source]

Delete a value from the database.

Return type:

None

clear()[source]

Clear all values from the database.

Return type:

None

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: BaseStorage

Redis-based storage implementation.

__init__(host='localhost', port=6379, db=0, prefix='checkpoint:')[source]
_get_key(key)[source]

Get prefixed key for Redis.

Return type:

str

store(key, value)[source]

Store a value with the given key.

Return type:

None

retrieve(key)[source]

Retrieve a value by key.

Return type:

Any

delete(key)[source]

Delete a value by key.

Return type:

None

clear()[source]

Clear all stored values.

Return type:

None

clone()[source]

Create a copy of the storage instance.

Return type:

RedisStorage

class true_storage.database.FileSystemStorage(base_dir=None)[source]

Bases: BaseStorage

File system-based storage implementation.

__init__(base_dir=None)[source]

Initialize the file system storage.

Parameters:

base_dir (Optional[str]) – Base directory for storage. If None, uses a temporary directory.

_get_path(key)[source]

Get a full path for a key.

Parameters:

key (str) – Storage key, can include directory separators.

Return type:

Path

Returns:

Path object for the key.

clone()[source]

Create a copy of the storage instance.

Return type:

BaseStorage

store(key, value)[source]

Store a value with the given key.

Parameters:
  • key (str) – Storage key

  • value (Any) – Value to store

Return type:

None

retrieve(key)[source]

Retrieve a value by key.

Parameters:

key (str) – Storage key

Return type:

Any

Returns:

Stored value

Raises:
  • KeyError – If key doesn’t exist

  • StorageError – If retrieval fails

delete(key)[source]

Delete a value by key.

Parameters:

key (str) – Storage key

Return type:

None

clear()[source]

Clear all stored values.

Return type:

None

exists(key)[source]

Check if a key exists.

Parameters:

key (str) – Storage key

Return type:

bool

Returns:

True if key exists, False otherwise

list_keys(prefix='')[source]

List all keys with given prefix.

Parameters:

prefix (str) – Key prefix to filter by

Return type:

list[str]

Returns:

List of matching keys

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: Exception

Base exception for storage-related errors.

exception true_storage.exceptions.StorageConnectionError[source]

Bases: StorageError

Raised 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