Local fs storage

Module to represent a local file storage system.

LocalFSStorage

Bases: FileStorage

Class to represent a local file storage system.

Source code in mkdocs/lakehouse_engine/packages/utils/storage/local_fs_storage.py

class LocalFSStorage(FileStorage):
    """Class to represent a local file storage system."""

    _LOGGER = LoggingHandler(__name__).get_logger()

    @classmethod
    def get_file_payload(cls, url: ParseResult) -> TextIO:
        """Get the payload of a file.

        Args:
            url: url of the file.

        Returns:
            file payload/content.
        """
        cls._LOGGER.info(f"Reading from file: {url.scheme}:{url.netloc}/{url.path}")
        return open(f"{url.netloc}/{url.path}", "r")

    @classmethod
    def write_payload_to_file(cls, url: ParseResult, content: str) -> None:
        """Write payload into a file.

        Args:
            url: url of the file.
            content: content to write into the file.
        """
        cls._LOGGER.info(f"Writing into file: {url.scheme}:{url.netloc}/{url.path}")
        os.makedirs(os.path.dirname(f"{url.netloc}/{url.path}"), exist_ok=True)
        with open(f"{url.netloc}/{url.path}", "w") as file:
            file.write(content)

get_file_payload(url) classmethod

Get the payload of a file.

Parameters:

Name	Type	Description	Default
url	ParseResult	url of the file.	required

Returns:

Type	Description
TextIO	file payload/content.

Source code in mkdocs/lakehouse_engine/packages/utils/storage/local_fs_storage.py

@classmethod
def get_file_payload(cls, url: ParseResult) -> TextIO:
    """Get the payload of a file.

    Args:
        url: url of the file.

    Returns:
        file payload/content.
    """
    cls._LOGGER.info(f"Reading from file: {url.scheme}:{url.netloc}/{url.path}")
    return open(f"{url.netloc}/{url.path}", "r")

write_payload_to_file(url, content) classmethod

Write payload into a file.

Parameters:

Name	Type	Description	Default
url	ParseResult	url of the file.	required
content	str	content to write into the file.	required

Source code in mkdocs/lakehouse_engine/packages/utils/storage/local_fs_storage.py

@classmethod
def write_payload_to_file(cls, url: ParseResult, content: str) -> None:
    """Write payload into a file.

    Args:
        url: url of the file.
        content: content to write into the file.
    """
    cls._LOGGER.info(f"Writing into file: {url.scheme}:{url.netloc}/{url.path}")
    os.makedirs(os.path.dirname(f"{url.netloc}/{url.path}"), exist_ok=True)
    with open(f"{url.netloc}/{url.path}", "w") as file:
        file.write(content)