"""
The schemas.sdp.processing_block module defines Marshmallow schemas .
"""
from marshmallow import fields, post_dump, post_load
from marshmallow_enum import EnumField
from ska_oso_pdm.entities.sdp import PbDependency, ProcessingBlock, Script, ScriptKind
from ska_oso_pdm.schemas.oso_marshmallow_schema import OSOSchema
__all__ = ["ScriptSchema", "PbDependencySchema", "ProcessingBlockSchema"]
[docs]class ScriptSchema(OSOSchema): # pylint: disable=too-few-public-methods
"""
Represents the type of workflow being configured on the SDP
"""
name = fields.String(data_key="name", required=True)
kind = EnumField(
ScriptKind,
dump_by=EnumField.VALUE,
load_by=EnumField.VALUE,
required=True,
data_key="kind",
error="{input} not a valid kind of processing script. Must be one of: {values}",
)
version = fields.String(data_key="version", required=True)
@post_load
def create_sdp_script(self, data, **_): # pylint: disable=no-self-use
"""
Convert parsed JSON back into a Script object.
:param data: Marshmallow-provided dict containing parsed JSON values
:param _: kwargs passed by Marshmallow
:return: SDP Script object populated from data
"""
name = data["name"]
kind = data["kind"]
version = data["version"]
return Script(name=name, kind=kind, version=version)
[docs]class PbDependencySchema(OSOSchema): # pylint: disable=too-few-public-methods
"""
Marshmallow schema for the PbDependency class.
"""
pb_id = fields.String(data_key="pb_id")
kind = fields.List(fields.String, data_key="kind")
@post_load
def create_pb_dependency(self, data, **_): # pylint: disable=no-self-use
"""
Convert parsed JSON back into a PbDependency object.
:param data: Marshmallow-provided dict containing parsed JSON values
:param _: kwargs passed by Marshmallow
:return: PbDependency object populated from data
"""
pb_id = data["pb_id"]
kind = data["kind"]
return PbDependency(pb_id=pb_id, kind=kind)
[docs]class ProcessingBlockSchema(OSOSchema):
"""
Marshmallow schema for the ProcessingBlock class.
"""
pb_id = fields.String(required=True)
script = fields.Nested(ScriptSchema, required=True)
sbi_ids = fields.List(fields.String, allow_none=True)
parameters = fields.Dict(allow_none=True)
dependencies = fields.Nested(
PbDependencySchema, many=True, missing=None, allow_none=True
)
@post_dump
def filter_nulls(self, data, **_): # pylint: disable=no-self-use
"""
Filter out null values from JSON.
:param data: Marshmallow-provided dict containing parsed object values
:param _: kwargs passed by Marshmallow
:return: dict suitable for PB configuration
"""
return {k: v for k, v in data.items() if v is not None}
@post_load
def create_processing_block(self, data, **_): # pylint: disable=no-self-use
"""
Convert parsed JSON back into a PB object.
:param data: Marshmallow-provided dict containing parsed JSON values
:param _: kwargs passed by Marshmallow
:return: PB object populated from data
"""
return ProcessingBlock(**data)