🐙 octavia-cli: add command to list existing sources, destinations and connections

octavia-cli/README.md

@@ -38,6 +38,7 @@ We welcome community contributions!
 
 | Date       | Milestone                           |
 |------------|-------------------------------------|
+| 2022-01-19 | Implement `octavia list workspace sources`, `octavia list workspace destinations`, `octavia list workspace connections`|
 | 2022-01-17 | Implement `octavia list connectors source` and `octavia list connectors destinations`|
 | 2022-01-17 | Generate an API Python client from our Open API spec |
 | 2021-12-22 | Bootstrapping the project's code base |

octavia-cli/octavia_cli/list/commands.py

@@ -6,7 +6,7 @@
 
 import click
 
-from .connectors_definitions import DestinationConnectorsDefinitions, SourceConnectorsDefinitions
+from .listings import Connections, DestinationConnectorsDefinitions, Destinations, SourceConnectorsDefinitions, Sources
 
 
 @click.group("list", help="List existing Airbyte resources.")
@@ -21,23 +21,56 @@ def connectors(ctx: click.Context):  # pragma: no cover
     pass
 
 
-@connectors.command(help="Latest information on supported sources.")
+@click.group("workspace", help="Latest information on workspace's sources and destinations.")
 @click.pass_context
-def sources(ctx: click.Context):
+def workspace(ctx: click.Context):  # pragma: no cover
+    pass
+
+
+@connectors.command(name="sources", help="Latest information on supported sources.")
+@click.pass_context
+def sources_connectors(ctx: click.Context):
     api_client = ctx.obj["API_CLIENT"]
     definitions = SourceConnectorsDefinitions(api_client)
     click.echo(definitions)
 
 
-@connectors.command(help="Latest information on supported destinations.")
+@connectors.command(name="destination", help="Latest information on supported destinations.")
 @click.pass_context
-def destinations(ctx: click.Context):
+def destinations_connectors(ctx: click.Context):
     api_client = ctx.obj["API_CLIENT"]
     definitions = DestinationConnectorsDefinitions(api_client)
     click.echo(definitions)
 
 
-AVAILABLE_COMMANDS: List[click.Command] = [connectors]
+@workspace.command(help="List existing sources in a workspace.")
+@click.pass_context
+def sources(ctx: click.Context):
+    api_client = ctx.obj["API_CLIENT"]
+    workspace_id = ctx.obj["WORKSPACE_ID"]
+    sources = Sources(api_client, workspace_id)
+    click.echo(sources)
+
+
+@workspace.command(help="List existing destinations in a workspace.")
+@click.pass_context
+def destinations(ctx: click.Context):
+    api_client = ctx.obj["API_CLIENT"]
+    workspace_id = ctx.obj["WORKSPACE_ID"]
+    destinations = Destinations(api_client, workspace_id)
+    click.echo(destinations)
+
+
+@workspace.command(help="List existing connections in a workspace.")
+@click.pass_context
+def connections(ctx: click.Context):
+    api_client = ctx.obj["API_CLIENT"]
+    workspace_id = ctx.obj["WORKSPACE_ID"]
+    connections = Connections(api_client, workspace_id)
+    click.echo(connections)
+
+
+AVAILABLE_COMMANDS: List[click.Command] = [connectors, workspace]
 
 
 def add_commands_to_list():

octavia-cli/octavia_cli/list/formatting.py

@@ -0,0 +1,59 @@
+#
+# Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+#
+
+from typing import List
+
+
+def compute_columns_width(data: List[List[str]], padding: int = 2) -> List[int]:
+    """Compute columns width for display purposes:
+    Find size for each columns in the data and add padding.
+    Args:
+        data (List[List[str]]): Tabular data containing rows and columns.
+        padding (int): Number of character to adds to create space between columns.
+    Returns:
+        columns_width (List[int]): The computed columns widths for each column according to input data.
+    """
+    columns_width = [0 for _ in data[0]]
+    for row in data:
+        for i, col in enumerate(row):
+            current_col_width = len(col) + padding
+            if current_col_width > columns_width[i]:
+                columns_width[i] = current_col_width
+    return columns_width
+
+
+def camelcased_to_uppercased_spaced(camelcased: str) -> str:
+    """Util function to transform a camelCase string to a UPPERCASED SPACED string
+    e.g: dockerImageName -> DOCKER IMAGE NAME
+    Args:
+        camelcased (str): The camel cased string to convert.
+
+    Returns:
+        (str): The converted UPPERCASED SPACED string
+    """
+    return "".join(map(lambda x: x if x.islower() else " " + x, camelcased)).upper()
+
+
+def display_as_table(data: List[List[str]]) -> str:
+    """Formats tabular input data into a displayable table with columns.
+    Args:
+        data (List[List[str]]): Tabular data containing rows and columns.
+    Returns:
+        table (str): String representation of input tabular data.
+    """
+    columns_width = compute_columns_width(data)
+    table = "\n".join(["".join(col.ljust(columns_width[i]) for i, col in enumerate(row)) for row in data])
+    return table
+
+
+def format_column_names(camelcased_column_names: List[str]) -> List[str]:
+    """Format camel cased column names to uppercased spaced column names
+
+    Args:
+        camelcased_column_names (List[str]): Column names in camel case.
+
+    Returns:
+        (List[str]): Column names in uppercase with spaces.
+    """
+    return [camelcased_to_uppercased_spaced(column_name) for column_name in camelcased_column_names]

octavia-cli/octavia_cli/list/listings.py

@@ -0,0 +1,111 @@
+#
+# Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+#
+
+import abc
+from typing import List
+
+import airbyte_api_client
+import octavia_cli.list.formatting as formatting
+from airbyte_api_client.api import connection_api, destination_api, destination_definition_api, source_api, source_definition_api
+from airbyte_api_client.model.workspace_id_request_body import WorkspaceIdRequestBody
+
+
+class BaseListing(abc.ABC):
+    COMMON_LIST_FUNCTION_KWARGS = {"_check_return_type": False}
+
+    @property
+    @abc.abstractmethod
+    def api(
+        self,
+    ):  # pragma: no cover
+        pass
+
+    @property
+    @abc.abstractmethod
+    def fields_to_display(
+        self,
+    ) -> List[str]:  # pragma: no cover
+        pass
+
+    @property
+    @abc.abstractmethod
+    def list_field_in_response(
+        self,
+    ) -> str:  # pragma: no cover
+        pass
+
+    @property
+    @abc.abstractmethod
+    def list_function_name(
+        self,
+    ) -> str:  # pragma: no cover
+        pass
+
+    @property
+    def _list_fn(self):
+        return getattr(self.api, self.list_function_name)
+
+    @property
+    def list_function_kwargs(self) -> dict:
+        return {}
+
+    def __init__(self, api_client: airbyte_api_client.ApiClient):
+        self.api_instance = self.api(api_client)
+
+    def _parse_response(self, api_response) -> List[List[str]]:
+        items = [[item[field] for field in self.fields_to_display] for item in api_response[self.list_field_in_response]]
+        return items
+
+    def get_listing(self) -> List[List[str]]:
+        api_response = self._list_fn(self.api_instance, **self.list_function_kwargs, **self.COMMON_LIST_FUNCTION_KWARGS)
+        return self._parse_response(api_response)
+
+    def __repr__(self):
+        items = [formatting.format_column_names(self.fields_to_display)] + self.get_listing()
+        return formatting.display_as_table(items)
+
+
+class SourceConnectorsDefinitions(BaseListing):
+    api = source_definition_api.SourceDefinitionApi
+    fields_to_display = ["name", "dockerRepository", "dockerImageTag", "sourceDefinitionId"]
+    list_field_in_response = "source_definitions"
+    list_function_name = "list_latest_source_definitions"
+
+
+class DestinationConnectorsDefinitions(BaseListing):
+    api = destination_definition_api.DestinationDefinitionApi
+    fields_to_display = ["name", "dockerRepository", "dockerImageTag", "destinationDefinitionId"]
+    list_field_in_response = "destination_definitions"
+    list_function_name = "list_latest_destination_definitions"
+
+
+class WorkspaceListing(BaseListing, abc.ABC):
+    def __init__(self, api_client: airbyte_api_client.ApiClient, workspace_id: str):
+        self.workspace_id = workspace_id
+        super().__init__(api_client)
+
+    @property
+    def list_function_kwargs(self) -> dict:
+        return {"workspace_id_request_body": WorkspaceIdRequestBody(workspace_id=self.workspace_id)}
+
+
+class Sources(WorkspaceListing):
+    api = source_api.SourceApi
+    fields_to_display = ["name", "sourceName", "sourceId"]
+    list_field_in_response = "sources"
+    list_function_name = "list_sources_for_workspace"
+
+
+class Destinations(WorkspaceListing):
+    api = destination_api.DestinationApi
+    fields_to_display = ["name", "destinationName", "destinationId"]
+    list_field_in_response = "destinations"
+    list_function_name = "list_destinations_for_workspace"
+
+
+class Connections(WorkspaceListing):
+    api = connection_api.ConnectionApi
+    fields_to_display = ["name", "connectionId", "status", "sourceId", "destinationId"]
+    list_field_in_response = "connections"
+    list_function_name = "list_connections_for_workspace"