-
Notifications
You must be signed in to change notification settings - Fork 2
Expand file tree
/
Copy pathreference_loader.py
More file actions
54 lines (45 loc) · 1.42 KB
/
reference_loader.py
File metadata and controls
54 lines (45 loc) · 1.42 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
from __future__ import annotations
from pathlib import Path
from typing import Any, Dict
from Implementations.Reference.common import (
DEMO_ARTIFACT_VERSION,
FrogPipelineError,
LoadedSource,
load_json_file,
sha256_text,
)
def _normalize_path(path: Path) -> str:
return str(path).replace("\\", "/")
def load_source(file_path: str) -> LoadedSource:
path = Path(file_path)
try:
text, document = load_json_file(path)
except FrogPipelineError as exc:
artifact = {
"artifact_kind": "frog_loaded_source",
"artifact_version": DEMO_ARTIFACT_VERSION,
"stage_ref": "load",
"status": "load_failed",
"source": {
"path": _normalize_path(path),
"content_hash": None,
"spec_version": None,
},
"document": None,
"diagnostics": [exc.as_dict()],
}
return LoadedSource(artifact=artifact)
artifact: Dict[str, Any] = {
"artifact_kind": "frog_loaded_source",
"artifact_version": DEMO_ARTIFACT_VERSION,
"stage_ref": "load",
"status": "ok",
"source": {
"path": _normalize_path(path),
"content_hash": sha256_text(text),
"spec_version": document.get("spec_version"),
},
"document": document,
"diagnostics": [],
}
return LoadedSource(artifact=artifact)