Source code for jina.jaml.parsers.default.v1
from typing import Any, Dict, Optional, Type
from jina.jaml import JAML, JAMLCompatible
from jina.jaml.parsers.base import VersionedYAMLParser
[docs]class V1Parser(VersionedYAMLParser):
"""V1 default parser, used as the default parser for objects."""
version = '1' # the version number this parser designed for
[docs] def parse(
self,
cls: Type['JAMLCompatible'],
data: Dict,
runtime_args: Optional[Dict[str, Any]] = None,
) -> 'JAMLCompatible':
"""
: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
:return: the YAML parser given the syntax version number
"""
expanded_data = JAML.expand_dict(data, None)
if 'with' in data:
obj = cls(**expanded_data.get('with', {}))
else:
obj = cls(**expanded_data)
return obj
[docs] def dump(self, data: 'JAMLCompatible') -> Dict:
"""
:param data: versioned flow object
:return: the dictionary given a versioned flow object
"""
a = V1Parser._dump_instance_to_yaml(data)
r = {}
if a:
r['with'] = a
return r
@staticmethod
def _dump_instance_to_yaml(instance):
import inspect
attributes = inspect.getmembers(instance, lambda a: not (inspect.isroutine(a)))
return {
a[0]: a[1]
for a in attributes
if not (a[0].startswith('__') and a[0].endswith('__'))
}