feat: add support for platform login, cluster API CRUD methods

setup.py

@@ -74,6 +74,8 @@
         "packaging",  # Use same version as eth-ape
         "pydantic_settings",  # Use same version as eth-ape
         "taskiq[metrics]>=0.11.3,<0.12",
+        "tomlkit>=0.12,<1",  # For reading/writing global platform profile
+        "fief-client[cli]>=0.19,<1",  # for platform auth/cluster login
     ],
     entry_points={
         "console_scripts": ["silverback=silverback._cli:cli"],

silverback/_cli.py

@@ -11,11 +11,14 @@
     verbosity_option,
 )
 from ape.exceptions import Abort
+from fief_client.integrations.cli import FiefAuthNotAuthenticatedError
 from taskiq import AsyncBroker
 from taskiq.cli.worker.run import shutdown_broker
 from taskiq.receiver import Receiver
 
 from silverback._importer import import_from_string
+from silverback.platform.client import DEFAULT_PROFILE, PlatformClient
+from silverback.platform.types import ClusterConfiguration, ClusterTier
 from silverback.runner import PollingRunner, WebsocketRunner
 
 
@@ -142,3 +145,161 @@ def run(cli_ctx, account, runner_class, recorder, max_exceptions, path):
 def worker(cli_ctx, account, workers, max_exceptions, shutdown_timeout, path):
     app = import_from_string(path)
     asyncio.run(run_worker(app.broker, worker_count=workers, shutdown_timeout=shutdown_timeout))
+
+
+def platform_client(display_userinfo: bool = True):
+    def get_client(ctx, param, value) -> PlatformClient:
+        client = PlatformClient(profile_name=value)
+
+        # NOTE: We need to be authenticated to display userinfo
+        if not display_userinfo:
+            return client
+
+        try:
+            userinfo = client.userinfo  # cache this
+        except FiefAuthNotAuthenticatedError as e:
+            raise click.UsageError("Not authenticated, please use `silverback login` first.") from e
+
+        user_id = userinfo["sub"]
+        username = userinfo["fields"].get("username")
+        click.echo(
+            f"{click.style('INFO', fg='blue')}: "
+            f"Logged in as '{click.style(username if username else user_id, bold=True)}'"
+        )
+        return client
+
+    return click.option(
+        "-p",
+        "--profile",
+        "platform_client",
+        default=DEFAULT_PROFILE,
+        callback=get_client,
+        help="Profile to use for Authentication and Platform API Host.",
+    )
+
+
+class PlatformCommands(click.Group):
+    # NOTE: Override so we get the list ordered by definition order
+    def list_commands(self, ctx: click.Context) -> list[str]:
+        return list(self.commands)
+
+    def command(self, *args, display_userinfo=True, **kwargs):
+        profile_option = platform_client(display_userinfo=display_userinfo)
+        outer = super().command
+
+        def decorator(fn):
+            return outer(*args, **kwargs)(profile_option(fn))
+
+        return decorator
+
+
+@cli.group(cls=PlatformCommands)
+def cluster():
+    """Connect to hosted application clusters"""
+
+
+@cluster.command(display_userinfo=False)  # Otherwise would fail because not authorized
+def login(platform_client: PlatformClient):
+    """Login to hosted clusters"""
+    platform_client.auth.authorize()
+    userinfo = platform_client.userinfo  # cache this
+    user_id = userinfo["sub"]
+    username = userinfo["fields"]["username"]
+    click.echo(
+        f"{click.style('SUCCESS', fg='green')}: Logged in as "
+        f"'{click.style(username, bold=True)}' (UUID: {user_id})"
+    )
+
+
+@cluster.command(name="list")
+def list_clusters(platform_client: PlatformClient):
+    """List available clusters"""
+    if clusters := platform_client.clusters:
+        for cluster_name, cluster_info in clusters.items():
+            click.echo(f"{cluster_name}:")
+            click.echo(f"  status: {cluster_info.status}")
+            click.echo("  configuration:")
+            click.echo(f"    cpu: {256 * 2 ** cluster_info.configuration.cpu / 1024} vCPU")
+            memory_display = (
+                f"{cluster_info.configuration.memory} GB"
+                if cluster_info.configuration.memory > 0
+                else "512 MiB"
+            )
+            click.echo(f"    memory: {memory_display}")
+            click.echo(f"    networks: {cluster_info.configuration.networks}")
+            click.echo(f"    bots: {cluster_info.configuration.bots}")
+            click.echo(f"    history: {cluster_info.configuration.history}m")
+
+    else:
+        click.secho("No clusters for this account", bold=True, fg="red")
+
+
+@cluster.command(name="new")
+@click.option(
+    "-n",
+    "--name",
+    "cluster_name",
+    default="",
+    help="Name for new cluster (Defaults to random)",
+)
+@click.option(
+    "-t",
+    "--tier",
+    default=ClusterTier.PERSONAL.name,
+)
+@click.option("-c", "--config", "config_updates", type=(str, str), multiple=True)
+def new_cluster(
+    platform_client: PlatformClient,
+    cluster_name: str,
+    tier: str,
+    config_updates: list[tuple[str, str]],
+):
+    """Create a new cluster"""
+    base_configuration = getattr(ClusterTier, tier.upper()).configuration()
+    upgrades = ClusterConfiguration(
+        **{k: int(v) if v.isnumeric() else v for k, v in config_updates}
+    )
+    cluster = platform_client.create_cluster(
+        cluster_name=cluster_name,
+        configuration=base_configuration | upgrades,
+    )
+    # TODO: Create a signature scheme for ClusterInfo
+    #         (ClusterInfo configuration as plaintext, .id as nonce?)
+    # TODO: Test payment w/ Signature validation of extra data
+    click.echo(f"{click.style('SUCCESS', fg='green')}: Created '{cluster.name}'")
+
+
+@cluster.command()
+@click.option(
+    "-c",
+    "--cluster",
+    "cluster_name",
+    help="Name of cluster to connect with.",
+    required=True,
+)
+def bots(platform_client: PlatformClient, cluster_name: str):
+    """List all bots in a cluster"""
+    if not (cluster := platform_client.clusters.get(cluster_name)):
+        if clusters := "', '".join(platform_client.clusters):
+            message = f"'{cluster_name}' is not a valid cluster, must be one of: '{clusters}'."
+
+        else:
+            suggestion = (
+                "Check out https://silverback.apeworx.io "
+                "for more information on how to get started"
+            )
+            message = "You have no valid clusters to chose from\n\n" + click.style(
+                suggestion, bold=True
+            )
+        raise click.BadOptionUsage(
+            option_name="cluster_name",
+            message=message,
+        )
+
+    if bots := cluster.bots:
+        click.echo("Available Bots:")
+        for bot_name, bot_info in bots.items():
+            click.echo(f"- {bot_name} (UUID: {bot_info.id})")
+
+    else:
+        click.secho("No bots in this cluster", bold=True, fg="red")

silverback/platform/__init__.py

@@ -0,0 +1 @@
+# NOTE: Don't import anything here from `.client`

silverback/platform/client.py

@@ -0,0 +1,138 @@
+import os
+from functools import cache
+from pathlib import Path
+from typing import ClassVar
+
+import httpx
+import tomlkit
+from fief_client import Fief, FiefAccessTokenInfo, FiefUserInfo
+from fief_client.integrations.cli import FiefAuth
+
+from silverback.platform.types import ClusterConfiguration
+from silverback.version import version
+
+from .types import BotInfo, ClusterInfo
+
+CREDENTIALS_FOLDER = Path.home() / ".silverback"
+CREDENTIALS_FOLDER.mkdir(exist_ok=True)
+DEFAULT_PROFILE = "production"
+
+
+class ClusterClient(ClusterInfo):
+    # NOTE: Client used only for this SDK
+    platform_client: ClassVar[httpx.Client | None] = None
+
+    def __hash__(self) -> int:
+        return int(self.id)
+
+    @property
+    @cache
+    def client(self) -> httpx.Client:
+        assert self.platform_client, "Forgot to link platform client"
+        # NOTE: DI happens in `PlatformClient.client`
+        return httpx.Client(
+            base_url=f"{self.platform_client.base_url}/clusters/{self.name}",
+            cookies=self.platform_client.cookies,
+            headers=self.platform_client.headers,
+        )
+
+    @property
+    @cache
+    def openapi_schema(self) -> dict:
+        return self.client.get("/openapi.json").json()
+
+    @property
+    def bots(self) -> dict[str, BotInfo]:
+        # TODO: Actually connect to cluster and display options
+        return {}
+
+
+class PlatformClient:
+    def __init__(self, profile_name: str = DEFAULT_PROFILE):
+        if not (profile_toml := (CREDENTIALS_FOLDER / "profile.toml")).exists():
+            if profile_name != DEFAULT_PROFILE:
+                raise RuntimeError(f"create '{profile_toml}' to add custom profile")
+
+            # Cache this for later
+            profile_toml.write_text(
+                tomlkit.dumps(
+                    {
+                        DEFAULT_PROFILE: {
+                            "auth-domain": "https://account.apeworx.io",
+                            "host-url": "https://silverback.apeworx.io",
+                            "client-id": "lcylrp34lnggGO-E-KKlMJgvAI4Q2Jhf6U2G6CB5uMg",
+                        }
+                    }
+                )
+            )
+
+        if not (profile := tomlkit.loads(profile_toml.read_text()).get(profile_name)):
+            raise RuntimeError(f"Unknown profile {profile_name}")
+
+        fief = Fief(profile["auth-domain"], profile["client-id"])
+        self.auth = FiefAuth(fief, str(CREDENTIALS_FOLDER / f"{profile_name}.json"))
+
+        # NOTE: Use `SILVERBACK_PLATFORM_HOST=http://127.0.0.1:8000` for local testing
+        self.base_url = os.environ.get("SILVERBACK_PLATFORM_HOST") or profile["host-url"]
+
+    @property
+    @cache
+    def client(self) -> httpx.Client:
+        client = httpx.Client(
+            base_url=self.base_url,
+            # NOTE: Raises `FiefAuthNotAuthenticatedError` if access token not available
+            cookies={"session": self.access_token_info["access_token"]},
+            headers={"User-Agent": f"Silverback SDK/{version}"},
+            follow_redirects=True,
+        )
+
+        # Detect connection fault early
+        try:
+            self.openapi = client.get("/openapi.json").json()
+        except httpx.ConnectError:
+            raise RuntimeError(f"No Platform API Host detected at '{self.base_url}'.")
+        except Exception:
+            raise RuntimeError(f"Error with API Host at '{self.base_url}'.")
+
+        # DI for `ClusterClient`
+        ClusterClient.platform_client = client  # Connect to client
+        return client
+
+    @property
+    def userinfo(self) -> FiefUserInfo:
+        return self.auth.current_user()
+
+    @property
+    def access_token_info(self) -> FiefAccessTokenInfo:
+        return self.auth.access_token_info()
+
+    @property
+    @cache
+    def clusters(self) -> dict[str, ClusterClient]:
+        response = self.client.get("/clusters/")
+        response.raise_for_status()
+        clusters = response.json()
+        # TODO: Support paging
+        return {cluster.name: cluster for cluster in map(ClusterClient.parse_obj, clusters)}
+
+    def create_cluster(
+        self,
+        cluster_name: str = "",
+        configuration: ClusterConfiguration = ClusterConfiguration(),
+    ) -> ClusterClient:
+        if (
+            response := self.client.post(
+                "/clusters/",
+                params=dict(name=cluster_name),
+                json=configuration.model_dump(),
+            )
+        ).status_code >= 400:
+            message = response.text
+            try:
+                message = response.json().get("detail", response.text)
+            except Exception:
+                pass
+
+            raise RuntimeError(message)
+
+        return ClusterClient.parse_raw(response.text)