"""Processing block configuration entities."""
from typing import Annotated
from pydantic import BaseModel, Field
from .base import MultiEntityBaseModel
from .common import ExecutionBlockId, ProcessingBlockId
from .script import Script
[docs]
class PBDependency(BaseModel):
"""A Processing Block dependency."""
kind: list[str]
"""How this dependency is meant to be interpreted."""
pb_id: ProcessingBlockId
"""The ID of the Processing Block that is depended on."""
[docs]
class ProcessingBlock(MultiEntityBaseModel):
"""Processing block entity.
Collects configuration information relating to a processing job for the
SDP. This might be either real-time (supporting a running observation) or
batch (to process data after the fact).
Actual execution of processing steps will be performed by a (parameterised)
processing script interpreting processing block information.
"""
key: ProcessingBlockId
eb_id: ExecutionBlockId | None
"""
ID of the Execution Block associated with this Processing Block, if any.
"""
script: Script.Key
"""The script that will execute this Processing Block."""
parameters: Annotated[dict, Field(default_factory=dict)]
"""Parameters for the Processing Block."""
dependencies: Annotated[list[PBDependency], Field(default_factory=list)]
"""Dependencies of this Processing Block on others."""
# kept for backwards compatibility
[docs]
def to_dict(self):
"""
Returns a dictionary representation of this entity, kept for backwards
compatibility. Prefer :meth:`model_dump`.
"""
return self.model_dump()
# kept for backwards compatibility
@property
def pb_id(self) -> str:
"""This Processing Block's ID"""
return self.key