from enum import Enum from typing import Any, Dict, List, Optional, Sequence, Union from fastapi import Query from pydantic import BaseModel class ProcessResult: return_code: int | None stdout: str stderr: str def __init__(self, return_code: int | None, stdout: str, stderr: str) -> None: self.return_code = return_code self.stdout = stdout self.stderr = stderr def __str__(self) -> str: return ( f"ProcessResult: \nreturn_code: {self.return_code}\n" f"stdout: {self.stdout}\n" f"stderr: {self.stderr}" ) class ApiErrors(str, Enum): INVALID_REQUEST_INPUT = "invalid_request_input" UNABLE_TO_PROCESS_ERROR = "unable_to_process_error" APP_STATUS_UPDATE_FAILED = "app_status_update_failed" APP_INVALID_FOR_PLATFORM = "app_invalid_for_platform" APP_MANAGE_ON_IS_ALREADY_INSTALLED = "app_manage_on_is_already_installed" APP_MANAGE_OFF_IS_NOT_INSTALLED = "app_manage_off_is_not_installed" APP_MANAGE_LOCK_HELD = "app_manage_lock_held" BACKGROUND_TASK_FAILED = "background_task_failed" UNKNOWN = "unknown" class RedisChannelMessageData(BaseModel): timestamp: str = Query(..., description="The timestamp of the message") key: str = Query(..., description="The key of the message") json_contents: str | None = Query(None, description="The contents of the message") class ErrorMessage(BaseModel): detail: str = Query(..., description="short text representation of the error") error_code: str = Query("", description="a unique identifier for the error") report: Optional[Union[str, Dict, List, Sequence[Any]]] = Query( None, description="optional more detailed message", ) trace: Optional[Sequence[str]] = Query( None, description="optional stack trace of the error " "(only when configured with BAPI_SEND_TRACE=true)", ) class StartupState(str, Enum): OFFLINE = "offline" BOOTSTRAPPING = "bootstrapping" LOCKED = "locked" DONE = "done" DISABLED = "disabled" class ApiStartupStatus(BaseModel): def __init__( __pydantic_self__, bitcoin: StartupState = StartupState.OFFLINE, bitcoin_msg: Optional[str] = "", lightning: StartupState = StartupState.OFFLINE, lightning_msg: Optional[str] = "", ) -> "ApiStartupStatus": super().__init__( bitcoin=bitcoin, bitcoin_msg=bitcoin_msg, lightning=lightning, lightning_msg=lightning_msg, ) bitcoin: StartupState bitcoin_msg: Optional[str] lightning: StartupState lightning_msg: Optional[str] def is_fully_initialized(self): return self.bitcoin == StartupState.DONE and ( self.lightning == StartupState.DONE or self.lightning == StartupState.DISABLED )