1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
|
"""Base types and internal models."""
from abc import ABC
from dataclasses import dataclass
from enum import StrEnum
from typing import Any, Literal
from mashumaro import DataClassDictMixin
from mashumaro.config import BaseConfig
from mashumaro.mixins.orjson import DataClassORJSONMixin
class SentinelMeta(type):
"""Metaclass for sentinel to improve representation and make falsy.
Credit to https://stackoverflow.com/a/69243488 .
"""
def __repr__(cls) -> str:
"""Represent class more like an enum."""
return f"<{cls.__name__}>"
def __bool__(cls) -> Literal[False]:
"""Return false as a sentinel is akin to an empty value."""
return False
class DEFAULT(metaclass=SentinelMeta):
"""Sentinel for default value when None is valid."""
class RequestConfig(BaseConfig):
"""Default Mashumaro config for all request models."""
omit_default = True
@dataclass(frozen=True)
class Request(ABC, DataClassDictMixin):
"""Omit default in requests to allow Supervisor to set default.
If None is a valid value, the default value should be the sentinel
DEFAULT for optional fields.
"""
class Config(RequestConfig):
"""Mashumaro config."""
@dataclass(frozen=True)
class Options(ABC, DataClassDictMixin):
"""Superclass for Options models to ensure a field is present.
All fields should be optional. If None is a valid value, use the DEFAULT
sentinel. Client should only pass changed fields to Supervisor.
"""
def __post_init__(self) -> None:
"""Validate at least one field is present."""
if not self.to_dict():
raise ValueError("At least one field must have a value")
class Config(RequestConfig):
"""Mashumaro config."""
@dataclass(frozen=True)
class ResponseData(ABC, DataClassDictMixin):
"""Superclass for all response data objects."""
class ResultType(StrEnum):
"""ResultType type."""
OK = "ok"
ERROR = "error"
@dataclass(frozen=True, slots=True)
class Response(DataClassORJSONMixin):
"""Response model for all JSON based endpoints."""
result: ResultType
data: Any | None = None
message: str | None = None
job_id: str | None = None
error_key: str | None = None
extra_fields: dict[str, Any] | None = None
@dataclass(frozen=True, slots=True)
class ContainerStats(ResponseData):
"""ContainerStats model."""
cpu_percent: float
memory_usage: int
memory_limit: int
memory_percent: float
network_rx: int
network_tx: int
blk_read: int
blk_write: int
|
