Filesystem Provider

Implementation of the data provider layer for the local filesystem.

This implementation supports both the old uncompressed and new compressed (Zstandard) formats.

This implementation stores the data in files, in directories named after the job id (8 digits zero-padded).

Classes:

Name	Description
- FSDataEntry	Filesystem data entry implementation.
- FSDictEntry	Filesystem dictionary entry implementation.
- FSJobEntry	Filesystem job entry implementation.
- FSDataProvider	Filesystem data provider implementation.
- FSDictProvider	Filesystem dictionary provider implementation.

FSDataEntry

Bases: DataEntry

Filesystem data entry implementation.

This implementation supports both the old uncompressed and new compressed (Zstandard) formats.

Source code in src/lhcbdirac_log/providers/filesystem/accessors.py

class FSDataEntry(DataEntry):
    """Filesystem data entry implementation.

    This implementation supports both the old uncompressed and new compressed (Zstandard) formats.
    """

    __slots__ = ("_path",)

    def __init__(self, path: PathLike[str] | str, job: int, *, compressed: bool, readonly: bool) -> None:
        """[Internal] Initialize the data entry.

        Args:
            path: the data file path
            job: the job id
            compressed: indicate whether the underlying data is compressed or not (in Zstandard)
            readonly: indicate weather the data is read-only or not
        """
        self._path = Path(path)
        super().__init__(self._path.name, job, compressed=compressed, readonly=readonly)

    @override
    def _reader(self) -> BinaryIO:
        try:
            return self._path.open("rb")
        except FileNotFoundError as err:  # pragma: no cover
            raise DataNotExistsError(self._path) from err

    @override
    def _writer(self) -> BinaryIO:
        return self._path.open("wb")

    @override
    def _size(self) -> int | None:
        try:
            return self._path.stat().st_size
        except FileNotFoundError:
            return None

    @override
    def _delete(self) -> None:
        try:
            self._path.unlink()
        except FileNotFoundError as err:
            raise DataNotExistsError(self._path) from err

__init__(path, job, *, compressed, readonly)

[Internal] Initialize the data entry.

Parameters:

Name	Type	Description	Default
path	PathLike[str] | str	the data file path	required
job	int	the job id	required
compressed	bool	indicate whether the underlying data is compressed or not (in Zstandard)	required
readonly	bool	indicate weather the data is read-only or not	required

Source code in src/lhcbdirac_log/providers/filesystem/accessors.py

def __init__(self, path: PathLike[str] | str, job: int, *, compressed: bool, readonly: bool) -> None:
    """[Internal] Initialize the data entry.

    Args:
        path: the data file path
        job: the job id
        compressed: indicate whether the underlying data is compressed or not (in Zstandard)
        readonly: indicate weather the data is read-only or not
    """
    self._path = Path(path)
    super().__init__(self._path.name, job, compressed=compressed, readonly=readonly)

FSDataProvider

Bases: DataProvider[FSJobEntry]

Filesystem data provider implementation.

This implementation supports both the old uncompressed and new compressed (Zstandard) formats.

Source code in src/lhcbdirac_log/providers/filesystem/providers.py

class FSDataProvider(DataProvider[FSJobEntry]):
    """Filesystem data provider implementation.

    This implementation supports both the old uncompressed and new compressed (Zstandard) formats.
    """

    __slots__ = ("_folder",)

    @property
    def folder(self) -> Path:
        """Get the folder path.

        Returns:
            the folder path
        """
        return self._folder

    def __init__(self, folder: PathLike[str] | str, dict_provider: DictProvider | None = None, *, readonly: bool = False) -> None:
        """Initialize the data provider.

        Args:
            folder: the folder path
            dict_provider: the dict provider associated to the data (default is None), specifying this implies that the provided data are compressed
            readonly: indicate weather the provider is read-only or not (default: False)
        """
        self._folder = Path(folder)
        self._folder.mkdir(parents=True, exist_ok=True)
        super().__init__(dict_provider, readonly=readonly)

    @override
    def _get(self, job: int, create: bool = False) -> FSJobEntry:
        return FSJobEntry(self._folder, job, compressed=self.compressed, readonly=self._readonly, create=create)

    @override
    def _create(self, job: int, exists_ok: bool = False) -> FSJobEntry:
        return FSJobEntry(self._folder, job, compressed=self.compressed, readonly=self._readonly, exists_ok=exists_ok)

    @override
    def jobs(self) -> Generator[int, None, None]:
        yield from map(int, (p.name for p in self._folder.iterdir() if p.is_dir() and p.name.isdecimal()))

    @override
    def _delete(self, job: int, *, force: bool = False) -> None:
        j = self.get(job)

        try:
            if force:
                rmtree(j.folder)
            else:
                j.folder.rmdir()
        except FileNotFoundError as err:  # pragma: no cover
            raise JobNotExistsError(job) from err
        except OSError as err:
            raise DataExistsError(job) from err

folder: Path property

Get the folder path.

Returns:

Type	Description
Path	the folder path

__init__(folder, dict_provider=None, *, readonly=False)

Initialize the data provider.

Parameters:

Name	Type	Description	Default
folder	PathLike[str] | str	the folder path	required
dict_provider	DictProvider | None	the dict provider associated to the data (default is None), specifying this implies that the provided data are compressed	None
readonly	bool	indicate weather the provider is read-only or not (default: False)	False

Source code in src/lhcbdirac_log/providers/filesystem/providers.py

def __init__(self, folder: PathLike[str] | str, dict_provider: DictProvider | None = None, *, readonly: bool = False) -> None:
    """Initialize the data provider.

    Args:
        folder: the folder path
        dict_provider: the dict provider associated to the data (default is None), specifying this implies that the provided data are compressed
        readonly: indicate weather the provider is read-only or not (default: False)
    """
    self._folder = Path(folder)
    self._folder.mkdir(parents=True, exist_ok=True)
    super().__init__(dict_provider, readonly=readonly)

FSDictEntry

Bases: DictEntry

Filesystem dictionary entry implementation.

Source code in src/lhcbdirac_log/providers/filesystem/accessors.py

class FSDictEntry(DictEntry):
    """Filesystem dictionary entry implementation."""

    __slots__ = ("_path",)

    def __init__(self, path: PathLike[str] | str, config: Config, data: bytes | None = None, zstd_id: int | None = None) -> None:
        """[Internal] Initialize the dictionary entry.

        Args:
            path: the dictionary file path
            config: the configuration to use for precomputing the dictionary
            data: the dictionary data (create a new dict if not None)
            zstd_id: the zstd dictionary id (None for unknown)
        """
        self._path = Path(path)
        super().__init__(self._path.name, config, data, zstd_id)

    @property
    @override
    def exists(self) -> bool:
        return self._path.is_file()

    @property
    @override
    def size(self) -> int:
        try:
            return self._path.stat().st_size
        except FileNotFoundError:
            return 0

    @override
    def _load_data(self) -> bytes:
        try:
            with self._path.open("rb") as file:
                return file.read()
        except FileNotFoundError as err:
            raise DictNotExistsError(self._path) from err

    @override
    def _save(self) -> None:
        with self._path.open("wb") as file:
            file.write(self._data)

__init__(path, config, data=None, zstd_id=None)

[Internal] Initialize the dictionary entry.

Parameters:

Name	Type	Description	Default
path	PathLike[str] | str	the dictionary file path	required
config	Config	the configuration to use for precomputing the dictionary	required
data	bytes | None	the dictionary data (create a new dict if not None)	None
zstd_id	int | None	the zstd dictionary id (None for unknown)	None

Source code in src/lhcbdirac_log/providers/filesystem/accessors.py

def __init__(self, path: PathLike[str] | str, config: Config, data: bytes | None = None, zstd_id: int | None = None) -> None:
    """[Internal] Initialize the dictionary entry.

    Args:
        path: the dictionary file path
        config: the configuration to use for precomputing the dictionary
        data: the dictionary data (create a new dict if not None)
        zstd_id: the zstd dictionary id (None for unknown)
    """
    self._path = Path(path)
    super().__init__(self._path.name, config, data, zstd_id)

FSDictProvider

Bases: DictProvider[FSDictEntry]

Filesystem dictionary provider implementation.

Source code in src/lhcbdirac_log/providers/filesystem/providers.py

class FSDictProvider(DictProvider[FSDictEntry]):
    """Filesystem dictionary provider implementation."""

    __slots__ = ("_folder",)

    @property
    def folder(self) -> Path:
        """Get the folder path.

        Returns:
            the folder path
        """
        return self._folder

    def __init__(self, folder: PathLike[str] | str, config: Config = DEFAULT_CONFIG, *, readonly: bool = False) -> None:
        """Initialize the dictionary provider.

        Args:
            folder: the folder path, where to store the dictionaries
            config: the configuration to use for precomputing the dictionaries (default: DEFAULT_CONFIG)
            readonly: indicate weather the provider is read-only or not (default: False)
        """
        self._folder = Path(folder)
        self._folder.mkdir(parents=True, exist_ok=True)
        super().__init__(config, readonly=readonly)

    @override
    def _load(self, name: str) -> FSDictEntry:
        p = self._folder / name

        if not p.is_file():
            raise DictNotExistsError(p)

        return FSDictEntry(p, self._config)

    @override
    def _add(self, name: str, data: bytes, zstd_id: int) -> FSDictEntry:
        p = self._folder / name

        if p.is_file():
            raise DictExistsError(p)

        return FSDictEntry(p, self._config, data, zstd_id)

    @override
    def _iter_all(self) -> Generator[str, None, None]:
        return (f.name for f in self._folder.iterdir() if f.is_file())

    @override
    def _delete(self, name: str) -> None:
        p = self._folder / name

        try:
            p.unlink()
        except (FileNotFoundError, OSError) as err:
            raise DictNotExistsError(p) from err

    @override
    @property
    def size(self) -> int:
        return sum(f.stat().st_size for f in self._folder.iterdir() if f.is_file())

folder: Path property

Get the folder path.

Returns:

Type	Description
Path	the folder path

__init__(folder, config=DEFAULT_CONFIG, *, readonly=False)

Initialize the dictionary provider.

Parameters:

Name	Type	Description	Default
folder	PathLike[str] | str	the folder path, where to store the dictionaries	required
config	Config	the configuration to use for precomputing the dictionaries (default: DEFAULT_CONFIG)	DEFAULT_CONFIG
readonly	bool	indicate weather the provider is read-only or not (default: False)	False

Source code in src/lhcbdirac_log/providers/filesystem/providers.py

def __init__(self, folder: PathLike[str] | str, config: Config = DEFAULT_CONFIG, *, readonly: bool = False) -> None:
    """Initialize the dictionary provider.

    Args:
        folder: the folder path, where to store the dictionaries
        config: the configuration to use for precomputing the dictionaries (default: DEFAULT_CONFIG)
        readonly: indicate weather the provider is read-only or not (default: False)
    """
    self._folder = Path(folder)
    self._folder.mkdir(parents=True, exist_ok=True)
    super().__init__(config, readonly=readonly)

FSJobEntry

Bases: JobEntry[FSDataEntry]

Filesystem job entry implementation.

This implementation supports both the old uncompressed and new compressed (Zstandard) formats.

Source code in src/lhcbdirac_log/providers/filesystem/accessors.py

class FSJobEntry(JobEntry[FSDataEntry]):
    """Filesystem job entry implementation.

    This implementation supports both the old uncompressed and new compressed (Zstandard) formats.
    """

    __slots__ = ("_folder",)

    def __init__(self, folder: Path, job: int, *, compressed: bool, readonly: bool, create: bool = True, exists_ok: bool = True) -> None:
        """[Internal] Initialize the job entry.

        Args:
            folder: the production folder path (parent of the job folders)
            job: the job id
            compressed: indicate whether the underlying data is compressed or not (in Zstandard)
            readonly: indicate weather the job is read-only or not
            create: create the job folder if it doesn't exist, otherwise raise an error
            exists_ok: ignore the error if the job folder already exists

        Raises:
            JobExistsError: if the job folder already exists and exists_ok is False
            JobNotExistsError: if the job folder doesn't exist and create is False
        """
        self._folder = folder / f"{job:08}"

        if self._folder.is_dir():
            if not exists_ok:
                raise JobExistsError(job)
        elif create:
            self._folder.mkdir(parents=True, exist_ok=True)
        else:
            raise JobNotExistsError(job)

        super().__init__(job, compressed=compressed, readonly=readonly)

    @property
    def folder(self) -> Path:
        """Get the job folder path.

        Returns:
            the job folder path
        """
        return self._folder

    @override
    def _get(self, name: str, *, create: bool = False) -> FSDataEntry:
        p = self._folder / name

        if not create and not p.is_file():
            raise DataNotExistsError(p)

        return FSDataEntry(p, self._job, compressed=self._compressed, readonly=self._readonly)

    @override
    def _create(self, name: str, *, exists_ok: bool = False) -> FSDataEntry:
        p = self._folder / name

        if p.is_file() and not exists_ok:
            raise DataExistsError(p)

        return FSDataEntry(p, self._job, compressed=self._compressed, readonly=self._readonly)

    @override
    def files(self) -> Generator[str, None, None]:
        yield from (p.name for p in self._folder.iterdir() if p.is_file())

    @property
    @override
    def data_size(self) -> int:  # optimizes the default implementation
        return sum((self._folder / f).stat().st_size for f in self.files())

    @override
    def delete(self, name: str) -> None:  # optimizes the default implementation
        if self._readonly:
            msg = f"Job '{self._job}' is read-only"
            raise ReadOnlyError(msg)

        p = self._folder / name

        try:
            p.unlink()
        except FileNotFoundError as err:
            raise DataNotExistsError(p) from err

    @override
    def _update_info(self) -> None:  # pragma: no cover
        pass

folder: Path property

Get the job folder path.

Returns:

Type	Description
Path	the job folder path

__init__(folder, job, *, compressed, readonly, create=True, exists_ok=True)

[Internal] Initialize the job entry.

Parameters:

Name	Type	Description	Default
folder	Path	the production folder path (parent of the job folders)	required
job	int	the job id	required
compressed	bool	indicate whether the underlying data is compressed or not (in Zstandard)	required
readonly	bool	indicate weather the job is read-only or not	required
create	bool	create the job folder if it doesn't exist, otherwise raise an error	True
exists_ok	bool	ignore the error if the job folder already exists	True

Raises:

Type	Description
JobExistsError	if the job folder already exists and exists_ok is False
JobNotExistsError	if the job folder doesn't exist and create is False

Source code in src/lhcbdirac_log/providers/filesystem/accessors.py

def __init__(self, folder: Path, job: int, *, compressed: bool, readonly: bool, create: bool = True, exists_ok: bool = True) -> None:
    """[Internal] Initialize the job entry.

    Args:
        folder: the production folder path (parent of the job folders)
        job: the job id
        compressed: indicate whether the underlying data is compressed or not (in Zstandard)
        readonly: indicate weather the job is read-only or not
        create: create the job folder if it doesn't exist, otherwise raise an error
        exists_ok: ignore the error if the job folder already exists

    Raises:
        JobExistsError: if the job folder already exists and exists_ok is False
        JobNotExistsError: if the job folder doesn't exist and create is False
    """
    self._folder = folder / f"{job:08}"

    if self._folder.is_dir():
        if not exists_ok:
            raise JobExistsError(job)
    elif create:
        self._folder.mkdir(parents=True, exist_ok=True)
    else:
        raise JobNotExistsError(job)

    super().__init__(job, compressed=compressed, readonly=readonly)