sync

src/compose/cfg/factory.py

@@ -1,19 +1,30 @@
+from pathlib import Path
 from typing import cast
 
-from compose.cfg.entity import CfgData, CfgDataYaml
+from compose.cfg.entity import CfgData, CfgDataYaml, OrgDataYaml
 from compose.cfg.get import cfg_get_orgs
 from compose.src_path.entity import SrcPaths
 from compose.src_path.get import src_path_get_services, src_path_get_volumes
-from compose.util import read_yml
+from compose.util import read_yml, validate_typed_dict
+
+
+def cfg_data_yml_factory(file: Path) -> CfgDataYaml:
+    data = cast(CfgDataYaml, read_yml(file))
+    validate_typed_dict(CfgDataYaml, data, file)
+
+    orgs_key = "orgs"
+    for org in data[orgs_key]:
+        validate_typed_dict(OrgDataYaml, org, file, (orgs_key,))
+    return data
 
 
 def cfg_data_factory(src_paths: SrcPaths) -> CfgData:
-    data = cast(CfgDataYaml, read_yml(src_paths.cfg_file))
+    data = cfg_data_yml_factory(src_paths.cfg_file)
     vols = frozenset(src_path_get_volumes(src_paths, data))
     return CfgData(
         src_paths,
         src_paths.cfg_dir.name,
         frozenset(src_path_get_services(src_paths, data)),
         vols if vols else None,
-        frozenset(cfg_get_orgs(data)),
+        frozenset(cfg_get_orgs(data, src_paths.cfg_file)),
     )

src/compose/cfg/get.py

@@ -1,29 +1,42 @@
 from collections.abc import Iterator
+from pathlib import Path
 from typing import cast
 
 from compose.cfg.entity import CfgData, CfgDataYaml, OrgData
 from compose.compose.entity import VolYaml
-from compose.service.entity import Service, ServiceYaml
+from compose.service.entity import Service
+from compose.service.factory import services_yaml_factory
 from compose.util import read_yml
 
 
-def cfg_get_orgs(data: CfgDataYaml) -> Iterator[OrgData]:
+def cfg_get_orgs(data: CfgDataYaml, path: Path) -> Iterator[OrgData]:
     # orgs = data.get("orgs")
     # if orgs is None:
     #     yield OrgData(
     #         org_data.get("org"),
     #         org_data.get("url"),
     #     )
-    for org_data in data["orgs"]:
-        yield OrgData(
-            org_data["org"],
-            org_data.get("url"),
-        )
+    orgs = "orgs"
+    try:
+        orgs = data[orgs]
+    except KeyError as e:
+        print(f'key "{orgs}" not in "{path!s}"')
+        raise KeyError(e)
+    org = "org"
+    for org_data in orgs:
+        try:
+            yield OrgData(
+                org_data[org],
+                org_data.get("url"),
+            )
+        except KeyError as e:
+            print(f'key "{orgs}.{org}" not in "{path!s}"')
+            raise KeyError(e)
 
 
 def cfg_get_services(cfg_data: CfgData) -> Iterator[tuple[str, Service]]:
     for path in cfg_data.services:
-        _dict = cast(ServiceYaml, read_yml(path))
+        _dict = services_yaml_factory(path)
         yield path.stem, Service.from_dict(_dict)
 
 

src/compose/service/entity.py

@@ -24,11 +24,11 @@ class ServiceYamlAbc[T_net: T_NetAbc](TypedDict):
     volumes: NotRequired[list[str]]
 
 
-TCo_ServiceYaml = TypeVar(
-    "TCo_ServiceYaml",
-    bound=ServiceYamlAbc[T_NetAbc],
-    covariant=True,
-)
+# TCo_ServiceYaml = TypeVar(
+#     "TCo_ServiceYaml",
+#     bound=ServiceYamlAbc[T_NetAbc],
+#     covariant=True,
+# )
 
 
 class ServiceYaml(ServiceYamlAbc[Literal["proxy", "internal"]]):

src/compose/service/factory.py

@@ -1,10 +1,12 @@
-# from typing import cast
+from pathlib import Path
+from typing import cast
 
-# from compose.cfg import TRAEFIK_PATH
-# from compose.service.entity import TraefikServiceYaml
-# from compose.util import read_yml
+from compose.service.entity import ServiceYaml
+from compose.util import read_yml, validate_typed_dict
 
 
-# def get_traefik_service():
-#     path = TRAEFIK_PATH.joinpath("service.yml")
-#     return cast(TraefikServiceYaml, read_yml(path))
+def services_yaml_factory(path: Path):
+    data = cast(ServiceYaml, read_yml(path))
+    # data = read_yml(path)
+    validate_typed_dict(ServiceYaml, data)
+    return data

src/compose/util.py

@@ -1,7 +1,7 @@
 import re
 from collections.abc import Mapping
 from pathlib import Path
-from typing import Any, cast, override
+from typing import Any, ClassVar, KeysView, Protocol, cast, override
 
 import yaml
 
@@ -40,6 +40,7 @@ def read_yml(path: Path) -> T_YamlDict:
     with path.open("rt") as f:
         return cast(T_YamlDict, yaml.safe_load(f))
 
+
 def to_yaml(data: T_YamlDict) -> str:
     _yaml = yaml.dump(data, Dumper=VerboseSafeDumper)
     return re.sub(r"(^\s*-)", r"  \g<1>", _yaml, flags=re.MULTILINE)
@@ -47,3 +48,30 @@ def to_yaml(data: T_YamlDict) -> str:
 
 def get_replace_name(name: str) -> str:
     return f"${{_{name.upper()}}}"
+
+
+class T_TypedDict(Protocol):
+    __required_keys__: ClassVar[frozenset[str]]
+
+    def keys(self) -> KeysView[str]: ...
+
+
+def validate_typed_dict(
+    typed_dict: type[T_TypedDict],
+    data: T_TypedDict,
+    path: Path | None = None,
+    pre: tuple[str, ...] | None = None,
+) -> None:
+    req = typed_dict.__required_keys__.difference(data.keys())
+    if not req:
+        return
+    if pre is None:
+        keys = (f'"{key}"' for key in req)
+    else:
+        key_pre = ".".join(pre)
+        keys = (f'"{key_pre}.{key}"' for key in req)
+    msg = f"key(s) ({', '.join(keys)}) not found"
+    if path is not None:
+        msg = f"{msg} in file {path!s}"
+    print(msg)
+    raise KeyError

src/compose/volumes/entity.py

@@ -1 +0,0 @@
-