This commit is contained in:
2026-01-10 09:57:52 -06:00
parent 38b6807e70
commit 377e481803
18 changed files with 332 additions and 184 deletions

2
.idea/compose_gen.iml generated
View File

@@ -5,7 +5,7 @@
<sourceFolder url="file://$MODULE_DIR$/src" isTestSource="false" /> <sourceFolder url="file://$MODULE_DIR$/src" isTestSource="false" />
<excludeFolder url="file://$MODULE_DIR$/.venv" /> <excludeFolder url="file://$MODULE_DIR$/.venv" />
</content> </content>
<orderEntry type="jdk" jdkName="uv (compose_gen)" jdkType="Python SDK" /> <orderEntry type="jdk" jdkName="uv (compose_gen) (2)" jdkType="Python SDK" />
<orderEntry type="sourceFolder" forTests="false" /> <orderEntry type="sourceFolder" forTests="false" />
</component> </component>
</module> </module>

5
.idea/misc.xml generated
View File

@@ -3,14 +3,11 @@
<component name="Black"> <component name="Black">
<option name="sdkName" value="uv (compose_gen)" /> <option name="sdkName" value="uv (compose_gen)" />
</component> </component>
<component name="ProjectRootManager" version="2" project-jdk-name="uv (compose_gen)" project-jdk-type="Python SDK" /> <component name="ProjectRootManager" version="2" project-jdk-name="uv (compose_gen) (2)" project-jdk-type="Python SDK" />
<component name="PyrightConfiguration"> <component name="PyrightConfiguration">
<option name="enabled" value="true" /> <option name="enabled" value="true" />
</component> </component>
<component name="RuffConfiguration"> <component name="RuffConfiguration">
<option name="enabled" value="true" /> <option name="enabled" value="true" />
</component> </component>
<component name="TyConfiguration">
<option name="enabled" value="true" />
</component>
</project> </project>

View File

@@ -8,6 +8,8 @@ from docker_compose.util.yaml_util import to_yaml
def load_all() -> Iterator[Rendered]: def load_all() -> Iterator[Rendered]:
for path in CFG_ROOT.iterdir(): for path in CFG_ROOT.iterdir():
if path.stem.startswith("."):
continue
if path == TRAEFIK_PATH: if path == TRAEFIK_PATH:
continue continue
yield from Rendered.from_path(path) yield from Rendered.from_path(path)
@@ -15,9 +17,7 @@ def load_all() -> Iterator[Rendered]:
def render_all() -> Iterator[str]: def render_all() -> Iterator[str]:
for rendered in load_all(): for rendered in load_all():
rendered.write() rendered()
rendered.write_bind_vols()
rendered.mk_bind_vols()
yield from rendered.proxy_nets yield from rendered.proxy_nets

View File

@@ -2,6 +2,6 @@ from pathlib import Path
ROOT = Path("/nas") ROOT = Path("/nas")
DATA_ROOT = ROOT.joinpath("apps") DATA_ROOT = ROOT.joinpath("apps")
CFG_ROOT = ROOT.joinpath("cfg/templates") CFG_ROOT = ROOT.joinpath("docker_templates")
TRAEFIK_PATH = CFG_ROOT.joinpath("traefik") TRAEFIK_PATH = CFG_ROOT.joinpath("traefik")
# TEMPLATE_DIR = CFG_ROOT.joinpath("templates") # TEMPLATE_DIR = CFG_ROOT.joinpath("templates")

View File

@@ -1,11 +1,11 @@
from collections.abc import Iterator from collections.abc import Iterator
from dataclasses import dataclass from dataclasses import dataclass
from itertools import chain from itertools import chain
from shutil import copyfile
from typing import Self, final from typing import Self, final
from docker_compose.cfg.compose_paths import ComposePaths from docker_compose.cfg.compose_paths import ComposePaths
from docker_compose.cfg.dest_path import DestPaths from docker_compose.cfg.dest_path import DestPaths
from docker_compose.cfg.env import Env
from docker_compose.cfg.org import OrgData from docker_compose.cfg.org import OrgData
from docker_compose.cfg.src_path import SrcPaths from docker_compose.cfg.src_path import SrcPaths
@@ -26,16 +26,17 @@ class CfgData:
yield cls( yield cls(
src_paths, src_paths,
org, org,
ComposePaths.from_iters( ComposePaths(
src_paths.service_dir.files, frozenset(src_paths.service_dir.files),
src_paths.vol_dir.files, frozenset(src_paths.vol_dir.files),
), ),
dest, dest,
) )
def pre_render(self, data: str) -> str: def pre_render(self, data: str) -> str:
for func in chain( for func in chain(
self.org_data.pre_render_funcs, self.dest_paths.pre_render_funcs self.org_data.pre_render_funcs,
self.dest_paths.pre_render_funcs,
): ):
data = func(data) data = func(data)
return data return data
@@ -55,8 +56,11 @@ class CfgData:
data = func(data) data = func(data)
return data return data
def mk_compose_env(self) -> None: def mk_compose_env(self, force: bool = False) -> None:
src = self.src_paths.env_file src = self.src_paths.env_file
dest = self.dest_paths.env_file dest = self.dest_paths.env_file
if src.exists() and not dest.exists(): if not src.exists():
_ = copyfile(src, dest) return
if dest.exists() and not force:
return
Env.copy(src, dest)

View File

@@ -1,56 +1,45 @@
from collections.abc import Callable, Iterable, 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 Self, cast, final, override from typing import ClassVar, cast, final
import yaml import yaml
from docker_compose.cfg.org import OrgData from docker_compose.cfg.org import OrgData
from docker_compose.cfg.record import Record, RecordCls, RecordName from docker_compose.cfg.replace import ReplaceDynamic, RecordCls
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 ServiceValNew: class ServiceVal(RecordCls):
val: Path rep: ClassVar[str] = "service"
@override
def __str__(self) -> str:
return str(RecordName(self.val.stem))
@property
def replace(self) -> Record[Self, str]:
return Record(self, self.val.stem)
@final
@dataclass(frozen=True, slots=True)
class ServiceVal(RecordCls[ServiceValNew]):
old = RecordName("service")
@property
def stage_two(self):
return self.new.replace
@final @final
@dataclass(frozen=True, slots=True) @dataclass(frozen=True, slots=True)
class ServicePath: class ServicePath:
fqdn = Record[RecordName, str](
RecordName("fqdn"),
f"{OrgData.org_app.old!s}_{ServiceVal.old!s}",
)
path: Path path: Path
replace: ServiceVal = field(init=False) fqdn: ReplaceDynamic = field(init=False)
replace_pre: ReplaceDynamic = field(init=False)
replace_post: ReplaceDynamic = field(init=False)
def __post_init__(self): def __post_init__(self):
setter = super(ServicePath, self).__setattr__ setter = super(ServicePath, self).__setattr__
setter("replace", ServiceVal(ServiceValNew(self.path))) pre, post = ServiceVal.two_stage(self.path.stem)
setter("replace_pre", pre)
setter("replace_post", post)
setter(
"fqdn",
ReplaceDynamic(
"fqdn",
f"{OrgData.org_app!s}_{pre!s}",
),
)
@property @property
def as_dict(self) -> ServiceYamlRead: def as_dict(self) -> ServiceYamlRead:
@@ -67,11 +56,10 @@ class ServicePath:
@property @property
def pre_render_funcs(self) -> Iterator[Callable[[str], str]]: def pre_render_funcs(self) -> Iterator[Callable[[str], str]]:
yield self.fqdn yield self.fqdn
yield self.replace yield self.replace_pre
@property # for service in self.services:
def render_funcs(self) -> Iterator[Callable[[str], str]]: # yield service.replace_pre
yield self.replace.stage_two
@final @final
@@ -94,19 +82,24 @@ class ComposePaths:
services: frozenset[ServicePath] services: frozenset[ServicePath]
volumes: frozenset[VolumePath] volumes: frozenset[VolumePath]
@classmethod @property
def from_iters(cls, services: Iterable[ServicePath], volumes: Iterable[VolumePath]): def render_funcs(self) -> Iterator[Callable[[str], str]]:
return cls( for service in self.services:
frozenset(services), yield service.replace_post
frozenset(volumes),
) # @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:
yield path.as_k_v yield path.as_k_v
@property # @property
def render_funcs(self) -> Iterator[Callable[[str], str]]: # def render_funcs(self) -> Iterator[Callable[[str], str]]:
for path in self.services: # for path in self.services:
yield from path.render_funcs # yield from path.render_funcs

View File

@@ -5,8 +5,8 @@ from pathlib import Path
from typing import Self, final 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 AppVal, OrgData, OrgVal from docker_compose.cfg.org import App, Org, OrgData
from docker_compose.cfg.record import Record, RecordName from docker_compose.cfg.replace import ReplaceDynamic
@final @final
@@ -43,10 +43,11 @@ class ComposeFileRendered:
@final @final
@dataclass(frozen=True, slots=True) @dataclass(frozen=True, slots=True)
class DestPaths: class DestPaths:
data_root = Record[RecordName, Path](RecordName("data_root"), DATA_ROOT) # data_root = Record("data_root", str(DATA_ROOT))
data_path = Record[RecordName, str]( data_root = ReplaceDynamic.auto_format("data", str(DATA_ROOT))
RecordName("data"), data_path = ReplaceDynamic(
sep.join((str(data_root.old), str(OrgVal.old), str(AppVal.old))), data_root.src,
sep.join((data_root.src, Org.rep, App.rep)),
) )
data_dir: Path data_dir: Path
env_file: Path = field(init=False) env_file: Path = field(init=False)
@@ -60,7 +61,7 @@ class DestPaths:
@classmethod @classmethod
def from_org(cls, org: OrgData) -> Self: def from_org(cls, org: OrgData) -> Self:
return cls.from_path(DATA_ROOT.joinpath(org.org.val, org.app.val)) return cls.from_path(DATA_ROOT.joinpath(str(org.org), str(org.app)))
@classmethod @classmethod
def from_path(cls, path: Path) -> Self: def from_path(cls, path: Path) -> Self:

View File

@@ -0,0 +1,53 @@
import re
import secrets
from collections.abc import Iterator
from dataclasses import dataclass
from functools import partial
from pathlib import Path
from typing import Self, final
from docker_compose.cfg.replace import ReplaceDynamic
@final
@dataclass
class Env:
pswd = ReplaceDynamic.auto_format(
"pswd",
partial(secrets.token_urlsafe, 12),
)
data: dict[str, str]
@classmethod
def get_lines(cls, data: str) -> Iterator[tuple[str, str]]:
line_valid = re.compile(r"(^\w+)=(.+)\s*")
for line in data.splitlines():
res = line_valid.match(line)
if not res:
continue
yield res.group(1), res.group(2)
@classmethod
def from_path(cls, path: Path) -> Self:
with path.open(mode="rt") as f:
data = f.read()
return cls({k: v for k, v in cls.get_lines(data)})
@property
def with_pass(self)->Iterator[tuple[str,str]]:
for k,v in self.data.items():
if self.pswd.src not in v:
yield k,v
continue
yield k,self.pswd(v)
@property
def as_txt(self) -> str:
return '\n'.join(sorted(map('='.join, self.with_pass)))
@classmethod
def copy(cls, src: Path, dest: Path) -> None:
txt = cls.from_path(src).as_txt
with dest.open(mode="wt") as f:
_ = f.write(txt)

View File

@@ -1,80 +1,106 @@
from collections.abc import Iterator from collections.abc import Iterator
from dataclasses import dataclass, field from dataclasses import dataclass
from typing import Callable, ClassVar, Self, final, override from typing import Callable, ClassVar, Self, final, override
from docker_compose.cfg.org_yaml import OrgDataYaml from docker_compose.cfg.org_yaml import OrgDataYaml
from docker_compose.cfg.record import Record, RecordCls, RecordName from docker_compose.cfg.replace import ReplaceDynamic, RecordCls
#
# _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 OrgVal(RecordCls[str]): class Org(RecordCls):
old: ClassVar[RecordName] = RecordName("org") rep: ClassVar[str] = "org"
@final @final
@dataclass(frozen=True, slots=True) @dataclass(frozen=True, slots=True)
class Org: class App(RecordCls):
val: str rep: ClassVar[str] = "name"
replace: OrgVal = field(init=False)
def __post_init__(self) -> None:
setter = super(Org, self).__setattr__
setter("replace", OrgVal(self.val))
@final @final
@dataclass(frozen=True, slots=True) @dataclass(frozen=True, slots=True)
class AppVal(RecordCls[str]): class Url(RecordCls):
old: ClassVar[RecordName] = RecordName("name") rep: ClassVar[str] = "url"
@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 @override
def __str__(self) -> str: @classmethod
if not self.val: def from_str(cls, string: str | None) -> Self:
return "" return super(Url, cls).from_str(".".join((string, "ccamper7", "net")) if string else "")
return ".".join((self.val, "ccamper7", "net"))
# return Record("url", ".".join((val, "ccamper7", "net")) if val else "")
@final #
@dataclass(frozen=True, slots=True) # @final
class UrlVal(RecordCls[UrlValNew]): # @dataclass(frozen=True, slots=True)
old = RecordName("url") # 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 # @final
@dataclass(frozen=True, slots=True) # @dataclass(frozen=True, slots=True)
class Url: # class AppVal(RecordCls[str]):
val: str | None # old: ClassVar[RecordName] = RecordName("name")
replace: UrlVal = field(init=False) #
#
# @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))
def __post_init__(self) -> None:
setter = super(Url, self).__setattr__ # @final
setter("replace", UrlVal(UrlValNew(self.val))) # @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 = Record[RecordName, str]( org_app = ReplaceDynamic(
RecordName(f"{OrgVal.old.val}_{AppVal.old.val}"), f"${Org.rep.upper()}_{Org.rep.upper()}",
f"{OrgVal.old!s}_{AppVal.old!s}", f"{ReplaceDynamic.get_format(Org.rep)}_{ReplaceDynamic.get_format(App.rep)}",
) )
app: App app: App
org: Org org: Org
@@ -83,16 +109,16 @@ 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(app), App.from_str(app),
Org(org), Org.from_str(org),
Url(data.get("url")), Url.from_str(data.get("url")),
) )
@property @property
def render_funcs(self) -> Iterator[Callable[[str], str]]: def render_funcs(self) -> Iterator[Callable[[str], str]]:
yield self.app.replace yield self.app
yield self.org.replace yield self.org
yield self.url.replace yield self.url
@property @property
def pre_render_funcs(self) -> Iterator[Callable[[str], str]]: def pre_render_funcs(self) -> Iterator[Callable[[str], str]]:

View File

@@ -1,38 +0,0 @@
from dataclasses import dataclass
from typing import ClassVar, Protocol, final, override
class String(Protocol):
@override
def __str__(self) -> str: ...
@final
@dataclass(frozen=True, slots=True)
class RecordName:
val: str
@override
def __str__(self) -> str:
return f"${{_{str(self.val).upper()}}}"
# def replace(self, string: str) -> str:
# return string.replace(str(self), str(self))
@dataclass(frozen=True, slots=True)
class RecordCls[T_New: String]:
old: ClassVar[String]
new: T_New
def __call__(self, string: str) -> str:
return string.replace(str(self.old), str(self.new))
@dataclass(frozen=True, slots=True)
class Record[T_Old: String, T_New: String]:
old: T_Old
new: T_New
def __call__(self, string: str) -> str:
return string.replace(str(self.old), str(self.new))

View File

@@ -0,0 +1,87 @@
from collections.abc import Callable
from dataclasses import dataclass, field
from typing import ClassVar, Self, override, final
#
# class String(Protocol):
# @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
@dataclass(frozen=True, slots=True)
class ReplaceStatic:
val: 'str| ReplaceStatic'
fmt: str = field(init=False)
def __post_init__(self):
setter = super(ReplaceStatic, self).__setattr__
setter('fmt', f"${{_{self.val.upper()}}}")
def __call__(self, string: str) -> str:
return string.replace(
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)
class ReplaceDynamic:
src: ReplaceStatic
dest: str | Callable[[], str]
def __call__(self, string: str) -> str:
return string.replace(
self.src,
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)
class RecordCls(ReplaceDynamic):
# val: ClassVar[str]# = 'org'
rep: ClassVar[ReplaceStatic] # = Record.get_format(val)
@override
@classmethod
def from_str(cls, string: str) -> Self:
return cls(cls.rep, string)
@classmethod
def two_stage(cls, dest: str) -> tuple[Self, Self]:
dest_var = ReplaceStatic(dest)
return cls(cls.rep, dest_var.fmt), cls(dest_var, dest)

View File

@@ -53,7 +53,7 @@ class Compose:
@property @property
def as_dict(self) -> ComposeYaml: def as_dict(self) -> ComposeYaml:
return ComposeYaml( return ComposeYaml(
name=str(OrgData.org_app.old), name=str(OrgData.org_app),
services={ services={
service.service_name: service.as_dict for service in self.services service.service_name: service.as_dict for service in self.services
}, },
@@ -67,3 +67,6 @@ class Compose:
def write_template(self): def write_template(self):
self.cfg.src_paths.compose_file.write(self.as_template) self.cfg.src_paths.compose_file.write(self.as_template)
def __call__(self):
self.write_template()

View File

@@ -27,7 +27,7 @@ class NetArgs:
@override @override
def __str__(self) -> str: def __str__(self) -> str:
return f"{OrgData.org_app.old!s}_{self.name}" return f"{OrgData.org_app!s}_{self.name}"
@property @property
def external(self) -> bool: def external(self) -> bool:

View File

@@ -1,7 +1,7 @@
from collections.abc import Iterator from collections.abc import Iterator
from dataclasses import dataclass from dataclasses import dataclass
from pathlib import Path from pathlib import Path
from typing import final from typing import final, override
from docker_compose.cfg import ROOT from docker_compose.cfg import ROOT
from docker_compose.compose.compose import Compose from docker_compose.compose.compose import Compose
@@ -52,3 +52,11 @@ class Rendered(Compose):
def write(self) -> None: def write(self) -> None:
self.cfg.dest_paths.compose_file.write(self.as_rendered) self.cfg.dest_paths.compose_file.write(self.as_rendered)
@override
def __call__(self, force_env:bool=False) -> None:
super(Rendered, self).__call__()
self.mk_bind_vols()
self.cfg.mk_compose_env(force_env)
self.write()

View File

@@ -1,8 +1,9 @@
from dataclasses import dataclass from dataclasses import dataclass
from typing import Self, final, override from typing import Self, final, override
from docker_compose.cfg.compose_paths import ServicePath, ServiceVal from docker_compose.cfg.compose_paths import ServicePath
from docker_compose.cfg.org import OrgData, UrlVal from docker_compose.cfg.org import OrgData, Url
from docker_compose.cfg.replace import ReplaceDynamic
from docker_compose.compose.services_yaml import ( from docker_compose.compose.services_yaml import (
HealthCheck, HealthCheck,
ServiceYamlRead, ServiceYamlRead,
@@ -16,24 +17,25 @@ 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.old!s}.rule=Host(`{UrlVal.old!s}`)", f"traefik.http.routers.{OrgData.org_app.src}.rule=Host(`{Url.rep}`)",
f"traefik.http.routers.{OrgData.org_app.old!s}.entrypoints=websecure", f"traefik.http.routers.{OrgData.org_app.src}.entrypoints=websecure",
f"traefik.docker.network={OrgData.org_app.old!s}_proxy", f"traefik.docker.network={OrgData.org_app.src}_proxy",
f"traefik.http.routers.{OrgData.org_app.old!s}.tls.certresolver=le", f"traefik.http.routers.{OrgData.org_app.src}.tls.certresolver=le",
) )
) )
@override @override
def __hash__(self) -> int: def __hash__(self) -> int:
return hash(self.service_name) return hash(str(self.fqdn))
@property @property
def service_name(self) -> str: def service_name(self) -> str:
return self.service_val.new.val.stem return str(self.fqdn).split("_", maxsplit=3)[-1]
_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
command: tuple[str, ...] command: tuple[str, ...]
entrypoint: tuple[str, ...] entrypoint: tuple[str, ...]
environment: dict[str, T_Primitive] environment: dict[str, T_Primitive]
@@ -46,21 +48,23 @@ class Service:
user: str | None user: str | None
volumes: frozenset[str] volumes: frozenset[str]
shm_size: str | None shm_size: str | None
depends_on: frozenset[str] depends_on: frozenset[str] | dict[str, dict[str, str]]
healthcheck: HealthCheck | None healthcheck: HealthCheck | None
ports: frozenset[str] ports: frozenset[str]
@classmethod @classmethod
def from_path(cls, path: ServicePath) -> Self: def from_path(cls, path: ServicePath) -> Self:
return cls.from_dict(path.replace, path.as_dict) return cls.from_dict(path.fqdn, path.as_dict)
@classmethod @classmethod
def from_dict(cls, service_val: ServiceVal, data: ServiceYamlRead) -> Self: def from_dict(cls, fqdn: ReplaceDynamic, 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", ()))
deps = data.get("depends_on", ())
return cls( return cls(
service_val, # service_val,
fqdn,
tuple(data.get("command", ())), tuple(data.get("command", ())),
tuple(data.get("entrypoint", ())), tuple(data.get("entrypoint", ())),
data.get("environment", {}), data.get("environment", {}),
@@ -75,7 +79,7 @@ class Service:
data.get("user"), data.get("user"),
frozenset(data.get("volumes", ())), frozenset(data.get("volumes", ())),
data.get("shm_size"), data.get("shm_size"),
frozenset(data.get("depends_on", ())), deps if isinstance(deps, dict) else frozenset(deps),
data.get("healthcheck"), data.get("healthcheck"),
frozenset(data.get("ports", ())), frozenset(data.get("ports", ())),
) )
@@ -93,7 +97,7 @@ class Service:
security_opt=self.security_opt, security_opt=self.security_opt,
user=self.user, user=self.user,
volumes=self.volumes, volumes=self.volumes,
container_name=self.service_val(ServicePath.fqdn.new), container_name=str(self.fqdn),
restart=self.restart, restart=self.restart,
shm_size=self.shm_size, shm_size=self.shm_size,
depends_on=self.depends_on, depends_on=self.depends_on,

View File

@@ -23,7 +23,7 @@ class ServiceYamlRead(TypedDict):
user: NotRequired[str] user: NotRequired[str]
volumes: NotRequired[list[str]] volumes: NotRequired[list[str]]
shm_size: NotRequired[str] shm_size: NotRequired[str]
depends_on: NotRequired[list[str]] depends_on: NotRequired[list[str]|dict[str,dict[str,str]]]
healthcheck: NotRequired[HealthCheck] healthcheck: NotRequired[HealthCheck]
ports: NotRequired[list[str]] ports: NotRequired[list[str]]
@@ -42,6 +42,6 @@ class ServiceYamlWrite(TypedDict):
container_name: str container_name: str
restart: str restart: str
shm_size: str | None shm_size: str | None
depends_on: frozenset[str] depends_on: frozenset[str]|dict[str,dict[str,str]]
healthcheck: HealthCheck | None healthcheck: HealthCheck | None
ports: frozenset[str] ports: frozenset[str]

View File

@@ -93,6 +93,8 @@ def get_types(
if isinstance(annotation, TypeAliasType): if isinstance(annotation, TypeAliasType):
annotation = annotation.__value__ # pyright: ignore[reportAny] annotation = annotation.__value__ # pyright: ignore[reportAny]
if isinstance(annotation, GenericAlias): if isinstance(annotation, GenericAlias):
# print(annotation)
# print(get_origin(annotation))
return get_origin(annotation) return get_origin(annotation)
if isinstance(annotation, UnionType): if isinstance(annotation, UnionType):
return tuple(get_union_types(annotation)) return tuple(get_union_types(annotation))

View File

@@ -142,10 +142,18 @@ def validate_typed_dict(
continue continue
# try: # try:
if not isinstance(val, get_types(t2)): # pyright: ignore[reportAny] # print(t2)
raise TypeError( # print(get_types(t2))
f"key: {key} expected *{type(t2).__name__}*, got *{type(val).__name__}*" # pyright: ignore[reportAny] t2 = get_types(t2)
if not isinstance(val, t2):
msg = (
", ".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.add_note(f"key: {key!s}")
raise e
# valid = isinstance(val, get_types(t2)) # valid = isinstance(val, get_types(t2))
# except TypeError: # except TypeError: