lakehouse_engine.utils.storage.local_fs_storage

Module to represent a local file storage system.

 1"""Module to represent a local file storage system."""
 2
 3import os
 4from typing import TextIO
 5from urllib.parse import ParseResult
 6
 7from lakehouse_engine.utils.logging_handler import LoggingHandler
 8from lakehouse_engine.utils.storage.file_storage import FileStorage
 9
10
11class LocalFSStorage(FileStorage):
12    """Class to represent a local file storage system."""
13
14    _LOGGER = LoggingHandler(__name__).get_logger()
15
16    @classmethod
17    def get_file_payload(cls, url: ParseResult) -> TextIO:
18        """Get the payload of a file.
19
20        Args:
21            url: url of the file.
22
23        Returns:
24            file payload/content.
25        """
26        cls._LOGGER.info(f"Reading from file: {url.scheme}:{url.netloc}/{url.path}")
27        return open(f"{url.netloc}/{url.path}", "r")
28
29    @classmethod
30    def write_payload_to_file(cls, url: ParseResult, content: str) -> None:
31        """Write payload into a file.
32
33        Args:
34            url: url of the file.
35            content: content to write into the file.
36        """
37        cls._LOGGER.info(f"Writing into file: {url.scheme}:{url.netloc}/{url.path}")
38        os.makedirs(os.path.dirname(f"{url.netloc}/{url.path}"), exist_ok=True)
39        with open(f"{url.netloc}/{url.path}", "w") as file:
40            file.write(content)
class LocalFSStorage(lakehouse_engine.utils.storage.file_storage.FileStorage): 
12class LocalFSStorage(FileStorage):
13    """Class to represent a local file storage system."""
14
15    _LOGGER = LoggingHandler(__name__).get_logger()
16
17    @classmethod
18    def get_file_payload(cls, url: ParseResult) -> TextIO:
19        """Get the payload of a file.
20
21        Args:
22            url: url of the file.
23
24        Returns:
25            file payload/content.
26        """
27        cls._LOGGER.info(f"Reading from file: {url.scheme}:{url.netloc}/{url.path}")
28        return open(f"{url.netloc}/{url.path}", "r")
29
30    @classmethod
31    def write_payload_to_file(cls, url: ParseResult, content: str) -> None:
32        """Write payload into a file.
33
34        Args:
35            url: url of the file.
36            content: content to write into the file.
37        """
38        cls._LOGGER.info(f"Writing into file: {url.scheme}:{url.netloc}/{url.path}")
39        os.makedirs(os.path.dirname(f"{url.netloc}/{url.path}"), exist_ok=True)
40        with open(f"{url.netloc}/{url.path}", "w") as file:
41            file.write(content)

Class to represent a local file storage system.

@classmethod
def get_file_payload(cls, url: urllib.parse.ParseResult) -> <class 'TextIO'>: 
17    @classmethod
18    def get_file_payload(cls, url: ParseResult) -> TextIO:
19        """Get the payload of a file.
20
21        Args:
22            url: url of the file.
23
24        Returns:
25            file payload/content.
26        """
27        cls._LOGGER.info(f"Reading from file: {url.scheme}:{url.netloc}/{url.path}")
28        return open(f"{url.netloc}/{url.path}", "r")

Get the payload of a file.

Arguments:
  • url: url of the file.
Returns:

file payload/content.

@classmethod
def write_payload_to_file(cls, url: urllib.parse.ParseResult, content: str) -> None: 
30    @classmethod
31    def write_payload_to_file(cls, url: ParseResult, content: str) -> None:
32        """Write payload into a file.
33
34        Args:
35            url: url of the file.
36            content: content to write into the file.
37        """
38        cls._LOGGER.info(f"Writing into file: {url.scheme}:{url.netloc}/{url.path}")
39        os.makedirs(os.path.dirname(f"{url.netloc}/{url.path}"), exist_ok=True)
40        with open(f"{url.netloc}/{url.path}", "w") as file:
41            file.write(content)

Write payload into a file.

Arguments:
  • url: url of the file.
  • content: content to write into the file.