Merge branch 'dev'

Author: Dr0p42

naas_python/__init__.py

@@ -1,5 +1,6 @@
 from .domains.registry.handlers.PythonHandler import primaryAdaptor as registry
 from .domains.space.handlers.PythonHandler import primaryAdaptor as space
+from .domains.secret.handlers.PythonHandler import primaryAdaptor as secret
 from .utils.log import initialize_logging
 
 logger = initialize_logging()

naas_python/cli.py

@@ -9,6 +9,11 @@
     primaryAdaptor as typerSpaceAdaptor,
 )
 
+# from naas_python.domains import secretCliAdaptor  # , secretCliAdaptor
+from naas_python.domains.secret.handlers.CLISecretHandler import (
+    primaryAdaptor as typerSecretAdaptor,
+)
+
 
 def _create_cli_app():
     app = typer.Typer(
@@ -21,6 +26,7 @@ def _create_cli_app():
     # Registry domain's related commands
     app.add_typer(typerSpaceAdaptor.app, name="space")
     app.add_typer(typerRegistryAdaptor.app, name="registry")
+    app.add_typer(typerSecretAdaptor.app, name="secret")
 
     return app
 

naas_python/domains/secret/SecretDomain.py

@@ -0,0 +1,40 @@
+from typing import List
+from .models.Secret import Secret
+
+from typing import List
+
+from naas_python.domains.secret.SecretSchema import (
+    ISecretDomain,
+    ISecretAdaptor,
+    Secret,
+    SecretListResponse,
+    SecretResponseError,
+    SecretError,
+    SecretCreateResponse,
+    SecretCreateRequest,
+    # SecretCredentialsResponse,
+    # SecretAdd
+)
+class SecretDomain(ISecretDomain):
+    def __init__(self, adaptor: ISecretAdaptor):
+        self.adaptor = adaptor
+
+    def create(self, name: str, value: str) -> None:
+        response = self.adaptor.create_secret(
+            name=name, value=value,
+        )
+
+        return response
+
+    def get(self, name: str) -> Secret:
+        response = self.adaptor.get_secret(name=name)
+        return response
+
+    def delete(self, name: str) -> None:
+        return self.adaptor.delete_secret(name=name)
+
+    def list(self, page_size: int, page_number: int) -> List[Secret]:
+        secrets = self.adaptor.list_secrets(
+            page_size=page_size, page_number=page_number
+        )
+        return secrets

naas_python/domains/secret/SecretSchema.py

@@ -0,0 +1,84 @@
+from abc import ABCMeta, abstractmethod
+from logging import getLogger
+from typing import Any
+from typing import List
+
+from naas_models.pydantic.secret_p2p import *
+
+from naas_python.utils.exceptions import NaasException
+
+logger = getLogger(__name__)
+
+# Secondary adaptor
+
+
+class ISecretAdaptor(metaclass=ABCMeta):
+
+    @abstractmethod
+    def create_secret(self, name: str, value: str) -> None:
+        raise NotImplementedError
+
+    @abstractmethod
+    def get_secret(self, name: str) -> Secret:
+        raise NotImplementedError
+
+    @abstractmethod
+    def list_secrets(self, page_size: int, page_number: int) -> List[Secret]:
+        raise NotImplementedError
+
+    @abstractmethod
+    def delete_secret(self, name: str) -> None:
+        raise NotImplementedError
+
+
+# Domain
+
+
+class ISecretDomain(metaclass=ABCMeta):
+    adaptor: ISecretAdaptor
+
+    @abstractmethod
+    def create(self, name: str, value: str) -> None:
+        raise NotImplementedError
+
+    @abstractmethod
+    def get(self, name: str) -> Secret:
+        raise NotImplementedError
+
+    @abstractmethod
+    def list(self, page_size: int, page_number: int) -> List[Secret]:
+        raise NotImplementedError
+
+    @abstractmethod
+    def delete(self, name: str) -> None:
+        raise NotImplementedError
+
+
+# Primary Adaptor
+
+
+class ISecretInvoker(metaclass=ABCMeta):
+    @abstractmethod
+    def create(self, **kwargs):
+        raise NotImplementedError
+
+    @abstractmethod
+    def get(self, **kwargs):
+        raise NotImplementedError
+
+    @abstractmethod
+    def list(self, **kwargs):
+        raise NotImplementedError
+
+    @abstractmethod
+    def delete(self, **kwargs):
+        raise NotImplementedError
+
+
+# Exceptions
+class SecretValidationError(NaasException):
+    pass
+class SecretConflictError(NaasException):
+    pass
+class SecretNotFound(NaasException):
+    pass

naas_python/domains/secret/adaptors/primary/SDKSecretAdaptor.py

@@ -0,0 +1,41 @@
+import os
+
+from rich.panel import Panel
+from rich import print as rprint
+from typing import List
+
+from naas_python.domains.secret.SecretSchema import (
+    ISecretDomain,
+    ISecretInvoker,
+    SecretCreateResponse,
+    SecretGetResponse,
+    SecretListResponse,
+    SecretDeleteResponse,
+    Secret
+)
+
+class SDKSecretAdaptor(ISecretInvoker):
+    domain: ISecretDomain
+
+    def __init__(self, domain: ISecretDomain):
+        self.domain = domain
+
+    def create(self, name: str = "", value: str ="") -> None:
+        """Create a secret with the given name"""
+        secret = self.domain.create(name=name, value=value)
+        return secret
+
+    def list(self, page_size: int = 0, page_number: int = 0) -> List[Secret]:
+        """List all secrets for the current user"""
+        secret_list = self.domain.list(page_size=page_size, page_number=page_number)
+        return secret_list
+
+    def get(self, name="") -> Secret:
+        """Get a secret with the given name"""
+        secret = self.domain.get(name=name)
+        return secret
+
+    def delete(self, name="") -> None:
+        """Delete a secret by name"""
+        secret = self.domain.delete(name=name)
+        return secret

naas_python/domains/secret/adaptors/primary/TyperSecretAdaptor.py

@@ -0,0 +1,174 @@
+import json
+import os
+import time
+from logging import getLogger
+from typing import List
+from uuid import UUID
+
+import rich
+import typer
+from click import Context
+from pydantic import BaseModel
+from rich.console import Console
+from rich.progress import Progress
+from rich.table import Table
+from typer.core import TyperGroup
+from typing_extensions import Annotated
+
+from naas_python.domains.secret.adaptors.primary.utils import PydanticTableModel
+from naas_python.domains.secret.SecretSchema import (
+    ISecretDomain,
+    ISecretInvoker,
+    SecretConflictError,
+)
+# from naas_python.domains.secret.SecretSchema import SecrettryConflictError
+from naas_python.utils.cicd import Pipeline
+
+logger = getLogger(__name__)
+
+
+class OrderCommands(TyperGroup):
+    def list_commands(self, ctx: Context):
+        """Return list of commands in the order appear."""
+        return list(self.commands)
+
+
+class TyperSecretAdaptor(ISecretInvoker):
+    def __init__(self, domain: ISecretDomain):
+        super().__init__()
+
+        self.domain = domain
+        self.console = Console()
+
+        self.app = typer.Typer(
+            cls=OrderCommands,
+            help="Naas Secret CLI",
+            add_completion=False,
+            no_args_is_help=True,
+            pretty_exceptions_enable=False,
+            rich_markup_mode="rich",
+            context_settings={"help_option_names": ["-h", "--help"]},
+        )
+
+        # Include all commands
+        self.app.command()(self.list)
+        self.app.command()(self.create)
+        self.app.command()(self.get)
+        self.app.command()(self.delete)
+
+    def _list_preview(self, data: List[dict], headers: list):
+        if not isinstance(data, list):
+            raise TypeError("Data must be a list of dicts, not {}".format(type(data)))
+
+        # Determine column widths based on the longest values
+        column_widths = [max(len(str(item)) for item in col) for col in zip(*data)]
+
+        # Print the headers
+        header_format = "  ".join(
+            f"{header:<{width}}" for header, width in zip(headers, column_widths)
+        )
+        print(header_format)
+
+        # Print the data
+        for row in data:
+            row_format = "  ".join(
+                f"{str(item):<{width}}" for item, width in zip(row, column_widths)
+            )
+            print(row_format)
+
+    def create(
+        self,
+        name: str = typer.Option(..., "--name", "-n", help="Name of the secret"),
+        value: str = typer.Option(..., "--value", "-v", help="Value of the secret"),
+
+        rich_preview: bool = typer.Option(
+            False,
+            "--rich-preview",
+            "-rp",
+            help="Rich preview of the Secret information as a table",
+        ),
+    ):
+        """Create a Secret with the given specifications"""
+        secret = self.domain.create(
+            name=name,
+            value=value
+        )
+
+        if secret is None:
+            print('Secret Successfully created')
+
+    def get(
+        self,
+        name: str = typer.Option(..., "--name", "-n", help="Name of the secret"),
+        rich_preview: bool = typer.Option(
+            os.environ.get("NAAS_CLI_RICH_PREVIEW", False),
+            "--rich-preview",
+            "-rp",
+            help="Rich preview of the secret information as a table",
+        ),
+    ):
+        """Get a secret with the given name"""
+        secret = self.domain.get(name=name)
+        
+        if rich_preview:
+            self.console.print(PydanticTableModel([secret]).table)
+
+        else:
+            print(secret.value)
+
+    def delete(
+        self,
+        name: str = typer.Option(..., "--name", "-n", help="Name of the secret"),
+    ):
+        self.domain.delete(name=name)
+
+        print(f"Secret '{name}' deleted successfully")
+
+    def list(
+        self,
+        page_size: int = typer.Option(0, help="Size of each page of results"),
+        page_number: int = typer.Option(0, help="Target page number of results"),
+        rich_preview: bool = typer.Option(
+            False,
+            "--rich-preview",
+            "-rp",
+            help="Rich preview of the secret information as a table",
+        ),
+    ):
+        """List all secrets for the current user"""
+        secret_list = self.domain.list(page_size=page_size, page_number=page_number)
+
+        data = []
+        headers = []
+
+        # Extract the data and headers
+        for secret in secret_list:
+            _secret_dict = secret.dict()
+
+
+            data.append(list(_secret_dict.values()))  # Append a list of values to data
+
+            headers = [key.upper() for key in _secret_dict.keys()]
+
+        if len(data) == 0:
+            print("No matching results found.")
+            return
+
+        headers = [key.upper() for key in _secret_dict.keys()]
+
+        if rich_preview:
+            # Create a Rich Table
+            table = Table(show_header=True, header_style="bold")
+            # Add columns to the table
+            for header in headers:
+                table.add_column(header, justify="center")
+
+            # Add data rows to the table
+            for row in data:
+                table.add_row(*row)
+
+            # Print the table
+            rich.print(table)
+
+        else:
+            self._list_preview(data, headers)