added helperf unctions for kadi

cbutil/kadi_helper.py

+from dateutil import parser
+from datetime import datetime
+from cbutil.util import remove_none
+import logging
+import builtins
+logger = logging.getLogger(__file__)
+
+
+def create_metadatum(key: str, type_name: str, value, **kwargs) -> dict:
+    """ Create a dict so that it can be used for kadiapy is add_metadatum."""
+    if isinstance(type_name, type):
+        type_name = type_name.__name__
+    return {"key": key,
+            "type": type_name,
+            "value": value,
+            **kwargs}
+
+
+def get_type_name(value) -> str:
+    if not isinstance(value, str):
+        return type(value).__name__
+    for possible_type in [int, float]:
+        try:
+            possible_type(value)
+        except ValueError:
+            pass
+        else:
+            return possible_type.__name__
+
+    try:
+        parser.parse(value)
+    except ValueError:
+        logger.info(f"{value} could not be parse as date")
+        pass
+    else:
+        return "date"
+
+    return str.__name__
+
+
+def handle_dates(value) -> str:
+    try:
+        return str(datetime.fromtimestamp(int(value)))
+    except (ValueError, OverflowError, OSError):
+        pass
+    return value
+
+
+def transform_to_type(type_name, value: str):
+    try:
+        return getattr(builtins, type_name)(value)
+    except AttributeError as e:
+        raise ValueError(f"could not transform {value} to {type_name} {e}")
+
+
+def build_meta_data(tags: dict):
+    for key, value in remove_none(tags).items():
+        type_name = get_type_name(value)
+        if key == "time" or type_name == "date":
+            type_name = "date"
+            value = handle_dates(value)
+        else:
+            value = transform_to_type(type_name, value)
+        yield create_metadatum(key, type_name, value)

tests/test_kadi_helper.py

+from cbutil.kadi_helper import create_metadatum, get_type_name, handle_dates, build_meta_data
+from datetime import datetime
+from dateutil import parser
+
+
+def test_create_metadatum():
+    key = "test_number"
+    type_name = int.__name__
+    value = 10
+    meta_datum = create_metadatum(key, type_name, value)
+    assert meta_datum["key"] == key
+    assert meta_datum["type"] == type_name
+    assert meta_datum["value"] == value
+
+
+def test_get_type_name_str():
+    assert get_type_name("Test String") == 'str'
+
+
+def test_get_type_name_int():
+    assert get_type_name("111") == 'int'
+    assert get_type_name(111) == 'int'
+
+
+def test_get_type_name_float():
+    assert get_type_name("1.0") == 'float'
+    assert get_type_name("1e-8") == 'float'
+    assert get_type_name(1e-8) == 'float'
+
+
+def test_get_type_name_date():
+    assert get_type_name("2007-08-31T16:47+00:00") == "date"
+    assert get_type_name("2009-01-01T12:00:00+01:00") == "date"
+    assert get_type_name(str(datetime.fromtimestamp(1678457781))) == "date"
+
+
+def test_handle_dates():
+    assert parser.parse(handle_dates(1678457781))
+    assert parser.parse(handle_dates(str(datetime.fromtimestamp(1678457781))))
+
+
+def test_build_meta_data():
+    time_stamp = 1678457781
+    pipeline_infos = {
+        'branch': 'devel/cb',
+        'commit': 'c36f8a71e798bcdb87dbfea52b93f35cdc97ab5c',
+        'commit_message': 'set micro problem size from the outside and also have have node jobs',
+        'host': 'skylakesp2',
+        'project_id': '614',
+        'time': time_stamp
+    }
+    actual = list(build_meta_data(pipeline_infos))
+    for meta_datum in actual:
+        if meta_datum["key"] == "time":
+            assert meta_datum["type"] == "date"
+            assert parser.parse(meta_datum["value"])
+        if meta_datum["type"] == 'int':
+            int(meta_datum["value"])
+        if meta_datum["type"] == 'float':
+            int(meta_datum["value"])
+        if meta_datum["type"] == 'str':
+            assert meta_datum["value"] == pipeline_infos[meta_datum["key"]]