chore: update redshift functions (#235)

* chore: update redshift functions

* feat: add datasources in get config

* feat: add get resource url api using cloud storage manager

* chore: rename all cloud_manager to cloud_storage_manager

* feat: add read api for cloud storage files

* fix: fix storage read

Author: vishnurk6247

wavefront/server/apps/floware/floware/controllers/config_controller.py

@@ -8,6 +8,9 @@
 from fastapi.params import Depends
 from dependency_injector.wiring import inject
 from dependency_injector.wiring import Provide
+from db_repo_module.models.datasource import Datasource
+from db_repo_module.repositories.sql_alchemy_repository import SQLAlchemyRepository
+from plugins_module.plugins_container import PluginsContainer
 from floware.services.config_service import ConfigService
 from user_management_module.utils.user_utils import get_current_user, check_is_admin
 from fastapi import HTTPException
@@ -74,6 +77,10 @@ async def get_config(
         ConfigService,
         Depends(Provide[ApplicationContainer.config_service]),
     ],
+    datasource_repository: Annotated[
+        SQLAlchemyRepository[Datasource],
+        Depends(Provide[PluginsContainer.datasource_repository]),
+    ],
     response_formatter: Annotated[
         ResponseFormatter,
         Depends(Provide[CommonContainer.response_formatter]),
@@ -84,16 +91,16 @@ async def get_config(
     such as logo, table to query, etc.
     """
     url, app_config = await config_service.get_app_config()
-    if not url:
-        return JSONResponse(
-            status_code=status.HTTP_200_OK,
-            content=response_formatter.buildSuccessResponse(
-                {'message': 'No config found'}
-            ),
-        )
+    datasources = await datasource_repository.find()
+    datasource_ids = [str(datasource.id) for datasource in datasources]
+
     return JSONResponse(
         status_code=status.HTTP_200_OK,
         content=response_formatter.buildSuccessResponse(
-            {'app_icon': url, 'app_config': app_config}
+            {
+                'app_icon': url,
+                'app_config': app_config,
+                'datasources': datasource_ids,
+            }
         ),
     )

wavefront/server/apps/floware/floware/di/application_container.py

@@ -2,7 +2,6 @@
 from dependency_injector import providers
 
 from floware.services.notification_service import NotificationService
-from flo_cloud.cloud_storage import CloudStorageManager
 from floware.services.config_service import ConfigService
 
 
@@ -22,20 +21,16 @@ class ApplicationContainer(containers.DeclarativeContainer):
     notification_repository = providers.Dependency()
     notification_user_repository = providers.Dependency()
     config_repository = providers.Dependency()
+    cloud_storage_manager = providers.Dependency()
 
     # services
     notification_service = providers.Singleton(
         NotificationService, notification_repository, notification_user_repository
     )
 
-    cloud_manager = providers.Singleton(
-        CloudStorageManager,
-        provider=config.cloud_config.cloud_provider,
-    )
-
     config_service = providers.Singleton(
         ConfigService,
         config_repository=config_repository,
-        cloud_manager=cloud_manager,
+        cloud_storage_manager=cloud_storage_manager,
         config=config,
     )

wavefront/server/apps/floware/floware/server.py

@@ -93,6 +93,7 @@
 from plugins_module.controllers.message_processor_controller import (
     message_processor_router,
 )
+from plugins_module.controllers.cloud_storage_controller import cloud_storage_router
 
 # API Services Module
 from api_services_module.api_services_container import create_api_services_container
@@ -119,6 +120,7 @@
 
 application_container = ApplicationContainer(
     db_client=db_repo_container.db_client,
+    cloud_storage_manager=common_container.cloud_storage_manager,
     email_repository=db_repo_container.email_repository,
     oauth_credential_repository=db_repo_container.oauth_credential_repository,
     user_repository=db_repo_container.user_repository,
@@ -138,7 +140,7 @@
 
 plugins_container = PluginsContainer(
     db_client=db_repo_container.db_client,
-    cloud_manager=common_container.cloud_storage_manager,
+    cloud_storage_manager=common_container.cloud_storage_manager,
     dynamic_query_repository=db_repo_container.dynamic_query_repository,
     cache_manager=db_repo_container.cache_manager,
 )
@@ -162,7 +164,7 @@
     knowledge_base_inference_repository=db_repo_container.knowledge_base_inference_repository,
     message_processor_repository=plugins_container.message_processor_repository,
     api_services_manager=api_services_container.api_service_manager,
-    cloud_manager=common_container.cloud_storage_manager,
+    cloud_storage_manager=common_container.cloud_storage_manager,
     message_processor_bucket_name=bucket_name,
 )
 
@@ -382,6 +384,7 @@ async def metrics(request: Request):
 app.include_router(voice_agent_router, prefix='/floware')
 app.include_router(tool_router, prefix='/floware')
 app.include_router(message_processor_router, prefix='/floware')
+app.include_router(cloud_storage_router, prefix='/floware')
 
 
 @app.exception_handler(Exception)
@@ -495,6 +498,7 @@ async def global_exception_handler(request: Request, exc: Exception):
     packages=[
         'plugins_module.controllers',
         'plugins_module.services',
+        'floware.controllers',
         'user_management_module.controllers',
         'user_management_module.authorization',
         'tools_module.datasources',

wavefront/server/apps/floware/floware/services/config_service.py

@@ -10,11 +10,11 @@ class ConfigService:
     def __init__(
         self,
         config_repository: SQLAlchemyRepository[Config],
-        cloud_manager: CloudStorageManager,
+        cloud_storage_manager: CloudStorageManager,
         config: dict[str, Any],
     ) -> None:
         self.config_repository = config_repository
-        self.cloud_manager = cloud_manager
+        self.cloud_storage_manager = cloud_storage_manager
         self.config = config
 
     def _get_floware_credentials(self) -> dict[str, Any]:
@@ -46,7 +46,7 @@ async def store_app_config(
 
         file_content = await file.read() if file else None
         if file_content:
-            self.cloud_manager.save_small_file(
+            self.cloud_storage_manager.save_small_file(
                 file_content,
                 config_credentials['asset_storage_bucket'],
                 config_credentials['config_file_name'],
@@ -74,7 +74,7 @@ async def get_app_config(self):
         config_path = config_record[0].value.get('app_icon')
         config_credentials = self._get_floware_credentials()
         # Generate new presigned URL
-        url = self.cloud_manager.generate_presigned_url(
+        url = self.cloud_storage_manager.generate_presigned_url(
             config_credentials['asset_storage_bucket'],
             config_path,
             'get',

wavefront/server/apps/floware/tests/conftest.py

@@ -153,7 +153,7 @@ def setup_containers(test_engine, test_session, test_user_id, test_session_id):
 
 
 @pytest.fixture
-def mock_cloud_manager():
+def mock_cloud_storage_manager():
     """Mock CloudStorageManager for testing"""
     mock_cloud = Mock(spec=CloudStorageManager)
     mock_cloud.save_small_file = Mock()
@@ -203,13 +203,13 @@ async def mock_get_app_config():
 
 @pytest.fixture
 def setup_application_container(
-    mock_cloud_manager, mock_config_repository, mock_config, mock_config_service
+    mock_cloud_storage_manager, mock_config_repository, mock_config, mock_config_service
 ):
     """Setup ApplicationContainer with mocked dependencies"""
     app_container = ApplicationContainer()
 
     # Override the dependencies
-    app_container.cloud_manager.override(mock_cloud_manager)
+    app_container.cloud_storage_manager.override(mock_cloud_storage_manager)
     app_container.config_repository.override(mock_config_repository)
     app_container.config.override(mock_config)
     app_container.config_service.override(mock_config_service)

wavefront/server/background_jobs/workflow_job/workflow_job/main.py

@@ -33,7 +33,7 @@
 
 plugins_container = PluginsContainer(
     db_client=db_repo_container.db_client,
-    cloud_manager=common_container.cloud_storage_manager,
+    cloud_storage_manager=common_container.cloud_storage_manager,
     dynamic_query_repository=db_repo_container.dynamic_query_repository,
     cache_manager=db_repo_container.cache_manager,
 )
@@ -51,7 +51,7 @@
     knowledge_base_inference_repository=db_repo_container.knowledge_base_inference_repository,
     message_processor_repository=plugins_container.message_processor_repository,
     api_services_manager=api_services_container.api_service_manager,
-    cloud_manager=common_container.cloud_storage_manager,
+    cloud_storage_manager=common_container.cloud_storage_manager,
     message_processor_bucket_name=bucket_name,
 )
 

wavefront/server/modules/plugins_module/plugins_module/controllers/__init__.py

@@ -1,4 +1,5 @@
 from .datasource_controller import datasource_router
 from .authenticator_controller import authenticator_router
+from .cloud_storage_controller import cloud_storage_router
 
-__all__ = ['datasource_router', 'authenticator_router']
+__all__ = ['datasource_router', 'authenticator_router', 'cloud_storage_router']

wavefront/server/modules/plugins_module/plugins_module/controllers/cloud_storage_controller.py

@@ -0,0 +1,102 @@
+import json
+from enum import Enum
+from typing import Optional
+
+from dependency_injector.wiring import inject, Provide
+from fastapi import Depends, Query, status
+from fastapi.responses import JSONResponse
+from fastapi.routing import APIRouter
+
+from common_module.common_container import CommonContainer
+from common_module.response_formatter import ResponseFormatter
+from flo_cloud.cloud_storage import CloudStorageManager
+from plugins_module.plugins_container import PluginsContainer
+
+
+cloud_storage_router = APIRouter()
+
+
+class StorageFileType(str, Enum):
+    json = 'json'
+
+
+@cloud_storage_router.get('/v1/storage/signed-url')
+@inject
+async def get_resource_presigned_url(
+    resource_url: str = Query(..., description='The cloud storage URL of the resource'),
+    expires_in: int = Query(
+        300, description='Expiry time in seconds for the presigned URL'
+    ),
+    response_formatter: ResponseFormatter = Depends(
+        Provide[CommonContainer.response_formatter]
+    ),
+    cloud_storage_manager: CloudStorageManager = Depends(
+        Provide[PluginsContainer.cloud_storage_manager]
+    ),
+):
+    try:
+        bucket_name, key = cloud_storage_manager.get_bucket_key(resource_url)
+        presigned_url = cloud_storage_manager.generate_presigned_url(
+            bucket_name=bucket_name,
+            key=key,
+            type='GET',
+            expiresIn=expires_in,
+        )
+        return JSONResponse(
+            status_code=status.HTTP_200_OK,
+            content=response_formatter.buildSuccessResponse(
+                {'resource_url': presigned_url}
+            ),
+        )
+    except Exception as e:
+        return JSONResponse(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            content=response_formatter.buildErrorResponse(str(e)),
+        )
+
+
+@cloud_storage_router.get('/v1/storage/read')
+@inject
+async def read_storage_file(
+    resource_url: str = Query(..., description='The cloud storage URL of the resource'),
+    type: StorageFileType = Query(StorageFileType.json, description='File type'),
+    projection: Optional[str] = Query(
+        None,
+        description='Comma-separated list of top-level fields to return from the parsed data',
+    ),
+    response_formatter: ResponseFormatter = Depends(
+        Provide[CommonContainer.response_formatter]
+    ),
+    cloud_storage_manager: CloudStorageManager = Depends(
+        Provide[PluginsContainer.cloud_storage_manager]
+    ),
+):
+    try:
+        bucket_name, key = cloud_storage_manager.get_bucket_key(resource_url)
+        file_content = cloud_storage_manager.read_file(bucket_name, key)
+
+        if type == StorageFileType.json:
+            data = json.loads(file_content)
+            if projection:
+                fields = {f.strip() for f in projection.split(',') if f.strip()}
+                if isinstance(data, dict):
+                    data = {k: v for k, v in data.items() if k in fields}
+                elif isinstance(data, list):
+                    data = [
+                        {k: v for k, v in item.items() if k in fields}
+                        if isinstance(item, dict)
+                        else item
+                        for item in data
+                    ]
+
+        return JSONResponse(
+            status_code=status.HTTP_200_OK,
+            content=response_formatter.buildSuccessResponse({'data': data}),
+        )
+    except (json.JSONDecodeError, ValueError) as e:
+        return JSONResponse(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            content=response_formatter.buildErrorResponse(
+                f'Failed to parse file as {type.value}: {str(e)}'
+            ),
+        )

wavefront/server/modules/plugins_module/plugins_module/plugins_container.py

@@ -39,21 +39,21 @@ class PluginsContainer(containers.DeclarativeContainer):
     # dynamic query service
     cloud_provider = config.cloud_config.cloud_provider
 
-    cloud_manager = providers.Singleton(
+    cloud_storage_manager = providers.Singleton(
         CloudStorageManager, provider=config.cloud_config.cloud_provider
     )
 
     dynamic_query_service = providers.Singleton(
         DynamicQueryService,
-        cloud_manager=cloud_manager,
+        cloud_storage_manager=cloud_storage_manager,
         dynamic_query_repo=dynamic_query_repository,
         bucket_name=config.floware.asset_storage_bucket,
     )
 
     message_processor_service = providers.Singleton(
         MessageProcessorService,
-        cloud_manager=cloud_manager,
+        cloud_storage_manager=cloud_storage_manager,
         message_processor_repository=message_processor_repository,
-        bucket_name=config.floware.asset_storage_bucket,
         hermes_url=config.hermes.url,
+        bucket_name=config.floware.asset_storage_bucket,
     )

wavefront/server/modules/plugins_module/plugins_module/services/dynamic_query_service.py

@@ -3,16 +3,17 @@
 from common_module.log.logger import logger
 from flo_cloud.cloud_storage import CloudStorageManager
 from db_repo_module.models.dynamic_query_yaml import DynamicQueryYaml
+from typing import Optional
 
 
 class DynamicQueryService:
     def __init__(
         self,
-        cloud_manager: CloudStorageManager,
+        cloud_storage_manager: CloudStorageManager,
         dynamic_query_repo: SQLAlchemyRepository[DynamicQueryYaml],
-        bucket_name: str = None,
+        bucket_name: Optional[str] = None,
     ):
-        self.cloud_manager = cloud_manager
+        self.cloud_storage_manager = cloud_storage_manager
         self.dynamic_query_repo = dynamic_query_repo
         self.bucket_name = bucket_name
         self.prefix = 'dynamic_query/v1'
@@ -40,7 +41,7 @@ async def store_yaml_to_bucket(self, yaml_content: dict, datasource_id: str):
             file_content = yaml_string.encode('utf-8')
 
             # storing to s3bucket
-            self.cloud_manager.save_small_file(
+            self.cloud_storage_manager.save_small_file(
                 file_content=file_content, bucket_name=self.bucket_name, key=file_key
             )
 
@@ -70,7 +71,7 @@ async def retrive_dynamic_query_yaml(self, page_number, page_size):
         Returns:
             dict: Contains yamls list, pagination info, and total count
         """
-        files_keys, has_more = self.cloud_manager.list_files(
+        files_keys, has_more = self.cloud_storage_manager.list_files(
             self.bucket_name, self.prefix, page_size, page_number
         )
         yamls = []
@@ -100,7 +101,7 @@ async def get_dynamic_yaml_query(self, query_id: str):
             dict: Contains yaml query and their parameters
         """
         file_key = f'{self.prefix}/{query_id}.yaml'
-        file_content = self.cloud_manager.read_file(self.bucket_name, file_key)
+        file_content = self.cloud_storage_manager.read_file(self.bucket_name, file_key)
         yaml_query = yaml.safe_load(file_content.decode('utf-8'))
         if not yaml_query:
             raise ValueError('YAML file is invalid')
@@ -141,7 +142,7 @@ async def delete_dynamic_query(self, datasource_id: str, query_id: str):
                 raise ValueError(f'Query {query_id} not found')
 
             # deleting the file from the cloud storage
-            self.cloud_manager.delete_file(self.bucket_name, query.file_path)
+            self.cloud_storage_manager.delete_file(self.bucket_name, query.file_path)
             # deleting the record from the database
             await self.dynamic_query_repo.delete_all(name=query_id)