Add token messages support

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- Token messages support.
+
 ## [9.0.0 - 2023-02-08]
 
 ### Added

src/sporestack/api.py

@@ -5,6 +5,8 @@ SporeStack API request/response models
 """
 
 
+from datetime import datetime
+from enum import Enum
 from typing import Dict, List, Optional
 
 from pydantic import BaseModel, Field
@@ -190,3 +192,25 @@ class OperatingSystems:
 
     class Response(BaseModel):
         operating_systems: List[str]
+
+
+class TokenMessageSender(Enum):
+    USER = "User"
+    SPORESTACK = "SporeStack"
+
+
+class TokenMessage(BaseModel):
+    message: str = Field(
+        ...,
+        title="Message",
+        min_length=1,
+        max_length=10_000,
+    )
+    sent_at: datetime = Field(
+        ...,
+        title="Sent At",
+        description="When the message was sent.",
+    )
+    sender: TokenMessageSender = Field(
+        ..., title="Sender", description="Who sent the message."
+    )

src/sporestack/api_client.py

@@ -2,9 +2,10 @@ import logging
 import os
 from dataclasses import dataclass
 from time import sleep
-from typing import Any, Dict, Optional
+from typing import Any, Dict, List, Optional
 
 import httpx
+from pydantic import parse_obj_as
 
 from . import __version__, api, exceptions
 
@@ -290,3 +291,23 @@ class APIClient:
         response = self._api_request(url=url)
         response_object = api.TokenBalance.Response.parse_obj(response)
         return response_object
+
+    def token_get_messages(self, token: str) -> List[api.TokenMessage]:
+        """Get messages for/from the token."""
+        url = self.api_endpoint + f"/token/{token}/messages"
+        log.debug(f"Token send message URL: {url}")
+        response = self._httpx_client.get(url=url)
+        if response.status_code == 422:
+            raise exceptions.SporeStackUserError(response.json()["detail"])
+        response.raise_for_status()
+
+        return parse_obj_as(List[api.TokenMessage], response.json())
+
+    def token_send_message(self, token: str, message: str) -> None:
+        """Send a message to SporeStack support."""
+        url = self.api_endpoint + f"/token/{token}/messages"
+        response = self._httpx_client.post(url=url, json={"message": message})
+        if response.status_code == 422:
+            raise exceptions.SporeStackUserError(response.json()["detail"])
+
+        response.raise_for_status()

src/sporestack/cli.py

@@ -44,7 +44,12 @@ cli.add_typer(token_cli, name="token")
 server_cli = typer.Typer(help="Commands to interact with SporeStack servers")
 cli.add_typer(server_cli, name="server")
 
-logging.basicConfig(level=logging.INFO)
+_log_level = os.getenv("LOG_LEVEL", "info").upper()
+_numeric_log_level = getattr(logging, _log_level, None)
+if _numeric_log_level is None:
+    raise ValueError(f"LOG_LEVEL: {_log_level} is invalid. Aborting!")
+assert isinstance(_numeric_log_level, int)
+logging.basicConfig(level=_numeric_log_level)
 
 DEFAULT_TOKEN = "primary"
 DEFAULT_FLAVOR = "vps-1vcpu-1gb"
@@ -760,6 +765,44 @@ def token_list() -> None:
         typer.echo(f"{token}: {key}")
 
 
+@token_cli.command()
+def messages(token: str = typer.Argument(DEFAULT_TOKEN)) -> None:
+    """
+    Show support messages.
+    """
+    token = load_token(token)
+
+    from .api_client import APIClient
+    from .client import Client
+
+    api_client = APIClient(api_endpoint=get_api_endpoint())
+    client = Client(api_client=api_client, client_token=token)
+
+    for message in client.token.messages():
+        typer.echo()
+        typer.echo(message.message)
+        typer.echo()
+        typer.echo(f"Sent at {message.sent_at}, by {message.sender.value}")
+
+
+@token_cli.command()
+def send_message(
+    token: str = typer.Argument(DEFAULT_TOKEN), message: str = typer.Option(...)
+) -> None:
+    """
+    Send a support message.
+    """
+    token = load_token(token)
+
+    from .api_client import APIClient
+    from .client import Client
+
+    api_client = APIClient(api_endpoint=get_api_endpoint())
+    client = Client(api_client=api_client, client_token=token)
+
+    client.token.send_message(message)
+
+
 @cli.command()
 def version() -> None:
     """

src/sporestack/client.py

@@ -64,6 +64,14 @@ class Token:
         """Returns the token's balance in cents."""
         return self.api_client.token_balance(token=self.token).cents
 
+    def messages(self) -> List[api.TokenMessage]:
+        """Returns support messages for/from the token."""
+        return self.api_client.token_get_messages(token=self.token)
+
+    def send_message(self, message: str) -> None:
+        """Returns support messages for/from the token."""
+        self.api_client.token_send_message(token=self.token, message=message)
+
     def servers(self) -> List[Server]:
         server_classes: List[Server] = []
         for server in self.api_client.servers_launched_from_token(self.token).servers: