calculate-utils-4-lib/calculate/server/ipc_draft/schemas.py

from typing import Any, List, Dict, Optional, Literal, Union
from pydantic import BaseModel, Field, HttpUrl
# from utils import ResponseStructure
# from pprint import pprint
# import json
# from fastapi import HTTPException, status


class NotTemplatedLinkData(BaseModel):
    href: HttpUrl = "URI"

    class Config:
        extra = "forbid"


class TemplatedLinkData(NotTemplatedLinkData):
    templated: bool = False


LinkData = Union[NotTemplatedLinkData, TemplatedLinkData]


# GET /
class ServerInfo(BaseModel):
    name: str = "Server name"
    version: str = "Server version"


class GetRootResponse(BaseModel):
    data: ServerInfo
    links: Dict[str, LinkData] = Field(..., alias="_links",
                                       description="Links to resources")


# GET /commands/{CID}/parameters
class ParameterInfo(BaseModel):
    id: str
    default: Any


class ParametersGroup(BaseModel):
    group_id: str
    parameters: List[ParameterInfo]


class GetCommandParametersResponse(BaseModel):
    data: List[ParametersGroup]
    links: Dict[str, LinkData] = Field(..., alias="_links")


# GET /commands/{CID}
class CommandData(BaseModel):
    id: str
    title: str
    category: str
    command: str


class CommandInfo(BaseModel):
    data: CommandData
    links: Dict[str, LinkData] = Field(..., alias="_links")


class FoundCommandInfo(CommandInfo):
    embedded: Dict[str,
                   GetCommandParametersResponse] = Field(...,
                                                         alias="_embedded")


# GET /commands/
class GetCommandsResponse(BaseModel):
    __root__: Dict[str, CommandInfo]

    # class Config:
    #     @staticmethod
    #     def schema_extra(schema, model):
    #         example = {}
    #         for num in range(2):
    #             resp = ResponseStructure()
    #             resp.add_data(id=f"command_{num}",
    #                           command=f"cl-command-{num}",
    #                           title=f"Command {num}",
    #                           category="cool-category"
    #                           )
    #             resp.add_link("parameters",
    #                           f"/commands/command_{num}/parameters")
    #             resp.add_link("configure", f"/configs/command_{num}")

    #             example[f"command_{num}"] = resp.get_dict()
    #         schema["example"] = example


# POST /configs/{command}
class CreateConfigResponse(BaseModel):
    links: Dict[str, LinkData] = Field(..., alias="_links")


# PATCH /configs/{WID}/parameters
class ParameterValue(BaseModel):
    id: str
    value: Any


class ModifyConfigRequest(BaseModel):
    __root__: List[ParameterValue]


class ModifyConfigCorrectResponse(BaseModel):
    links: Dict[str, LinkData] = Field(..., alias="_links")

    class Config:
        extra = "forbid"


class ModifyConfigUncorrectResponse(ModifyConfigCorrectResponse):
    data: Optional[Dict[str, str]]
    links: Dict[str, LinkData] = Field(..., alias="_links")


ModifyConfigResponse = Union[ModifyConfigCorrectResponse,
                             ModifyConfigUncorrectResponse]


# PUT /configs/{WID}/parameters
class ResetConfigRequest(BaseModel):
    __root__: List[ParameterValue]


class ResetConfigResponse(BaseModel):
    data: Optional[Dict[str, str]]
    links: Dict[str, LinkData] = Field(..., alias="_links")


# POST /executions/{WID}
class CreateExecutionResponse(BaseModel):
    links: Dict[str, LinkData] = Field(..., alias="_links")


# DELETE /executions/{WID}
class OutputData(BaseModel):
    type: Literal["output"]
    logging: Union[int, None]
    text: str
    source: str


class OutputMessage(BaseModel):
    data: OutputData


class StopExecutionsResponse(BaseModel):
    __root__: List[OutputMessage]


# DELETE /configs/{WID}
class StopConfigResponse(BaseModel):
    __root__: List[OutputMessage]


# GET /executions/{WID}/output
class InputRequestData(BaseModel):
    type: Literal["input"]
    text: str
    source: str


class InputMessage(BaseModel):
    data: InputRequestData
    links: Dict[str, LinkData] = Field(..., alias="_links")


class GetExecutionOutputResponse(BaseModel):
    __root__: List[Union[OutputMessage, InputMessage]]


# PATCH /executions/{WID}/input
class WriteToExecutionRequest(BaseModel):
    data: Any


# GET /workers/
class WorkerData(BaseModel):
    socket: str
    command: str
    pid: int


class WorkerInfo(BaseModel):
    data: WorkerData
    links: Dict[str, LinkData] = Field(..., alias="_links")


class GetWorkersResponse(BaseModel):
    __root__: Dict[int, WorkerInfo]

# DELETE /workers/{WID}


# Сообщения протокола взаимодействия сервера и воркеров.
class WorkerMessageBase(BaseModel):
    state: str
    status: int