from dataclasses import dataclass
from pathlib import Path
from typing import Any
import logging
from .file import FileStorage
from .base import SaveError, ValueMixin, CallMixin
from .destructuring import DestructuringMixin
from pyiron_snippets.import_alarm import ImportAlarm
[docs]
logger = logging.getLogger("fleche.storage.bagofholding_file")
with ImportAlarm(
"BagOfHoldingH5File requires 'bagofholding' to be installed. "
"Install it with `pip install fleche[bagofholding]`.",
raise_exception=True,
) as bagofholding_alarm:
from bagofholding import H5Bag
@dataclass(frozen=True)
[docs]
class BagOfHoldingH5FileBackend(FileStorage):
@bagofholding_alarm
[docs]
def __post_init__(self):
if hasattr(super(), "__post_init__"):
super().__post_init__()
[docs]
def _to_file(self, value: Any, path: Path) -> None:
try:
H5Bag.save(value, path)
except (ValueError, TypeError): # h5py choked on something, pass it along
raise SaveError(value) from None
[docs]
def _from_file(self, path: Path) -> Any:
try:
return H5Bag(path).load()
except FileNotFoundError:
raise KeyError(path) from None
except OSError as e:
logger.error(f"Corrupt file present in cache at path {path}: {e}")
raise KeyError(path) from e
@dataclass(frozen=True)
[docs]
class ValueBagOfHoldingH5File(ValueMixin, DestructuringMixin, BagOfHoldingH5FileBackend): ...
@dataclass(frozen=True)
[docs]
class CallBagOfHoldingH5File(CallMixin, BagOfHoldingH5FileBackend): ...