PyStore

PyStore follows a hierarchy: Store $\rightarrow$ Collection $\rightarrow$ Item.

1. Set Path: Use pystore.set_path() to define where data is stored. Defaults to ~/pystore or the PYSTORE_PATH environment variable.
2. Connect: Use pystore.store("name") to connect to a datastore.
3. Collection: Use store.collection("name") to access a namespace (e.g., by frequency like EOD or ONEMINUTE).
4. Write: Use collection.write("item_name", dataframe, metadata={...}) to save data.
5. Read: Use collection.item("item_name") to retrieve an item. You can access the data as a Dask dataframe via .data, metadata via .metadata, or convert it to a Pandas DataFrame via .to_pandas().
6. Append: Use collection.append("item_name", dataframe) to add new rows to an existing item.

import pystore
import yfinance as yf

pystore.set_path("~/pystore")
store = pystore.store("mydatastore")
collection = store.collection("NASDAQ")

# Load and write data
aapl = yf.download("AAPL", multi_level_index=False)
collection.write("AAPL", aapl[:100], metadata={"source": "yfinance"})

# Read data
item = collection.item("AAPL")
df = item.to_pandas()

# Append data
collection.append("AAPL", aapl[100:])

PyStore provides two levels of transactional management:

1. Transaction: Provides full atomicity and rollback capabilities. It creates backups of existing items before modifying them. If an error occurs, it restores the backups to ensure the collection remains in a consistent state.
2. BatchTransaction: An optimization layer. It does not manage its own backups/rollbacks directly; instead, it collects all operations in memory and then executes them by wrapping them in a single standard Transaction. This is significantly more efficient for multiple append operations on the same item, as it performs a single pd.concat and a single write operation per item.

When to use which:

• Use transaction when you need immediate atomicity for a sequence of distinct operations.
• Use batch_transaction when you are performing many updates (especially appends) to the same items and want to minimize filesystem and Pandas overhead.

If you have a custom Collection class, you can inject validation support using add_validation_to_collection(collection_class). This monkey-patches the class to add:

• set_validator(validator: DataValidator): Attaches a validator to the collection instance.
• get_validator() -> Optional[DataValidator]: Retrieves the attached validator.
• write(item, data, ...): Validates data (if it's a DataFrame) before writing.
• append(item, data, ...): Validates data (if it's a DataFrame) before appending.

from pystore.validation import add_validation_to_collection, create_validator, ColumnExistsRule

class MyCollection:
    def write(self, item, data, metadata={}, **kwargs): 
        pass
    def append(self, item, data, **kwargs): 
        pass

# Inject validation logic
add_validation_to_collection(MyCollection)

collection = MyCollection()
validator = create_validator()
validator.add_rule(ColumnExistsRule(['price']))

# Attach the validator
collection.set_validator(validator)

# Now, calling write() will trigger validation
collection.write('item1', pd.DataFrame({'wrong_col': [1]})) # Raises ValidationError

To integrate schema evolution into your PyStore workflow, use enable_schema_evolution(collection_class, strategy). This injects methods into your collection class to manage schema metadata for individual items.

Once enabled, you can:

• item.enable_schema_evolution(item_name, strategy): Initialize evolution for a specific item.
• item.get_item_evolution(item_name): Retrieve the evolution manager for an item.
• item.set_item_evolution(item_name, evolution_obj): Save a specific evolution configuration to an item's metadata.

# Assuming 'MyCollection' is your PyStore collection class
from pystore.schema_evolution import enable_schema_evolution, EvolutionStrategy

enable_schema_evolution(MyCollection, strategy=EvolutionStrategy.COMPATIBLE)

# Now your collection instance can use these methods
my_collection.enable_schema_evolution('my_item_id', strategy=EvolutionStrategy.COMPATIBLE)

For scenarios involving many operations on the same items, use batch_transaction(collection). This is an optimized version that aggregates multiple append operations for the same item into a single pd.concat call before executing them within a standard atomic transaction. This reduces the overhead of multiple individual write/append calls.

Key Methods in BatchTransaction:

• write(item, data, **kwargs): Stages a write for item.
• append(item, data, **kwargs): Stages an append for item. Multiple calls to append for the same item will be combined.
• delete(item): Stages a deletion for item.

from pystore import batch_transaction

with batch_transaction(collection) as batch:
    batch.write('item1', df1)
    batch.write('item2', df2)
    batch.append('item3', df3)
    batch.append('item3', df4)  # These two appends will be combined into one
# All operations are committed together