mirror of
https://github.com/fusion44/blitz_api.git
synced 2026-08-14 12:02:46 +02:00
f1aa2107fd
- rename various keys - update app status updater code to work like the app management code - remove some unused code refs #123
90 lines
2.8 KiB
Python
90 lines
2.8 KiB
Python
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
|
|
)
|