"""
Implements the abstract class :class:`FileSink` which defines the base class
for storage target containers.
"""
from __future__ import annotations
from scistag.filestag.protocols import AZURE_PROTOCOL_HEADER, \
ZIP_SOURCE_PROTOCOL
[docs]class FileStorageOptions:
"""
Advanced file storage parameters
Not yet defined.
"""
pass
[docs]class FileSink:
"""
A file sink is the abstract base class for a file storage container target,
eg.g. for storing files in a batch process such as converting a large set
of images and storing them in an output directory, archive or cloud storage.
"""
def __init__(self, target: str, **params):
"""
:param target: The storage target
:param params: Additional parameters
"""
self._target = target
"The sink's storage target"
self._closed = False
"Defines if the file sink was finalized (e.g. a zip archive closed)"
[docs] @staticmethod
def with_target(target: str, **params):
"""
Creates a file source with given target.
:param target: The target cloud storage or zip archive at which the data
shall be stored.
Supported types (as of now) are:
- "azure://DefaultEndpoints..." to store data in a
FileSinkAzureStorage
- "zip://" w/o a filename to create a memory zip
:param params: Further parameters to be passed on
:return: The FileSink instance
"""
if target == ZIP_SOURCE_PROTOCOL:
from .zip_file_sink import ZipFileSink
return ZipFileSink(target=target, **params)
if target.startswith(AZURE_PROTOCOL_HEADER):
from .azure.azure_storage_file_sink import AzureStorageFileSink
return AzureStorageFileSink(target=target, **params)
raise ValueError("Unsupported target type")
[docs] def store(self,
filename: str,
data: bytes,
overwrite: bool = True,
options: FileStorageOptions | None = None) -> bool:
"""
Stores a single file in the file sink
:param filename: The file's name
:param data: The data to be stored
:param overwrite: Defines if the file may be overwritten if it does
already exist.
:param options: Advanced storage and file options
:return: True on success
"""
return self._store_int(filename, data, overwrite=overwrite,
options=options)
[docs] def _store_int(self,
filename: str,
data: bytes,
overwrite: bool,
options: FileStorageOptions | None = None) -> bool:
"""
The internal storage function to be implemented for the different
kinds of target types.
:param filename: The name of the file to be stored
:param data: The data to be stored
:param options: Advanced storage and file options
:return: True on success
"""
raise NotImplementedError("Missing storage handler implementation")
[docs] def close(self):
"""
This function should be called when ever you finished adding all files
to the file sink to finalize it (e.g. a zip archive)
Alternatively you can create and fill the sink within a `with` block
such as
.. code-block: python:
with MyFileSink() as fs:
fs.add("myfile.txt", b"Hello world")
"""
if self._closed:
raise AssertionError("Tried to close FileSink twice")
self._closed = True
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.close()