Class FileStore

python-client/src/datapane/processors/file_store.py:160–206

class FileStore:
    # TODO - make this a CAS (index by object hash itself?)
    # NOTE - currently we pass dir_path via the FileStore, could move into the file themselves?
    def __init__(self, fw_klass: t.Type[FileEntry], assets_dir: t.Optional[Path] = None):
        super().__init__()
        self.fw_klass = fw_klass
        self.files: t.List[FileEntry] = []
        self.dir_path = assets_dir

    def __add__(self, other: FileStore) -> Self:
        # TODO - ensure factory is the same for both
        self.files.extend(other.files)
        return self

    @property
    def store_count(self) -> int:
        return len(self.files)

    @property
    def file_list(self) -> t.List[t.BinaryIO]:
        return [f.wrapped for f in self.files]

    def get_file(self, ext: str, mime: str) -> FileEntry:
        return self.fw_klass(ext, mime, self.dir_path)

    def add_file(self, fw: FileEntry) -> None:
        fw.freeze()
        self.files.append(fw)

    def load_file(self, path: Path) -> FileEntry:
        """load a file into the store (makes a copy)"""
        # TODO - ideally lazily-link a path to the store (rather than copy it in)
        ext = "".join(path.suffixes)
        dest_obj = self.fw_klass(ext=ext, dir_path=self.dir_path)
        with path.open("rb") as src_obj:
            copyfileobj(src_obj, dest_obj.file)
        self.add_file(dest_obj)
        return dest_obj

    def as_dict(self) -> dict:
        """Build a json structure suitable for embedding in a html file, json-rpc response, etc."""
        x: FileEntry  # noqa: F842
        return {x.hash: x.as_dict() for x in self.files}

    def get_entry(self, hash: str) -> t.Optional[FileEntry]:
        # TODO - change self.files to a dict[hash, FileEntry]?
        return next((f for f in self.files if f.hash == hash), None)