Time-Based Storage System

A Python package providing two implementations of a time-based storage system for managing events with timestamps. This library is useful for applications that need to efficiently store and query time-based data, such as event logs, time series data, monitoring systems, and schedulers.

Features

Three storage implementations:
- TimeBasedStorage: Uses a dictionary for simple key-value access
- TimeBasedStorageHeap: Uses a heap for efficient insertion and earliest event access
- TimeBasedStorageRBTree: Uses a Red-Black Tree for balanced performance (O(log n) insertions and efficient range queries)
Thread-safe variants:
- ThreadSafeTimeBasedStorage: Thread-safe version of TimeBasedStorage
- ThreadSafeTimeBasedStorageHeap: Thread-safe version of TimeBasedStorageHeap
- ThreadSafeTimeBasedStorageRBTree: Thread-safe version of TimeBasedStorageRBTree
Support for:
- Event creation and deletion
- Range queries
- Duration-based queries
- Day-of-week queries
- Earliest/latest event access
- Timestamp collision handling
- Generic typing (store any data type)

Documentation

Comprehensive documentation is available to help you get the most out of this library:

Architecture Guide - Design principles, implementation details, and performance considerations
Code Examples - Practical usage examples and patterns
Concurrent Use Cases - Real-world scenarios for concurrent access
Alternative Approaches - Limitations of current implementation and alternative storage strategies

The rest of this README provides an overview of installation, basic usage, and API reference.

Installation

pip install time_based_storage

Basic Usage

from datetime import datetime, timedelta
from time_based_storage import TimeBasedStorage, TimeBasedStorageHeap

# Create storage instances
storage = TimeBasedStorage[str]()  # Type annotation for stored values
heap_storage = TimeBasedStorageHeap[str]()

# Add events with timestamps
storage.add(datetime(2024, 1, 1, 10, 0), "Event 1")
storage.add(datetime(2024, 1, 1, 11, 0), "Event 2")
storage.add(datetime(2024, 1, 1, 12, 0), "Event 3")

# Query events in a time range
start_time = datetime(2024, 1, 1, 10, 30)
end_time = datetime(2024, 1, 1, 11, 30)
events_in_range = storage.get_range(start_time, end_time)  # Returns ["Event 2"]

# Query events within a duration (last hour)
duration = 3600  # seconds
recent_events = storage.get_duration(duration)

# Get all events and timestamps
all_events = storage.get_all()
all_timestamps = storage.get_timestamps()

# Remove an event
storage.remove(datetime(2024, 1, 1, 11, 0))  # Removes "Event 2"

# Clear all events
storage.clear()

Thread-Safe Usage

For multithreaded applications, use the thread-safe variants:

from time_based_storage import ThreadSafeTimeBasedStorage, ThreadSafeTimeBasedStorageHeap
import threading

# Create thread-safe storage
storage = ThreadSafeTimeBasedStorage[int]()

# Use in multiple threads
def producer():
    for i in range(10):
        storage.add(datetime.now(), i)
        
def consumer():
    # Wait for data with timeout
    if storage.wait_for_data(timeout=1.0):
        data = storage.get_all()
        print(f"Received: {data}")

# Start threads
producer_thread = threading.Thread(target=producer)
consumer_thread = threading.Thread(target=consumer)
producer_thread.start()
consumer_thread.start()

Choosing the Right Implementation