"""
The schemas.common.procedures defines Marshmallow schema that map the
activities section of an SKA scheduling block to/from a JSON
representation.
"""
from marshmallow import Schema, fields, post_dump, post_load
from marshmallow_oneofschema import OneOfSchema
from ...entities.common.procedures import (
EmbeddedScript,
FileSystemScript,
GitScript,
PythonArguments,
)
[docs]class PythonArgumentsSchema(Schema):
"""
Schema for the PythonArguments, used in the activities section of an SKA scheduling block
"""
args = fields.List(fields.String())
kwargs = fields.Dict()
[docs] @post_load
def make_pythonarguments(self, data, **_): # pylint: disable=no-self-use
"""
Convert parsed JSON back into a PythonArguments object.
"""
args = data["args"]
kwargs = data["kwargs"]
return PythonArguments(args, kwargs)
[docs]class EmbeddedScriptSchema(Schema):
"""
Schema for an EmbeddedScript, used in the activities section of an SKA scheduling block
"""
default_args = fields.Dict(fields.String(), fields.Nested(PythonArgumentsSchema))
content = fields.String()
[docs] @post_load
def make_embeddedscript(self, data, **_): # pylint: disable=no-self-use
"""
Convert parsed JSON back into a EmbeddedScript object.
"""
default_args = data["default_args"]
content = data["content"]
return EmbeddedScript(content, default_args)
[docs]class FileSystemScriptSchema(Schema):
"""
Schema for a FileSystemScript, used in the activities section of an SKA scheduling block
"""
default_args = fields.Dict(fields.String(), fields.Nested(PythonArgumentsSchema))
path = fields.String()
[docs] @post_load
def make_filesystemscript(self, data, **_): # pylint: disable=no-self-use
"""
Convert parsed JSON back into a FileSystemScript object.
"""
default_args = data["default_args"]
path = data["path"]
return FileSystemScript(path, default_args)
[docs]class GitScriptSchema(Schema):
"""
Schema for a GitScript, used in the activities section of an SKA scheduling block
"""
default_args = fields.Dict(fields.String(), fields.Nested(PythonArgumentsSchema))
path = fields.String()
repo = fields.String()
branch = fields.String()
commit = fields.String()
[docs] @post_load
def make_gitscript(self, data, **_): # pylint: disable=no-self-use
"""
Convert parsed JSON back into a GitScript object.
"""
return GitScript(**data)
[docs] @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}
[docs]class PythonProcedureSchema(OneOfSchema):
"""
Schema for an abstract PythonProcedure, used in the activities
section of anSKA scheduling block
"""
type_field = "procedure_type"
type_schemas = {
"embeddedscript": EmbeddedScriptSchema,
"filesystemscript": FileSystemScriptSchema,
"gitscript": GitScriptSchema,
}
def get_obj_type(self, obj):
if isinstance(obj, EmbeddedScript):
return "embeddedscript"
if isinstance(obj, GitScript):
return "gitscript"
if isinstance(obj, FileSystemScript):
return "filesystemscript"
raise Exception(f"Unknown object type: {obj.__class__.__name__}")