Source code for jina.jaml.parsers.base

from typing import TYPE_CHECKING, Any, Dict, Optional, Union

    from jina.orchestrate.flow.base import Flow
    from jina.serve.executors import BaseExecutor

[docs]class VersionedYAMLParser: """Flow YAML parser for specific version Every :class:`VersionedYAMLParser` must implement two methods and one class attribute: - :meth:`parse`: to load data (in :class:`dict`) into a :class:`BaseFlow` or :class:`BaseExecutor` object - :meth:`dump`: to dump a :class:`BaseFlow` or :class:`BaseExecutor` object into a :class:`dict` - :attr:`version`: version number in :class:`str` in format ``MAJOR.[MINOR]`` """ version = 'legacy' #: the version number this parser designed for
[docs] def parse( self, cls: type, data: Dict, runtime_args: Optional[Dict[str, Any]] ) -> Union['Flow', 'BaseExecutor']: """Return the Flow YAML parser given the syntax version number .. # noqa: DAR401 :param cls: target class type to parse into, must be a :class:`JAMLCompatible` type :param data: flow yaml file loaded as python dict :param runtime_args: Optional runtime_args to be directly passed without being parsed into a yaml config """ raise NotImplementedError
[docs] def dump(self, data: Union['Flow', 'BaseExecutor']) -> Dict: """Return the dictionary given a versioned flow object .. # noqa: DAR401 :param data: versioned flow object """ raise NotImplementedError