This commit is contained in:
2026-01-11 12:35:27 -06:00
parent 4216e833e5
commit 894dc2f3e0
12 changed files with 121 additions and 86 deletions

18
docs/cls.md Normal file
View File

@@ -0,0 +1,18 @@
```mermaid
erDiagram
CfgData {
SrcPaths src_paths
OrgData org_data
ComposePaths compose_paths
DestPaths dest_paths
}
OrgData {
org_app ReplaceUniqe
App app
Org org
Url url
}
CfgData ||--|| OrgData : contains
```

View File

@@ -4,7 +4,7 @@ from typing import cast
from docker_compose.cfg import CFG_ROOT, TRAEFIK_PATH from docker_compose.cfg import CFG_ROOT, TRAEFIK_PATH
from docker_compose.compose.net_yaml import NetArgsYaml from docker_compose.compose.net_yaml import NetArgsYaml
from docker_compose.compose.rendered import Rendered from docker_compose.compose.rendered import Rendered
from docker_compose.util.Ts import TypeYamlCompatableDict from docker_compose.util.Ts import TypeYamlCompatibleDict
from docker_compose.util.yaml_util import to_yaml from docker_compose.util.yaml_util import to_yaml
@@ -31,7 +31,7 @@ if __name__ == "__main__":
data["networks"] = {net: NetArgsYaml(name=f"{net}_proxy") for net in nets} data["networks"] = {net: NetArgsYaml(name=f"{net}_proxy") for net in nets}
cfg = traefik.cfg cfg = traefik.cfg
data["services"]["traefik"]["networks"] = nets data["services"]["traefik"]["networks"] = nets
data = cast(TypeYamlCompatableDict, cast(object, data)) data = cast(TypeYamlCompatibleDict, cast(object, data))
template = cfg.pre_render(to_yaml(data)) template = cfg.pre_render(to_yaml(data))
cfg.src_paths.compose_file.write(template) cfg.src_paths.compose_file.write(template)
cfg.dest_paths.compose_file.write(cfg.render(template)) cfg.dest_paths.compose_file.write(cfg.render(template))

View File

@@ -12,18 +12,18 @@ from docker_compose.compose.volume_yaml import VolYaml
from docker_compose.util.Ts import TypeYamlDict from docker_compose.util.Ts import TypeYamlDict
from docker_compose.util.yaml_util import path_to_typed, read_yaml from docker_compose.util.yaml_util import path_to_typed, read_yaml
#
@final # @final
@dataclass(frozen=True, slots=True) # @dataclass(frozen=True, slots=True)
class ServiceVal(ReplaceStatic): # class ServiceVal(ReplaceStatic):
src = ReplaceDynamic("service") # src = ReplaceDynamic("service")
@final @final
@dataclass(frozen=True, slots=True) @dataclass(frozen=True, slots=True)
class ServicePath: class ServicePath:
path: Path path: Path
fqdn: ReplaceUnique = field(init=False) # fqdn: ReplaceUnique = field(init=False)
replace_pre: ReplaceStatic = field(init=False) replace_pre: ReplaceStatic = field(init=False)
replace_post: ReplaceStatic = field(init=False) replace_post: ReplaceStatic = field(init=False)
@@ -33,12 +33,15 @@ class ServicePath:
setter("replace_pre", pre) setter("replace_pre", pre)
setter("replace_post", post) setter("replace_post", post)
setter( # setter(
"fqdn", # "fqdn",
ReplaceUnique.auto_format( # ReplaceUnique.build_placeholder(
"fqdn", str(Org.src.build_placeholder(App.src, pre.src)) # "fqdn",
), # Org,
) # App,
# ServiceVal,
# ),
# )
@property @property
def as_dict(self) -> ServiceYamlRead: def as_dict(self) -> ServiceYamlRead:

View File

@@ -47,7 +47,7 @@ class DestPaths:
data_root = ReplaceUnique.auto_format("data", str(DATA_ROOT)) data_root = ReplaceUnique.auto_format("data", str(DATA_ROOT))
data_path = ReplaceUnique( data_path = ReplaceUnique(
data_root.src, data_root.src,
sep.join((data_root.src, str(Org), str(App))), sep.join((data_root.src, Org.src.fmt, App.src.fmt)),
) )
data_dir: Path data_dir: Path
env_file: Path = field(init=False) env_file: Path = field(init=False)
@@ -61,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(str(org.org), str(org.app))) return cls.from_path(DATA_ROOT.joinpath(org.org.dest, org.app.dest))
@classmethod @classmethod
def from_path(cls, path: Path) -> Self: def from_path(cls, path: Path) -> Self:

View File

@@ -3,7 +3,7 @@ from dataclasses import dataclass
from typing import Callable, 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, ReplaceStatic from docker_compose.cfg.replace import ReplaceDynamic, ReplaceStatic, ReplaceUnique
@final @final
@@ -35,7 +35,7 @@ class Url(ReplaceStatic):
@final @final
@dataclass(frozen=True, slots=True) @dataclass(frozen=True, slots=True)
class OrgData: class OrgData:
org_app = Org.src.build_placeholder(App.src) org_app = ReplaceUnique.build_placeholder('dn', Org, App)
app: App app: App
org: Org org: Org
url: Url url: Url

View File

@@ -1,6 +1,5 @@
from collections.abc import Callable from collections.abc import Callable
from dataclasses import dataclass, field from dataclasses import dataclass, field
from itertools import chain
from typing import ClassVar, Self, final from typing import ClassVar, Self, final
@@ -21,28 +20,32 @@ class ReplaceUnique:
def auto_format(cls, src: str, dest: str): def auto_format(cls, src: str, dest: str):
return cls(format_src(src), dest) return cls(format_src(src), dest)
@classmethod
def build_placeholder(cls, src: str, *dest: "type[ReplaceStatic]") -> Self:
return cls.auto_format(src, '_'.join(arg.src.fmt for arg in dest),)
@final @final
@dataclass(frozen=True, slots=True) @dataclass(frozen=True, slots=True)
class ReplaceDynamic: class ReplaceDynamic:
val: str val: str
fmt: str = field(init=False) fmt: str
def __post_init__(self): @classmethod
setter = super(ReplaceDynamic, self).__setattr__ def factory(cls, val:str):
setter("fmt", format_src(self.val)) return cls(val, format_src(val))
def __call__(self, string: str) -> str: def __call__(self, string: str) -> str:
return string.replace(self.fmt, self.val) return string.replace(self.fmt, self.val)
# def __str__(self) -> str: # def __str__(self) -> str:
# return self.val if isinstance(self.val, str) else self.val.fmt # return self.val if isinstance(self.val, str) else self.val.fmt
def build_placeholder(self, *args: "ReplaceDynamic"): # def build_placeholder(self, *args: "ReplaceDynamic"):
data = ((rep.val.upper(), rep.fmt) for rep in chain((self,), args)) # data = ((rep.val.upper(), rep.fmt) for rep in chain((self,), args))
src: tuple[str, ...] # src: tuple[str, ...]
dest: tuple[str, ...] # dest: tuple[str, ...]
src, dest = zip(*data) # src, dest = zip(*data)
return ReplaceUnique("_".join(src), "_".join(dest)) # return ReplaceUnique("_".join(src), "_".join(dest))
@dataclass(frozen=True, slots=True) @dataclass(frozen=True, slots=True)

View File

@@ -5,14 +5,14 @@ from typing import Self, cast, final
from docker_compose.cfg.compose_paths import ServicePath, VolumePath from docker_compose.cfg.compose_paths import ServicePath, VolumePath
from docker_compose.cfg.org_yaml import OrgYaml from docker_compose.cfg.org_yaml import OrgYaml
from docker_compose.util.Ts import TypeYamlCompatableDict, TypeYamlCompatableRes from docker_compose.util.Ts import TypeYamlCompatibleDict, TypeYamlCompatibleRes
from docker_compose.util.yaml_util import read_yaml, write_yaml from docker_compose.util.yaml_util import read_yaml, write_yaml
YAML_EXTS = frozenset((".yml", ".yaml")) YAML_EXTS = frozenset((".yml", ".yaml"))
class ComposeFileTemplate(Path): class ComposeFileTemplate(Path):
def write_dict(self, data: TypeYamlCompatableDict) -> None: def write_dict(self, data: TypeYamlCompatibleDict) -> None:
write_yaml(data, self) write_yaml(data, self)
def write(self, data: str) -> None: def write(self, data: str) -> None:
@@ -68,7 +68,7 @@ class VolumesDir(YamlDir):
class VolumeData(Path): class VolumeData(Path):
def write(self, data: TypeYamlCompatableRes) -> None: def write(self, data: TypeYamlCompatibleRes) -> None:
write_yaml(data, self) write_yaml(data, self)

View File

@@ -10,7 +10,7 @@ from docker_compose.compose.compose_yaml import ComposeYaml
from docker_compose.compose.net import Net from docker_compose.compose.net import Net
from docker_compose.compose.services import Service from docker_compose.compose.services import Service
from docker_compose.compose.volume_yaml import VolYaml from docker_compose.compose.volume_yaml import VolYaml
from docker_compose.util.Ts import TypeYamlCompatableDict from docker_compose.util.Ts import TypeYamlCompatibleDict
from docker_compose.util.yaml_util import to_yaml from docker_compose.util.yaml_util import to_yaml
@@ -54,7 +54,7 @@ class Compose:
@property @property
def as_dict(self) -> ComposeYaml: def as_dict(self) -> ComposeYaml:
return ComposeYaml( return ComposeYaml(
name=str(OrgData.org_app.dest), name=OrgData.org_app.dest,
services={ services={
service.service_name: service.as_dict for service in self.services service.service_name: service.as_dict for service in self.services
}, },
@@ -64,7 +64,7 @@ class Compose:
@property @property
def as_template(self) -> str: def as_template(self) -> str:
data = cast(TypeYamlCompatableDict, cast(object, self.as_dict)) data = cast(TypeYamlCompatibleDict, cast(object, self.as_dict))
return self.cfg.pre_render(to_yaml(data)) return self.cfg.pre_render(to_yaml(data))
def write_template(self): def write_template(self):

View File

@@ -1,6 +1,6 @@
from collections.abc import Iterable, Iterator from collections.abc import Iterable, Iterator
from dataclasses import dataclass from dataclasses import dataclass, field
from typing import Self, final, override from typing import Self, final
from docker_compose.cfg.org import OrgData from docker_compose.cfg.org import OrgData
from docker_compose.compose.net_yaml import NetArgsYaml, NetYaml from docker_compose.compose.net_yaml import NetArgsYaml, NetYaml
@@ -11,11 +11,18 @@ from docker_compose.compose.services import Service
@dataclass(frozen=True, slots=True) @dataclass(frozen=True, slots=True)
class NetArgs: class NetArgs:
name: str name: str
full_name: str = field(init=False)
external: bool = field(init=False)
def __post_init__(self):
setter = super(NetArgs, self).__setattr__
setter("full_name", f"{OrgData.org_app.dest}_{self.name}")
setter("external", self.name == "proxy")
@property @property
def as_dict(self) -> NetArgsYaml: def as_dict(self) -> NetArgsYaml:
yaml_dict = NetArgsYaml( yaml_dict = NetArgsYaml(
name=str(self), name=self.full_name,
) )
if self.external: if self.external:
yaml_dict["external"] = self.external yaml_dict["external"] = self.external
@@ -25,14 +32,6 @@ class NetArgs:
# def as_key_dict(self) -> tuple[str, NetArgsYaml]: # def as_key_dict(self) -> tuple[str, NetArgsYaml]:
# return str(self), self.as_dict # return str(self), self.as_dict
@override
def __str__(self) -> str:
return f"{OrgData.org_app.dest}_{self.name}"
@property
def external(self) -> bool:
return self.name == "proxy"
@final @final
@dataclass @dataclass
@@ -58,4 +57,4 @@ class Net:
for net in self.data: for net in self.data:
if not net.external: if not net.external:
continue continue
yield str(net)[:-6] yield net.full_name[:-6]

View File

@@ -1,9 +1,9 @@
from dataclasses import dataclass from dataclasses import dataclass, field
from typing import Self, final, override from typing import Self, final, override
from docker_compose.cfg.compose_paths import ServicePath from docker_compose.cfg.compose_paths import ServicePath, ServiceVal
from docker_compose.cfg.org import OrgData, Url from docker_compose.cfg.org import App, Org, OrgData, Url
from docker_compose.cfg.replace import ReplaceUnique from docker_compose.cfg.replace import ReplaceDynamic, ReplaceStatic, ReplaceUnique
from docker_compose.compose.services_yaml import ( from docker_compose.compose.services_yaml import (
HealthCheck, HealthCheck,
ServiceYamlRead, ServiceYamlRead,
@@ -12,6 +12,12 @@ from docker_compose.compose.services_yaml import (
from docker_compose.util.Ts import T_Primitive from docker_compose.util.Ts import T_Primitive
@final
@dataclass(frozen=True, slots=True)
class ServiceVal(ReplaceStatic):
src = ReplaceDynamic("service")
@final @final
@dataclass(frozen=True, slots=True) @dataclass(frozen=True, slots=True)
class Service: class Service:
@@ -23,19 +29,17 @@ class Service:
f"traefik.http.routers.{OrgData.org_app.dest}.tls.certresolver=le", f"traefik.http.routers.{OrgData.org_app.dest}.tls.certresolver=le",
) )
) )
@override
def __hash__(self) -> int:
return hash(str(self.fqdn))
@property
def service_name(self) -> str:
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
fqdn = ReplaceUnique.build_placeholder("fqdn", Org, App, ServiceVal)
# @property
# def service_name(self) -> str:
# return self.fqdn.dest.split("_", maxsplit=3)[-1]
service_rep: ServiceVal = field(init=False)
service_name: str
# service_val: ServiceVal # service_val: ServiceVal
fqdn: ReplaceUnique # fqdn: ReplaceUnique
command: tuple[str, ...] command: tuple[str, ...]
entrypoint: tuple[str, ...] entrypoint: tuple[str, ...]
environment: dict[str, T_Primitive] environment: dict[str, T_Primitive]
@@ -52,19 +56,27 @@ class Service:
healthcheck: HealthCheck | None healthcheck: HealthCheck | None
ports: frozenset[str] ports: frozenset[str]
@classmethod def __post_init__(self):
def from_path(cls, path: ServicePath) -> Self: setter = super(Service, self).__setattr__
return cls.from_dict(path.fqdn, path.as_dict) setter("service_rep", ServiceVal(self.service_name))
@override
def __hash__(self) -> int:
return hash(self.service_name)
@classmethod @classmethod
def from_dict(cls, fqdn: ReplaceUnique, data: ServiceYamlRead) -> Self: def from_path(cls, path: ServicePath) -> Self:
return cls.from_dict(path.path.stem, path.as_dict)
@classmethod
def from_dict(cls, name: str, 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", ()) deps = data.get("depends_on", ())
return cls( return cls(
# service_val, # service_val,
fqdn, name,
tuple(data.get("command", ())), tuple(data.get("command", ())),
tuple(data.get("entrypoint", ())), tuple(data.get("entrypoint", ())),
data.get("environment", {}), data.get("environment", {}),
@@ -97,7 +109,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=str(self.fqdn), container_name=self.fqdn.dest,
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

@@ -38,10 +38,10 @@ class TypeYamlDict(Protocol):
#yaml compatible data #yaml compatible data
type TypeYamlCompatableIters = Sequence[TypeYamlCompatable] | Set[TypeYamlCompatable] | Iterator[TypeYamlCompatable] type TypeYamlCompatibleIters = Sequence[TypeYamlCompatible] | Set[TypeYamlCompatible] | Iterator[TypeYamlCompatible]
type TypeYamlCompatableDict = MutableMapping[str, TypeYamlCompatable] type TypeYamlCompatibleDict = MutableMapping[str, TypeYamlCompatible]
type TypeYamlCompatableRes = TypeYamlCompatableIters | TypeYamlCompatableDict type TypeYamlCompatibleRes = TypeYamlCompatibleIters | TypeYamlCompatibleDict
type TypeYamlCompatable = T_Primitive | TypeYamlCompatableRes type TypeYamlCompatible = T_Primitive | TypeYamlCompatibleRes
# type T_YamlPostDict = MutableMapping[str, T_YamlPost] # type T_YamlPostDict = MutableMapping[str, T_YamlPost]

View File

@@ -11,10 +11,10 @@ from typing import (
import yaml import yaml
from docker_compose.util.Ts import ( from docker_compose.util.Ts import (
TypeYamlCompatable, TypeYamlCompatible,
TypeYamlCompatableDict, TypeYamlCompatibleDict,
TypeYamlCompatableIters, TypeYamlCompatibleIters,
TypeYamlCompatableRes, TypeYamlCompatibleRes,
TypeYamlDict, TypeYamlDict,
TypeYamlRes, TypeYamlRes,
get_types, get_types,
@@ -45,7 +45,7 @@ class VerboseSafeDumper(yaml.SafeDumper):
return True return True
def yaml_prep(data: TypeYamlCompatableRes) -> TypeYamlCompatableRes: def yaml_prep(data: TypeYamlCompatibleRes) -> TypeYamlCompatibleRes:
if isinstance(data, MutableMapping): if isinstance(data, MutableMapping):
return dict_prep(data) return dict_prep(data)
if isinstance(data, Sequence): if isinstance(data, Sequence):
@@ -57,7 +57,7 @@ def yaml_prep(data: TypeYamlCompatableRes) -> TypeYamlCompatableRes:
return res return res
def list_prep(data: TypeYamlCompatableIters) -> Iterator[TypeYamlCompatable]: def list_prep(data: TypeYamlCompatibleIters) -> Iterator[TypeYamlCompatible]:
for v in data: for v in data:
if isinstance(v, (MutableMapping, tuple, list, Set, Iterator)): if isinstance(v, (MutableMapping, tuple, list, Set, Iterator)):
yield yaml_prep(v) yield yaml_prep(v)
@@ -70,7 +70,7 @@ def list_prep(data: TypeYamlCompatableIters) -> Iterator[TypeYamlCompatable]:
continue continue
def dict_prep(data: TypeYamlCompatableDict) -> TypeYamlCompatableDict: def dict_prep(data: TypeYamlCompatibleDict) -> TypeYamlCompatibleDict:
keys = tuple(data.keys()) keys = tuple(data.keys())
for k in keys: for k in keys:
v = data[k] v = data[k]
@@ -82,10 +82,10 @@ def dict_prep(data: TypeYamlCompatableDict) -> TypeYamlCompatableDict:
if isinstance(v, bool): if isinstance(v, bool):
continue continue
del data[k] del data[k]
return data return data
def to_yaml(data: TypeYamlCompatableRes) -> str: def to_yaml(data: TypeYamlCompatibleRes) -> str:
dict_ = yaml_prep(data) dict_ = yaml_prep(data)
res = yaml.dump(dict_, Dumper=VerboseSafeDumper) res = yaml.dump(dict_, Dumper=VerboseSafeDumper)
res = re.sub(r"(^\s?-)", r" \g<1>", res, flags=re.MULTILINE) res = re.sub(r"(^\s?-)", r" \g<1>", res, flags=re.MULTILINE)
@@ -93,7 +93,7 @@ def to_yaml(data: TypeYamlCompatableRes) -> str:
def write_yaml( def write_yaml(
data: TypeYamlCompatableRes, data: TypeYamlCompatibleRes,
path: Path, path: Path,
) -> None: ) -> None:
with path.open("wt") as f: with path.open("wt") as f: