S3 storage

Module to represent a s3 file storage system.

S3Storage

Bases: FileStorage

Class to represent a s3 file storage system.

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

class S3Storage(FileStorage):
    """Class to represent a s3 file storage system."""

    _LOGGER = LoggingHandler(__name__).get_logger()

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

        Args:
            url: url of the file.

        Returns:
            File payload/content.
        """
        s3 = boto3.resource("s3")
        obj = s3.Object(url.netloc, url.path.lstrip("/"))
        cls._LOGGER.info(
            f"Trying with s3_storage: "
            f"Reading from file: {url.scheme}://{url.netloc}{url.path}"
        )
        return obj.get()["Body"]

    @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.
        """
        s3 = boto3.resource("s3")
        obj = s3.Object(url.netloc, url.path.lstrip("/"))
        cls._LOGGER.info(
            f"Trying with s3_storage: "
            f"Writing into file: {url.scheme}://{url.netloc}{url.path}"
        )
        obj.put(Body=content)

get_file_payload(url) classmethod

Get the payload of a config file.

Parameters:

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

Returns:

Type	Description
Any	File payload/content.

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

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

    Args:
        url: url of the file.

    Returns:
        File payload/content.
    """
    s3 = boto3.resource("s3")
    obj = s3.Object(url.netloc, url.path.lstrip("/"))
    cls._LOGGER.info(
        f"Trying with s3_storage: "
        f"Reading from file: {url.scheme}://{url.netloc}{url.path}"
    )
    return obj.get()["Body"]

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/s3_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.
    """
    s3 = boto3.resource("s3")
    obj = s3.Object(url.netloc, url.path.lstrip("/"))
    cls._LOGGER.info(
        f"Trying with s3_storage: "
        f"Writing into file: {url.scheme}://{url.netloc}{url.path}"
    )
    obj.put(Body=content)