ccamper7/compose_gen
SHA256
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

working

Browse Source
This commit is contained in:
Christian Camper
2026-01-10 14:26:03 -06:00
parent 377e481803
commit 0ef6530f92
9 changed files with 156 additions and 243 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.idea/dictionaries/project.xml generated
View File

@@ -1,9 +1,11 @@
<component name="ProjectDictionaryState"> <component name="ProjectDictionaryState">
<dictionary name="project"> <dictionary name="project">
<words> <words>
<w>Pswd</w>
<w>ccamper</w> <w>ccamper</w>
<w>certresolver</w> <w>certresolver</w>
<w>exts</w> <w>exts</w>
<w>pswd</w>
<w>stryten</w> <w>stryten</w>
<w>traefik</w> <w>traefik</w>
<w>websecure</w> <w>websecure</w>

32
src/docker_compose/cfg/compose_paths.py
View File

@@ -1,43 +1,42 @@
from collections.abc import Callable, Iterator, MutableMapping from collections.abc import Callable, Iterator, MutableMapping
from dataclasses import dataclass, field from dataclasses import dataclass, field
from pathlib import Path from pathlib import Path
from typing import ClassVar, cast, final from typing import cast, final
import yaml import yaml
from docker_compose.cfg.org import OrgData from docker_compose.cfg.org import App, Org
from docker_compose.cfg.replace import ReplaceDynamic, RecordCls from docker_compose.cfg.replace import ReplaceDynamic, ReplaceStatic, ReplaceUnique
from docker_compose.compose.services_yaml import ServiceYamlRead from docker_compose.compose.services_yaml import ServiceYamlRead
from docker_compose.compose.volume_yaml import VolYaml from docker_compose.compose.volume_yaml import VolYaml
from docker_compose.util.Ts import T_YamlRW from docker_compose.util.Ts import T_YamlRW
from docker_compose.util.yaml_util import path_to_typed, read_yaml from docker_compose.util.yaml_util import path_to_typed, read_yaml
# _SERVICE = "service"
@final @final
@dataclass(frozen=True, slots=True) @dataclass(frozen=True, slots=True)
class ServiceVal(RecordCls): class ServiceVal(ReplaceStatic):
rep: ClassVar[str] = "service" rep = ReplaceDynamic("service")
@final @final
@dataclass(frozen=True, slots=True) @dataclass(frozen=True, slots=True)
class ServicePath: class ServicePath:
path: Path path: Path
fqdn: ReplaceDynamic = field(init=False) fqdn: ReplaceUnique = field(init=False)
replace_pre: ReplaceDynamic = field(init=False) replace_pre: ReplaceStatic = field(init=False)
replace_post: ReplaceDynamic = field(init=False) replace_post: ReplaceStatic = field(init=False)
def __post_init__(self): def __post_init__(self):
setter = super(ServicePath, self).__setattr__ setter = super(ServicePath, self).__setattr__
pre, post = ServiceVal.two_stage(self.path.stem) pre, post = ServiceVal.two_stage(self.path.stem)
setter("replace_pre", pre) setter("replace_pre", pre)
setter("replace_post", post) setter("replace_post", post)
setter( setter(
"fqdn", "fqdn",
ReplaceDynamic( ReplaceUnique.auto_format(
"fqdn", "fqdn", str(Org.src.build_placeholder(App.src, pre.src))
f"{OrgData.org_app!s}_{pre!s}",
), ),
) )
@@ -51,7 +50,7 @@ class ServicePath:
if not isinstance(data_dict, MutableMapping): if not isinstance(data_dict, MutableMapping):
raise TypeError raise TypeError
path_to_typed(ServiceYamlRead, data_dict, self.path) path_to_typed(ServiceYamlRead, data_dict, self.path)
return data_dict # pyright: ignore[reportReturnType] return cast(ServiceYamlRead, cast(object, data_dict))
@property @property
def pre_render_funcs(self) -> Iterator[Callable[[str], str]]: def pre_render_funcs(self) -> Iterator[Callable[[str], str]]:
@@ -87,13 +86,6 @@ class ComposePaths:
for service in self.services: for service in self.services:
yield service.replace_post yield service.replace_post
# @classmethod
# def from_iters(cls, services: Iterable[ServicePath], volumes: Iterable[VolumePath]):
# return cls(
# frozenset(services),
# frozenset(volumes),
# )
@property @property
def volumes_k_v(self) -> Iterator[tuple[str, VolYaml]]: def volumes_k_v(self) -> Iterator[tuple[str, VolYaml]]:
for path in self.volumes: for path in self.volumes:

8
src/docker_compose/cfg/dest_path.py
View File

@@ -6,7 +6,7 @@ from typing import Self, final
from docker_compose.cfg import DATA_ROOT from docker_compose.cfg import DATA_ROOT
from docker_compose.cfg.org import App, Org, OrgData from docker_compose.cfg.org import App, Org, OrgData
from docker_compose.cfg.replace import ReplaceDynamic from docker_compose.cfg.replace import ReplaceUnique
@final @final
@@ -44,10 +44,10 @@ class ComposeFileRendered:
@dataclass(frozen=True, slots=True) @dataclass(frozen=True, slots=True)
class DestPaths: class DestPaths:
# data_root = Record("data_root", str(DATA_ROOT)) # data_root = Record("data_root", str(DATA_ROOT))
data_root = ReplaceDynamic.auto_format("data", str(DATA_ROOT)) data_root = ReplaceUnique.auto_format("data", str(DATA_ROOT))
data_path = ReplaceDynamic( data_path = ReplaceUnique(
data_root.src, data_root.src,
sep.join((data_root.src, Org.rep, App.rep)), sep.join((data_root.src, str(Org), str(App))),
) )
data_dir: Path data_dir: Path
env_file: Path = field(init=False) env_file: Path = field(init=False)

27
src/docker_compose/cfg/env.py
View File

@@ -6,16 +6,19 @@ from functools import partial
from pathlib import Path from pathlib import Path
from typing import Self, final from typing import Self, final
from docker_compose.cfg.replace import ReplaceDynamic from docker_compose.cfg.replace import ReplaceDynamic, ReplaceStatic
@final
@dataclass(frozen=True, slots=True)
class Pswd(ReplaceStatic):
src = ReplaceDynamic("pswd")
@final @final
@dataclass @dataclass
class Env: class Env:
pswd = ReplaceDynamic.auto_format( pswd = Pswd(partial(secrets.token_urlsafe, 12))
"pswd",
partial(secrets.token_urlsafe, 12),
)
data: dict[str, str] data: dict[str, str]
@classmethod @classmethod
@@ -34,17 +37,17 @@ class Env:
return cls({k: v for k, v in cls.get_lines(data)}) return cls({k: v for k, v in cls.get_lines(data)})
@property @property
def with_pass(self)->Iterator[tuple[str,str]]: def with_pass(self) -> Iterator[tuple[str, str]]:
for k,v in self.data.items(): p = self.pswd
if self.pswd.src not in v: for k, v in self.data.items():
yield k,v if self.pswd.src.fmt not in v:
yield k, v
continue continue
yield k,self.pswd(v) yield k, p(v)
@property @property
def as_txt(self) -> str: def as_txt(self) -> str:
return '\n'.join(sorted(map('='.join, self.with_pass))) return "\n".join(sorted(map("=".join, self.with_pass)))
@classmethod @classmethod
def copy(cls, src: Path, dest: Path) -> None: def copy(cls, src: Path, dest: Path) -> None:

106
src/docker_compose/cfg/org.py
View File

@@ -1,107 +1,40 @@
from collections.abc import Iterator from collections.abc import Iterator
from dataclasses import dataclass from dataclasses import dataclass
from typing import Callable, ClassVar, Self, final, override from typing import Callable, Self, final, override
from docker_compose.cfg.org_yaml import OrgDataYaml from docker_compose.cfg.org_yaml import OrgDataYaml
from docker_compose.cfg.replace import ReplaceDynamic, RecordCls from docker_compose.cfg.replace import ReplaceDynamic, ReplaceStatic
#
# _ORG = "org"
# _APP = "name"
# Org = partial(Record, ORG)
# App = partial(Record, APP)
@final @final
@dataclass(frozen=True, slots=True) @dataclass(frozen=True, slots=True)
class Org(RecordCls): class Org(ReplaceStatic):
rep: ClassVar[str] = "org" src = ReplaceDynamic("org")
@final @final
@dataclass(frozen=True, slots=True) @dataclass(frozen=True, slots=True)
class App(RecordCls): class App(ReplaceStatic):
rep: ClassVar[str] = "name" src = ReplaceDynamic("name")
@final @final
@dataclass(frozen=True, slots=True) @dataclass(frozen=True, slots=True)
class Url(RecordCls): class Url(ReplaceStatic):
rep: ClassVar[str] = "url" src = ReplaceDynamic("url")
@override @override
@classmethod def __str__(self) -> str:
def from_str(cls, string: str | None) -> Self: val = super(Url, self).__str__()
return super(Url, cls).from_str(".".join((string, "ccamper7", "net")) if string else "") if not val:
return val
# return Record("url", ".".join((val, "ccamper7", "net")) if val else "") return ".".join((val, "ccamper7", "net"))
#
# @final
# @dataclass(frozen=True, slots=True)
# class Org:
# val: str
# replace: Record = field(init=False)
#
# def __post_init__(self) -> None:
# setter = super(Org, self).__setattr__
# setter("replace", OrgVal(self.val))
# @final
# @dataclass(frozen=True, slots=True)
# class AppVal(RecordCls[str]):
# old: ClassVar[RecordName] = RecordName("name")
#
#
# @final
# @dataclass(frozen=True, slots=True)
# class App:
# val: str
# replace: AppVal = field(init=False)
#
# def __post_init__(self) -> None:
# setter = super(App, self).__setattr__
# setter("replace", AppVal(self.val))
# @final
# @dataclass(frozen=True, slots=True)
# class UrlValNew:
# val: str | None
#
# @override
# def __str__(self) -> str:
# if not self.val:
# return ""
# return ".".join((self.val, "ccamper7", "net"))
#
#
# @final
# @dataclass(frozen=True, slots=True)
# class UrlVal(RecordCls[UrlValNew]):
# old = RecordName("url")
#
#
# @final
# @dataclass(frozen=True, slots=True)
# class Url:
# val: str | None
# replace: UrlVal = field(init=False)
#
# def __post_init__(self) -> None:
# setter = super(Url, self).__setattr__
# setter("replace", UrlVal(UrlValNew(self.val)))
@final @final
@dataclass(frozen=True, slots=True) @dataclass(frozen=True, slots=True)
class OrgData: class OrgData:
org_app = ReplaceDynamic( org_app = Org.src.build_placeholder(App.src)
f"${Org.rep.upper()}_{Org.rep.upper()}",
f"{ReplaceDynamic.get_format(Org.rep)}_{ReplaceDynamic.get_format(App.rep)}",
)
app: App app: App
org: Org org: Org
url: Url url: Url
@@ -109,9 +42,9 @@ class OrgData:
@classmethod @classmethod
def from_dict(cls, app: str, org: str, data: OrgDataYaml) -> Self: def from_dict(cls, app: str, org: str, data: OrgDataYaml) -> Self:
return cls( return cls(
App.from_str(app), App(app),
Org.from_str(org), Org(org),
Url.from_str(data.get("url")), Url(data.get("url")),
) )
@property @property
@@ -123,8 +56,3 @@ class OrgData:
@property @property
def pre_render_funcs(self) -> Iterator[Callable[[str], str]]: def pre_render_funcs(self) -> Iterator[Callable[[str], str]]:
yield self.org_app yield self.org_app
# def render_yaml(self, yaml: str) -> str:
# for func in self.render_funcs:
# yaml = func(yaml)
# return yaml

120
src/docker_compose/cfg/replace.py
View File

@@ -1,87 +1,75 @@
from collections.abc import Callable from collections.abc import Callable
from dataclasses import dataclass, field from dataclasses import dataclass, field
from typing import ClassVar, Self, override, final from itertools import chain
from typing import ClassVar, Self, final, override
# def format_src(src: str) -> str:
# class String(Protocol): return f"${{_{src.upper()}}}"
# @override
# def __str__(self) -> str: ...
# def replace(self, string: str) -> str:
# return string.replace(str(self), str(self))
# @dataclass(frozen=True, slots=True)
# class RecordCls[T_New: str | Callable[[None],str]]:
# old: ClassVar[str]
# new: T_New
#
# def __call__(self, string: str) -> str:
# return string.replace(str(self.old), self.new if isinstance(self.new, str) else self.new())
@final @final
@dataclass(frozen=True, slots=True) @dataclass(frozen=True, slots=True)
class ReplaceStatic: class ReplaceUnique:
val: 'str| ReplaceStatic' src: str
dest: str # | Callable[[], str]
def __call__(self, string: str) -> str:
return string.replace(self.src, self.dest)
@classmethod
def auto_format(cls, src: str, dest: str):
return cls(format_src(src), dest)
# @override
# def __str__(self) -> str:
# if not self.dest:
# return ""
# if isinstance(self.dest, str):
# return self.dest
# return self.dest()
@final
@dataclass(frozen=True, slots=True)
class ReplaceDynamic:
val: str
fmt: str = field(init=False) fmt: str = field(init=False)
def __post_init__(self): def __post_init__(self):
setter = super(ReplaceStatic, self).__setattr__ setter = super(ReplaceDynamic, self).__setattr__
setter('fmt', f"${{_{self.val.upper()}}}") setter("fmt", format_src(self.val))
def __call__(self, string: str) -> str: def __call__(self, string: str) -> str:
return string.replace( return string.replace(self.fmt, self.val)
self.fmt,
str(self)
)
def __str__(self) -> str:
return self.val if isinstance(self.val, str) else self.val.fmt
@dataclass(frozen=True, slots=True) # def __str__(self) -> str:
class ReplaceDynamic: # return self.val if isinstance(self.val, str) else self.val.fmt
src: ReplaceStatic def build_placeholder(self, *args: "ReplaceDynamic"):
dest: str | Callable[[], str] data = ((rep.val.upper(), rep.fmt) for rep in chain((self,), args))
src: tuple[str, ...]
def __call__(self, string: str) -> str: dest: tuple[str, ...]
return string.replace( src, dest = zip(*data)
self.src, return ReplaceUnique("_".join(src), "_".join(dest))
str(self),
)
@override
def __str__(self) -> str:
return self.dest if isinstance(self.dest, str) else self.dest()
#
# @staticmethod
# def get_format(val: str) -> str:
# return f"${{_{val.upper()}}}"
@classmethod
def auto_format(cls, src: str, dest: str | Callable[[], str]) -> Self:
return cls(ReplaceStatic(src), dest)
@classmethod
def from_str(cls, string: str) -> Self:
return cls.auto_format(string, string)
# @property
# def stage_two(self) -> 'Record':
# return Record.from_str(self.dest if isinstance(self.dest, str) else self.dest())
@dataclass(frozen=True, slots=True) @dataclass(frozen=True, slots=True)
class RecordCls(ReplaceDynamic): class ReplaceStatic:
# val: ClassVar[str]# = 'org' src: ClassVar[ReplaceDynamic]
rep: ClassVar[ReplaceStatic] # = Record.get_format(val) dest: None | str | Callable[[], str]
def __call__(self, string: str) -> str:
return string.replace(self.src.fmt, str(self))
@override @override
@classmethod def __str__(self) -> str:
def from_str(cls, string: str) -> Self: if not self.dest:
return cls(cls.rep, string) return ""
if isinstance(self.dest, str):
return self.dest
return self.dest()
@classmethod @classmethod
def two_stage(cls, dest: str) -> tuple[Self, Self]: def two_stage(cls, dest: str) -> tuple[Self, ReplaceDynamic]:
dest_var = ReplaceStatic(dest) dest_var = ReplaceDynamic(dest)
return cls(cls.rep, dest_var.fmt), cls(dest_var, dest) return cls(dest_var.fmt), dest_var

8
src/docker_compose/compose/services.py
View File

@@ -3,7 +3,7 @@ from typing import Self, final, override
from docker_compose.cfg.compose_paths import ServicePath from docker_compose.cfg.compose_paths import ServicePath
from docker_compose.cfg.org import OrgData, Url from docker_compose.cfg.org import OrgData, Url
from docker_compose.cfg.replace import ReplaceDynamic from docker_compose.cfg.replace import ReplaceStatic, ReplaceUnique
from docker_compose.compose.services_yaml import ( from docker_compose.compose.services_yaml import (
HealthCheck, HealthCheck,
ServiceYamlRead, ServiceYamlRead,
@@ -17,7 +17,7 @@ from docker_compose.util.Ts import T_Primitive
class Service: class Service:
_traefik_labels = frozenset( _traefik_labels = frozenset(
( (
f"traefik.http.routers.{OrgData.org_app.src}.rule=Host(`{Url.rep}`)", f"traefik.http.routers.{OrgData.org_app!s}.rule=Host(`{Url!s}`)",
f"traefik.http.routers.{OrgData.org_app.src}.entrypoints=websecure", f"traefik.http.routers.{OrgData.org_app.src}.entrypoints=websecure",
f"traefik.docker.network={OrgData.org_app.src}_proxy", f"traefik.docker.network={OrgData.org_app.src}_proxy",
f"traefik.http.routers.{OrgData.org_app.src}.tls.certresolver=le", f"traefik.http.routers.{OrgData.org_app.src}.tls.certresolver=le",
@@ -35,7 +35,7 @@ class Service:
_sec_opts = frozenset(("no-new-privileges:true",)) _sec_opts = frozenset(("no-new-privileges:true",))
# service_name: str # service_name: str
# service_val: ServiceVal # service_val: ServiceVal
fqdn: ReplaceDynamic fqdn: ReplaceUnique
command: tuple[str, ...] command: tuple[str, ...]
entrypoint: tuple[str, ...] entrypoint: tuple[str, ...]
environment: dict[str, T_Primitive] environment: dict[str, T_Primitive]
@@ -57,7 +57,7 @@ class Service:
return cls.from_dict(path.fqdn, path.as_dict) return cls.from_dict(path.fqdn, path.as_dict)
@classmethod @classmethod
def from_dict(cls, fqdn: ReplaceDynamic, data: ServiceYamlRead) -> Self: def from_dict(cls, fqdn: ReplaceStatic, data: ServiceYamlRead) -> Self:
# helper = ServiceYamlProps(data) # helper = ServiceYamlProps(data)
labels = frozenset(data.get("labels", ())) labels = frozenset(data.get("labels", ()))
# ports = (f'"{p}"' for p in data.get("ports", ())) # ports = (f'"{p}"' for p in data.get("ports", ()))

69
src/docker_compose/util/Ts.py
View File

@@ -2,26 +2,31 @@ from collections.abc import ItemsView, Iterator, KeysView, MutableMapping, Set
from types import GenericAlias, UnionType from types import GenericAlias, UnionType
from typing import ( from typing import (
ClassVar, ClassVar,
Never,
Protocol, Protocol,
TypeAliasType, TypeAliasType,
cast, cast,
get_args, get_args,
get_origin, get_origin,
overload,
) )
type T_Primitive = None | bool | int | str
class TypedYamlDict[K: object, V: object](Protocol): # type T_TDict = MutableMapping[T_Primitive, T_Prim]
def __getitem__(self, key: str | K, /) -> V: ... type T_TIters = list[T_Prim]
type T_T = T_Primitive | T_TIters | TypedYamlDict
class TypedYamlDict(Protocol):
def __getitem__(self, key: str, /) -> T_T: ...
# def __setitem__(self, key: str, value: V, /) -> V: ... # def __setitem__(self, key: str, value: V, /) -> V: ...
def __delitem__(self, key: Never | K, /) -> None: ... # def __delitem__(self, key: Never | K, /) -> None: ...
def __contains__(self, key: K, /) -> bool: ... def __contains__(self, key: str, /) -> bool: ...
def __iter__(self) -> Iterator[K]: ... def __iter__(self) -> Iterator[str]: ...
def __len__(self) -> int: ... def __len__(self) -> int: ...
def keys(self) -> KeysView[K]: ... def keys(self) -> KeysView[str]: ...
def items(self) -> ItemsView[K, V]: ... def items(self) -> ItemsView[str, T_T]: ...
def pop(self, key: Never | K, /) -> V: ...
# def pop(self, key: Never | K, /) -> V: ...
# def popitem(self) -> tuple[K, V]: ... # def popitem(self) -> tuple[K, V]: ...
@@ -45,7 +50,6 @@ class TypedYamlDict[K: object, V: object](Protocol):
# #
# is_typed_dict_test(x) # is_typed_dict_test(x)
type T_Primitive = None | bool | int | str
type T_PrimIters = tuple[T_Prim, ...] | list[T_Prim] | Set[T_Prim] | Iterator[T_Prim] type T_PrimIters = tuple[T_Prim, ...] | list[T_Prim] | Set[T_Prim] | Iterator[T_Prim]
type T_PrimDict = MutableMapping[T_Primitive, T_Prim] type T_PrimDict = MutableMapping[T_Primitive, T_Prim]
@@ -57,45 +61,40 @@ type T_YamlRW = T_YamlIters | T_YamlDict
type T_Yaml = T_Primitive | T_YamlRW type T_Yaml = T_Primitive | T_YamlRW
type T_YamlPostDict = TypedYamlDict[str, T_YamlPost] type T_YamlPostDict = MutableMapping[str, T_YamlPost]
type T_YamlPostRes = tuple[T_YamlPost, ...] | T_YamlPostDict type T_YamlPostRes = tuple[T_YamlPost, ...] | T_YamlPostDict
type T_YamlPost = T_Primitive | T_YamlPostRes type T_YamlPost = T_Primitive | T_YamlPostRes
def get_union_types(annotations: UnionType) -> Iterator[type]: def get_union_types(annotations: UnionType) -> Iterator[type]:
for annotation in get_args(annotations): # pyright: ignore[reportAny] for annotation in get_args(annotations): # pyright: ignore[reportAny]
annotation = cast(TypeAliasType | GenericAlias | UnionType | type, annotation)
if isinstance(annotation, TypeAliasType): if isinstance(annotation, TypeAliasType):
annotation = annotation.__value__ # pyright: ignore[reportAny] yield from get_types(
cast(GenericAlias | UnionType | type, annotation.__value__)
)
continue
if isinstance(annotation, UnionType): if isinstance(annotation, UnionType):
yield from get_union_types(annotation) yield from get_union_types(annotation)
continue continue
yield get_types(annotation) # pyright: ignore[reportAny] yield from get_types(annotation)
@overload
def get_types(annotation: UnionType) -> tuple[type]:
pass
@overload
def get_types(annotation: GenericAlias) -> type:
pass
@overload
def get_types(annotation: TypeAliasType) -> type | tuple[type, ...]:
pass
def get_types( def get_types(
annotation: TypeAliasType | GenericAlias | UnionType, annotation: TypeAliasType | GenericAlias | UnionType | type,
) -> type | tuple[type, ...]: ) -> Iterator[type]:
if isinstance(annotation, TypeAliasType): if isinstance(annotation, TypeAliasType):
annotation = annotation.__value__ # pyright: ignore[reportAny] yield from get_types(
cast(GenericAlias | UnionType | type, annotation.__value__)
)
return
if isinstance(annotation, GenericAlias): if isinstance(annotation, GenericAlias):
# print(annotation) # print(annotation)
# print(get_origin(annotation)) # print(get_origin(annotation))
return get_origin(annotation) yield get_origin(annotation)
return
if isinstance(annotation, UnionType): if isinstance(annotation, UnionType):
return tuple(get_union_types(annotation)) yield from get_union_types(annotation)
return cast(type, annotation) # pyright: ignore[reportInvalidCast] return
yield annotation
return

27
src/docker_compose/util/yaml_util.py
View File

@@ -1,7 +1,12 @@
import re import re
from collections.abc import Iterator, MutableMapping, Set from collections.abc import Iterator, MutableMapping, Set
from pathlib import Path from pathlib import Path
from typing import cast, get_type_hints, is_typeddict, override from typing import (
cast,
get_type_hints,
is_typeddict,
override,
)
import yaml import yaml
@@ -101,7 +106,7 @@ def read_yaml(path: Path) -> T_YamlPostRes:
return yaml.safe_load(f) # pyright: ignore[reportAny] return yaml.safe_load(f) # pyright: ignore[reportAny]
def read_typed_yaml[T: TypedYamlDict[object, object]]( def read_typed_yaml[T: TypedYamlDict](
type_: type[T], type_: type[T],
path: Path, path: Path,
) -> T: ) -> T:
@@ -112,7 +117,7 @@ def read_typed_yaml[T: TypedYamlDict[object, object]](
def path_to_typed( def path_to_typed(
type_: type[TypedYamlDict[object, object]], type_: type[TypedYamlDict],
data: T_YamlDict, data: T_YamlDict,
path: Path, path: Path,
) -> None: ) -> None:
@@ -124,7 +129,7 @@ def path_to_typed(
def validate_typed_dict( def validate_typed_dict(
t: type[TypedYamlDict[object, object]], t: type[TypedYamlDict],
data: T_YamlDict, data: T_YamlDict,
) -> None: ) -> None:
keys = frozenset(data.keys()) keys = frozenset(data.keys())
@@ -136,21 +141,17 @@ def validate_typed_dict(
raise KeyError(f"extra key(s): {', '.join(map(str, extra))}") raise KeyError(f"extra key(s): {', '.join(map(str, extra))}")
hints = get_type_hints(t) hints = get_type_hints(t)
for key, val in data.items(): for key, val in data.items():
t2 = hints[key] # pyright: ignore[reportAny] t2 = cast(type, cast(object, hints[key]))
if is_typeddict(t2): # pyright: ignore[reportAny] if is_typeddict(t2):
validate_typed_dict(t2, cast(T_YamlDict, val)) # pyright: ignore[reportAny] validate_typed_dict(t2, cast(T_YamlDict, val))
continue continue
# try: # try:
# print(t2) # print(t2)
# print(get_types(t2)) # print(get_types(t2))
t2 = get_types(t2) t2 = tuple(get_types(t2))
if not isinstance(val, t2): if not isinstance(val, t2):
msg = ( msg = ", ".join(t.__name__ for t in t2)
", ".join(t.__name__ for t in t2)
if isinstance(t2, tuple)
else t.__name__
)
e = TypeError(f"key: {key} expected *{msg}*, got *{type(val).__name__}*") e = TypeError(f"key: {key} expected *{msg}*, got *{type(val).__name__}*")
e.add_note(f"key: {key!s}") e.add_note(f"key: {key!s}")
raise e raise e