Commands¶

class CommandPayload(BaseModel):
    """Base class for the payload models of commands.

    Unknown fields are ignored, so callers can keep sending extra keys
    without breaking a command.

    """

    model_config = ConfigDict(extra="ignore")

class PluginCmd(BaseCmd[TCmd]):
    """A command that acts on a plugin."""

@dataclass
class PluginCmdContext(BaseCmdContext, Generic[P]):
    """Context for a command acting on a plugin.

    Attributes:
        plugin: The plugin the command acts on.

    """

    plugin: P

class ServiceCmd(BaseCmd[TCmd]):
    """A command that acts on a service."""

@dataclass
class ServiceCmdContext(BaseCmdContext, Generic[S]):
    """Context for a command acting on a service.

    Attributes:
        service: The service the command acts on.

    """

    service: S

class CommandError(Exception):
    """Base for expected, client-caused command-handling failures.

    These represent bad input from a caller (unknown command, bad payload),
    not internal bugs. Callers may catch this family to log without a stack
    trace and return a structured error response.

    """

    def __str__(self) -> str:
        # Return the plain message. Avoids KeyError's quote-wrapping of
        # args[0] so logs and error responses read cleanly.
        return str(self.args[0]) if self.args else super().__str__()

class MessageFormatError(CommandError, KeyError):
    """Raised when a required key is missing from a message dict.

    Args:
        key: The offending key.
        detail: What is wrong with the message, when it is not simply a missing
            key. Replaces the default "missing key" text.

    Attributes:
        key: The offending key.

    """

    def __init__(self, key: str, *, detail: str | None = None):
        reason = detail or f"the required key '{key}' is missing from the dictionary"
        super().__init__(f"Message format error: {reason}")
        self.key = key

class UnknownCommandError(CommandError, ValueError):
    """Raised when a command name is not registered.

    Args:
        command: The unknown command name.

    Attributes:
        command: The unknown command name.

    """

    def __init__(self, command: str):
        super().__init__(f"Unknown command error: {command}")
        self.command = command

class MissingPluginIdError(CommandError, ValueError):
    """Raised when a command payload has no plugin 'id'."""

    def __init__(self) -> None:
        super().__init__("Missing plugin 'id' in command payload.")

class PluginNotFoundError(CommandError, KeyError):
    """Raised when no plugin is registered for the given id.

    Args:
        plugin_id: The id used for lookup.

    """

    def __init__(self, plugin_id: str):
        super().__init__(f"Plugin with id '{plugin_id}' not found.")
        self.plugin_id = plugin_id

class InvalidPayloadError(CommandError, ValueError):
    """Raised when a command payload does not match what the command expects.

    Args:
        command: The name of the command that rejected the payload.
        details: What is wrong with the payload.

    Attributes:
        command: The name of the command that rejected the payload.
        details: What is wrong with the payload.

    """

    def __init__(self, command: str, details: str):
        super().__init__(f"Invalid payload for '{command}': {details}")
        self.command = command
        self.details = details

    @classmethod
    def from_validation_error(cls, command: str, error: ValidationError) -> InvalidPayloadError:
        """Build the error from a Pydantic validation failure.

        Args:
            command: The name of the command that rejected the payload.
            error: The validation failure, one `loc: message` entry per issue.

        Returns:
            The error, with every issue listed in `details`.

        """
        details = "; ".join(
            f"{'.'.join(str(part) for part in e['loc']) or '<payload>'}: {e['msg']}" for e in error.errors()
        )
        return cls(command, details)