Pluggability of Cloud Storage

In this guide, we will learn how to plugin a new cloud storage wrapper by extending the BaseStorageClass interface.

You can seamlessly leverage it within existing Sakhi API workflows with minimal code changes.

Integration

Cloud storage is used for storing the response generated from LLM which is then converted to audio format using a translation util. Link of the audio response files will be used by clients connecting to sakhi-api-service for playing the response.

BaseStorageClass

This BaseStorageClass class has below methods which are to be implemented mandatorily.

Method/Property	Description	Required/Optional
__init__	Used to instantiate the storage account client	Required
upload_to_storage	Takes name of the file to be uploaded	Required
generate_public_url	Takes name of the file for which the public url is to be generated	Required

Implementation

Let's say we create a YourStorageClass class inheriting from BaseStorageClass. The class needs to implement the following methods __init__, upload_to_storage and generate_public_url.

Add the necessary environment variables to .env file. BUCKET_TYPE is to be defined mandatorily.

BUCKET_TYPE=your_cloud

Now go to the 'storage' folder and create your plugin: your_cloud.py

your_cloud.py

from storage.base import BaseStorageClass


class YourBucketClass(BaseStorageClass):
    def __init__(self):
        your_client = # instantiate your client
        super().__init__(your_client)

    def upload_to_storage(self, file_name: str, object_name: Optional[str] = None) -> bool:
        # code to upload file to your cloud storage
        return True

    def generate_public_url(self, object_name: str) -> Union[tuple[str, None], tuple[None, str]]:
        try:
            public_url = # you code to fetch public url of the uploaded file
            return public_url, None
        except Exception as e:
            logger.error(f"Exception Preparing public URL: {e}", exc_info=True)
            return None, "Error while generating public URL"

Go to the 'storage' folder and update __init__.py with the module lookup entry for "YourBucketClass" and also for TYPE_CHECKING.

if TYPE_CHECKING:
    ...
    from storage.your_cloud import (
        YourBucketClass
    )

_module_lookup = {
    ...,
    "YourBucketClass": "storage.your_cloud"
}

Modify env_manager.py to import YourStorageClass and add a mapping for YourStorageClass in the self.indexes dictionary.

from storage import (
    ...,
    YourBucketClass
)

self.indexes = {
    "storage": {
        "class": {
            ...,
            "your_cloud": YourBucketClass
        },
        "env_key": "BUCKET_TYPE"
    }
}

This setup ensures that YourBucketClass can be instantiated based on specific environment variables, effectively integrating it into the environment management system. The self.indexes the dictionary now includes a mapping where "your_cloud" corresponds to the "YourBucketClass" class and uses "BUCKET_TYPE" as the environment key.