1from __future__ import annotations
2
3from typing import Any
4
5import daggerml.api as codec_mod
6from daggerml.contrib import adapters as areg
7from daggerml.contrib.executors import _base as ereg
8
9
10def _diag(*, scope: str, code: str, message: str) -> dict[str, str]:
11 return {
12 "severity": "error",
13 "scope": scope,
14 "code": code,
15 "message": message,
16 }
17
18
19def _fqn(obj: Any) -> str:
20 module = getattr(obj, "__module__", None) or type(obj).__module__
21 qualname = getattr(obj, "__qualname__", None) or type(obj).__qualname__
22 return f"{module}:{qualname}"
23
24
25def _implements(obj: Any, names: tuple[str, ...]) -> dict[str, bool]:
26 return {name: callable(getattr(obj, name, None)) for name in names}
27
28
29def _registration(
30 kind: str, key: str, obj: Any, diagnostics: list[dict[str, Any]] | None = None
31) -> dict[str, Any]:
32 required = {
33 "adapter": ("resolve_runnable", "send", "cli"),
34 "executor": ("resolve_runnable", "start", "poll", "cleanup", "cancel"),
35 "codec": ("can_encode", "encode"),
36 }[kind]
37 implements = _implements(obj, required)
38 missing = [name for name, implemented in implements.items() if not implemented]
39 if missing and diagnostics is not None:
40 diagnostics.append(
41 _diag(
42 scope=kind,
43 code="required_operation_missing",
44 message=f"{key} is missing required operations: {', '.join(missing)}",
45 )
46 )
47 return {
48 "key": key,
49 "fqn": _fqn(obj),
50 "effective": not missing,
51 "implements": implements,
52 }
53
54
55def _load_plugins(scope: str, load: Any, diagnostics: list[dict[str, Any]]) -> None:
56 try:
57 load()
58 except Exception as e:
59 diagnostics.append(_diag(scope=scope, code="plugin_load_failed", message=str(e)))
60
61
62def _adapter_status(diagnostics: list[dict[str, Any]]) -> list[dict[str, Any]]:
63 _load_plugins("adapter", areg.load_adapter_plugins, diagnostics)
64 with areg._LOCK:
65 items = sorted(areg._ADAPTER_SPECS.items())
66 return [_registration("adapter", name, spec, diagnostics) for name, spec in items]
67
68
69def _executor_status(diagnostics: list[dict[str, Any]]) -> list[dict[str, Any]]:
70 _load_plugins("executor", ereg.load_executor_plugins, diagnostics)
71 with ereg._LOCK:
72 items = sorted(ereg._EXECUTOR_SPECS.items())
73 return [_registration("executor", f"{adapter}:{name}", spec, diagnostics) for (adapter, name), spec in items]
74
75
76def _codec_status(diagnostics: list[dict[str, Any]]) -> list[dict[str, Any]]:
77 try:
78 codecs = list(codec_mod.iter_codecs())
79 except Exception as e:
80 diagnostics.append(_diag(scope="codec", code="plugin_load_failed", message=str(e)))
81 codecs = []
82 return [_registration("codec", str(order), codec, diagnostics) for order, codec in enumerate(codecs)]
83
84
85def status() -> dict[str, object]:
86 diagnostics: list[dict[str, Any]] = []
87 adapters = _adapter_status(diagnostics)
88 executors = _executor_status(diagnostics)
89 codecs = _codec_status(diagnostics)
90 diagnostics.sort(key=lambda item: (item["scope"], item["code"], item["message"]))
91
92 return {
93 "schema_version": 0,
94 "summary": {
95 "has_errors": bool(diagnostics),
96 "diagnostic_count": len(diagnostics),
97 "adapter_registration_count": len(adapters),
98 "adapter_effective_count": sum(item["effective"] for item in adapters),
99 "executor_registration_count": len(executors),
100 "executor_effective_count": sum(item["effective"] for item in executors),
101 "codec_registration_count": len(codecs),
102 "codec_effective_count": sum(item["effective"] for item in codecs),
103 },
104 "adapters": adapters,
105 "executors": executors,
106 "codecs": codecs,
107 "diagnostics": diagnostics,
108 }