Source code for

#  Author:   Niels Nuyttens  <>
#  License: Apache Software License 2.0
import abc
import logging
from copy import deepcopy
from pathlib import Path
from typing import Any, Dict, Optional

import fsspec

from nannyml._typing import Result
from import Writer, _get_protocol_and_path

[docs]class FileWriter(Writer, abc.ABC): """An abstract Writer implementation that writes results out to a filesystem (either local or remote / cloud).""" _logger = logging.getLogger(__name__) def __init__( self, path: str, write_args: Optional[Dict[str, Any]] = None, credentials: Optional[Dict[str, Any]] = None, fs_args: Optional[Dict[str, Any]] = None, ): """ Creates a new ``FileWriter`` Parameters ---------- path : str The path to read data from. Can be a regular file path or contain a protocol. write_args : Dict[str, Any], default=None Specific arguments passed along the method performing the actual writing. credentials : Dict[str, Any], default=None Used to provide credential information following specific ``fsspec`` implementations. fs_args : default=None Specific arguments passed along to the ``fsspec`` filesystem initializer. """ super().__init__() _fs_args = deepcopy(fs_args) or {} _credentials = deepcopy(credentials) or {} protocol, path = _get_protocol_and_path(path) if protocol == "file": _fs_args.setdefault("auto_mkdir", True) self._protocol = protocol self.filepath = path self._storage_options = {**_credentials, **_fs_args} self._fs = fsspec.filesystem(self._protocol, **self._storage_options) self._write_args = write_args or {} # type: Dict[str, Any] def _write(self, result: Result, **kwargs): raise NotImplementedError( f"'{self.__class__.__name__}' is a subclass of Writer and it must implement the _write method" )
def _write_bytes_to_filesystem(bytez, save_path: Path, fs: fsspec.spec.AbstractFileSystem): with, mode="wb") as fs_file: fs_file.write(bytez)