Source code for ska_oso_pdm.execution_block

from datetime import datetime, timezone
from typing import Optional

from pydantic import AwareDatetime, Field

from ska_oso_pdm._shared import (
    ExecutionBlockID,
    Metadata,
    PdmObject,
    PythonArguments,
    SBDefinitionID,
    SBInstanceID,
    TelescopeType,
)




class ErrorWrapper(PdmObject):
    detail: Optional[str] = None
    stacktrace: Optional[str] = None





class ResponseWrapper(PdmObject):
    result: Optional[str] = None





class RequestResponse(PdmObject):
    request: Optional[str] = None
    request_args: Optional[PythonArguments] = None
    status: Optional[str] = None
    response: Optional[ResponseWrapper] = None
    error: Optional[ErrorWrapper] = None
    request_sent_at: AwareDatetime = Field(
        default_factory=lambda: datetime.now(timezone.utc)
    )
    response_received_at: Optional[AwareDatetime] = None





class OSOExecutionBlock(PdmObject):
    interface: Optional[str] = None
    eb_id: Optional[ExecutionBlockID] = None
    metadata: Optional[Metadata] = None
    telescope: TelescopeType
    sbd_ref: Optional[SBDefinitionID] = None
    sbd_version: Optional[int] = None
    sbi_ref: Optional[SBInstanceID] = None
    request_responses: list[RequestResponse] = Field(default_factory=list)