Source code for ska_oso_pdm.execution_block

from datetime import datetime, timezone

from pydantic import AwareDatetime, Field

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




class ErrorWrapper(PdmObject):
    detail: str | None = None
    stacktrace: str | None = None





class ResponseWrapper(PdmObject):
    result: str | None = None





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





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