This commit is contained in:
2026-01-12 22:24:07 -06:00
parent 894dc2f3e0
commit 5899992240
43 changed files with 871 additions and 810 deletions

10
.idea/.gitignore generated vendored
View File

@@ -1,10 +0,0 @@
# Default ignored files
/shelf/
/workspace.xml
# Ignored default folder with query files
/queries/
# Datasource local storage ignored files
/dataSources/
/dataSources.local.xml
# Editor-based HTTP Client requests
/httpRequests/

View File

@@ -1,5 +0,0 @@
<component name="ProjectCodeStyleConfiguration">
<state>
<option name="PREFERRED_PROJECT_CODE_STYLE" value="Default" />
</state>
</component>

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) (2)" jdkType="Python SDK" /> <orderEntry type="jdk" jdkName="uv (compose_gen)" jdkType="Python SDK" />
<orderEntry type="sourceFolder" forTests="false" /> <orderEntry type="sourceFolder" forTests="false" />
</component> </component>
</module> </module>

View File

@@ -1,7 +1,6 @@
<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>

View File

@@ -0,0 +1,7 @@
<component name="InspectionProjectProfileManager">
<profile version="1.0">
<option name="myName" value="Project Default" />
<inspection_tool class="PyInconsistentReturnsInspection" enabled="false" level="WEAK WARNING" enabled_by_default="false" />
<inspection_tool class="PyTypeCheckerInspection" enabled="false" level="WARNING" enabled_by_default="false" />
</profile>
</component>

View File

@@ -1,6 +1,5 @@
<component name="InspectionProjectProfileManager"> <component name="InspectionProjectProfileManager">
<settings> <settings>
<option name="PROJECT_PROFILE" value="Default" />
<option name="USE_PROJECT_PROFILE" value="false" /> <option name="USE_PROJECT_PROFILE" value="false" />
<version value="1.0" /> <version value="1.0" />
</settings> </settings>

8
.idea/misc.xml generated
View File

@@ -3,11 +3,5 @@
<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) (2)" project-jdk-type="Python SDK" /> <component name="ProjectRootManager" version="2" project-jdk-name="uv (compose_gen)" project-jdk-type="Python SDK" />
<component name="PyrightConfiguration">
<option name="enabled" value="true" />
</component>
<component name="RuffConfiguration">
<option name="enabled" value="true" />
</component>
</project> </project>

95
.idea/workspace.xml generated Normal file
View File

@@ -0,0 +1,95 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="ChangeListManager">
<list default="true" id="05d408a8-5231-4663-b888-27493ff1596d" name="Changes" comment="">
<change afterPath="$PROJECT_DIR$/src/docker_compose/compose_data/data.py" afterDir="false" />
<change afterPath="$PROJECT_DIR$/src/docker_compose/compose_data/dest_paths.py" afterDir="false" />
<change afterPath="$PROJECT_DIR$/src/docker_compose/compose_data/main.py" afterDir="false" />
<change afterPath="$PROJECT_DIR$/src/docker_compose/compose_data/src_paths.py" afterDir="false" />
<change afterPath="$PROJECT_DIR$/src/docker_compose/env/data.py" afterDir="false" />
<change afterPath="$PROJECT_DIR$/src/docker_compose/env/main.py" afterDir="false" />
<change afterPath="$PROJECT_DIR$/src/docker_compose/org/__init__.py" afterDir="false" />
<change afterPath="$PROJECT_DIR$/src/docker_compose/org/data.py" afterDir="false" />
<change afterPath="$PROJECT_DIR$/src/docker_compose/render/__init__.py" afterDir="false" />
<change afterPath="$PROJECT_DIR$/src/docker_compose/render/main.py" afterDir="false" />
<change afterPath="$PROJECT_DIR$/src/docker_compose/util/replace.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/.idea/.gitignore" beforeDir="false" />
<change beforePath="$PROJECT_DIR$/.idea/codeStyles/codeStyleConfig.xml" beforeDir="false" />
<change beforePath="$PROJECT_DIR$/.idea/compose_gen.iml" beforeDir="false" />
<change beforePath="$PROJECT_DIR$/.idea/dictionaries/project.xml" beforeDir="false" />
<change beforePath="$PROJECT_DIR$/.idea/inspectionProfiles/profiles_settings.xml" beforeDir="false" />
<change beforePath="$PROJECT_DIR$/.idea/misc.xml" beforeDir="false" />
<change beforePath="$PROJECT_DIR$/.idea/modules.xml" beforeDir="false" />
<change beforePath="$PROJECT_DIR$/.idea/vcs.xml" beforeDir="false" />
<change beforePath="$PROJECT_DIR$/README.md" beforeDir="false" afterPath="$PROJECT_DIR$/src/docker_compose/__main__.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/src/docker_compose/__init__.py" beforeDir="false" afterPath="$PROJECT_DIR$/src/docker_compose/__init__.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/src/docker_compose/cfg/__init__.py" beforeDir="false" />
<change beforePath="$PROJECT_DIR$/src/docker_compose/cfg/cfg_paths.py" beforeDir="false" />
<change beforePath="$PROJECT_DIR$/src/docker_compose/cfg/compose_paths.py" beforeDir="false" />
<change beforePath="$PROJECT_DIR$/src/docker_compose/cfg/dest_path.py" beforeDir="false" />
<change beforePath="$PROJECT_DIR$/src/docker_compose/cfg/env.py" beforeDir="false" />
<change beforePath="$PROJECT_DIR$/src/docker_compose/cfg/org.py" beforeDir="false" />
<change beforePath="$PROJECT_DIR$/src/docker_compose/cfg/org_yaml.py" beforeDir="false" afterPath="$PROJECT_DIR$/src/docker_compose/org/org_yaml.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/src/docker_compose/cfg/replace.py" beforeDir="false" />
<change beforePath="$PROJECT_DIR$/src/docker_compose/cfg/src_path.py" beforeDir="false" />
<change beforePath="$PROJECT_DIR$/src/docker_compose/compose/__init__.py" beforeDir="false" afterPath="$PROJECT_DIR$/src/docker_compose/compose_data/__init__.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/src/docker_compose/compose/compose.py" beforeDir="false" />
<change beforePath="$PROJECT_DIR$/src/docker_compose/compose/compose_yaml.py" beforeDir="false" afterPath="$PROJECT_DIR$/src/docker_compose/compose_data/compose_yaml.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/src/docker_compose/compose/net.py" beforeDir="false" afterPath="$PROJECT_DIR$/src/docker_compose/compose_data/net.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/src/docker_compose/compose/net_yaml.py" beforeDir="false" afterPath="$PROJECT_DIR$/src/docker_compose/compose_data/net_yaml.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/src/docker_compose/compose/rendered.py" beforeDir="false" />
<change beforePath="$PROJECT_DIR$/src/docker_compose/compose/services.py" beforeDir="false" afterPath="$PROJECT_DIR$/src/docker_compose/compose_data/service.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/src/docker_compose/compose/services_yaml.py" beforeDir="false" afterPath="$PROJECT_DIR$/src/docker_compose/compose_data/services_yaml.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/src/docker_compose/compose/volume_yaml.py" beforeDir="false" afterPath="$PROJECT_DIR$/src/docker_compose/compose_data/volume_yaml.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/src/docker_compose/compose/volumes.py" beforeDir="false" afterPath="$PROJECT_DIR$/src/docker_compose/env/__init__.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/src/docker_compose/util/Ts.py" beforeDir="false" afterPath="$PROJECT_DIR$/src/docker_compose/util/Ts.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/src/docker_compose/util/yaml_util.py" beforeDir="false" afterPath="$PROJECT_DIR$/src/docker_compose/util/yaml_util.py" afterDir="false" />
</list>
<option name="SHOW_DIALOG" value="false" />
<option name="HIGHLIGHT_CONFLICTS" value="true" />
<option name="HIGHLIGHT_NON_ACTIVE_CHANGELIST" value="false" />
<option name="LAST_RESOLUTION" value="IGNORE" />
</component>
<component name="Git.Settings">
<option name="RECENT_GIT_ROOT_PATH" value="$PROJECT_DIR$" />
</component>
<component name="ProblemsViewState">
<option name="selectedTabId" value="ProjectErrors" />
</component>
<component name="ProjectColorInfo"><![CDATA[{
"associatedIndex": 1
}]]></component>
<component name="ProjectId" id="38Bbj81fMWB2dc2XPXcErk6xWjp" />
<component name="ProjectViewState">
<option name="hideEmptyMiddlePackages" value="true" />
<option name="showLibraryContents" value="true" />
</component>
<component name="PropertiesComponent"><![CDATA[{
"keyToString": {
"ModuleVcsDetector.initialDetectionPerformed": "true",
"RunOnceActivity.ShowReadmeOnStart": "true",
"RunOnceActivity.git.unshallow": "true",
"git-widget-placeholder": "main",
"last_opened_file_path": "/data/git/compose_gen",
"run.code.analysis.last.selected.profile": "pProject Default"
}
}]]></component>
<component name="SharedIndexes">
<attachedChunks>
<set>
<option value="bundled-python-sdk-ca5e2b39c7df-6e1f45a539f7-com.jetbrains.pycharm.pro.sharedIndexes.bundled-PY-253.29346.308" />
</set>
</attachedChunks>
</component>
<component name="TaskManager">
<task active="true" id="Default" summary="Default task">
<changelist id="05d408a8-5231-4663-b888-27493ff1596d" name="Changes" comment="" />
<created>1768276046955</created>
<option name="number" value="Default" />
<option name="presentableId" value="Default" />
<updated>1768276046955</updated>
<workItem from="1768276048173" duration="1016000" />
</task>
<servers />
</component>
</project>

View File

@@ -1,39 +1,6 @@
from collections.abc import Iterator from pathlib import Path
from typing import cast
from docker_compose.cfg import CFG_ROOT, TRAEFIK_PATH ROOT = Path("/nas")
from docker_compose.compose.net_yaml import NetArgsYaml TEMPLATE_ROOT = ROOT.joinpath("templates")
from docker_compose.compose.rendered import Rendered APP_ROOT = ROOT.joinpath("apps")
from docker_compose.util.Ts import TypeYamlCompatibleDict TRAEFIK_PATH = TEMPLATE_ROOT.joinpath("traefik")
from docker_compose.util.yaml_util import to_yaml
def load_all() -> Iterator[Rendered]:
for path in CFG_ROOT.iterdir():
if path.stem.startswith("."):
continue
if path == TRAEFIK_PATH:
continue
yield from Rendered.from_path(path)
def render_all() -> Iterator[str]:
for rendered in load_all():
rendered()
yield from rendered.proxy_nets
if __name__ == "__main__":
# renders = render_all()
nets = frozenset(render_all())
traefik = next(Rendered.from_path(TRAEFIK_PATH))
data = traefik.as_dict
data["networks"] = {net: NetArgsYaml(name=f"{net}_proxy") for net in nets}
cfg = traefik.cfg
data["services"]["traefik"]["networks"] = nets
data = cast(TypeYamlCompatibleDict, cast(object, data))
template = cfg.pre_render(to_yaml(data))
cfg.src_paths.compose_file.write(template)
cfg.dest_paths.compose_file.write(cfg.render(template))
traefik.write_bind_vols()
traefik.mk_bind_vols()

View File

@@ -0,0 +1,28 @@
from collections.abc import Iterator
from typing import cast
from docker_compose import TRAEFIK_PATH
from docker_compose.compose_data.net_yaml import NetArgsYaml
from docker_compose.render.main import RenderByApp, RenderByOrg
from docker_compose.util.Ts import TypeYamlCompatibleDict
from docker_compose.util.yaml_util import to_yaml
def render_all() -> Iterator[str]:
apps = RenderByApp.load_all()
apps()
return apps.proxy_nets
if __name__ == "__main__":
renderers = RenderByOrg.from_path(TRAEFIK_PATH)
traefik =renderers["util"]
data = traefik.template.compose_data.as_dict
nets = frozenset(render_all())
data["networks"] = {net: NetArgsYaml(name=f"{net}_proxy") for net in nets}
data["services"]["traefik"]["networks"] = nets
data = cast(TypeYamlCompatibleDict, cast(object, data))
txt = traefik.write(to_yaml(data), render=True)
renderers.write_bind_vol_data()
traefik.bind_vols()

View File

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

View File

@@ -1,66 +0,0 @@
from collections.abc import Iterator
from dataclasses import dataclass
from itertools import chain
from typing import Self, final
from docker_compose.cfg.compose_paths import ComposePaths
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.src_path import SrcPaths
@final
@dataclass(frozen=True, slots=True)
class CfgData:
src_paths: SrcPaths
org_data: OrgData
compose_paths: ComposePaths
dest_paths: DestPaths
@classmethod
def from_src_paths(cls, src_paths: SrcPaths) -> Iterator[Self]:
for org, args in src_paths.org_file.as_dict.items():
org = OrgData.from_dict(src_paths.app, org, args)
dest = DestPaths.from_org(org)
yield cls(
src_paths,
org,
ComposePaths(
frozenset(src_paths.service_dir.files),
frozenset(src_paths.vol_dir.files),
),
dest,
)
def pre_render(self, data: str) -> str:
for func in chain(
self.org_data.pre_render_funcs,
self.dest_paths.pre_render_funcs,
):
data = func(data)
return data
def render(self, data: str) -> str:
for func in chain(
self.org_data.render_funcs,
self.dest_paths.render_funcs,
self.compose_paths.render_funcs,
):
data = func(data)
return data
def render_all(self, data: str) -> str:
for func in (self.pre_render, self.render):
# noinspection PyArgumentList
data = func(data)
return data
def mk_compose_env(self, force: bool = False) -> None:
src = self.src_paths.env_file
dest = self.dest_paths.env_file
if not src.exists():
return
if dest.exists() and not force:
return
Env.copy(src, dest)

View File

@@ -1,100 +0,0 @@
from collections.abc import Callable, Iterator, MutableMapping
from dataclasses import dataclass, field
from pathlib import Path
from typing import cast, final
import yaml
from docker_compose.cfg.org import App, Org
from docker_compose.cfg.replace import ReplaceDynamic, ReplaceStatic, ReplaceUnique
from docker_compose.compose.services_yaml import ServiceYamlRead
from docker_compose.compose.volume_yaml import VolYaml
from docker_compose.util.Ts import TypeYamlDict
from docker_compose.util.yaml_util import path_to_typed, read_yaml
#
# @final
# @dataclass(frozen=True, slots=True)
# class ServiceVal(ReplaceStatic):
# src = ReplaceDynamic("service")
@final
@dataclass(frozen=True, slots=True)
class ServicePath:
path: Path
# fqdn: ReplaceUnique = field(init=False)
replace_pre: ReplaceStatic = field(init=False)
replace_post: ReplaceStatic = field(init=False)
def __post_init__(self):
setter = super(ServicePath, self).__setattr__
pre, post = ServiceVal.two_stage(self.path.stem)
setter("replace_pre", pre)
setter("replace_post", post)
# setter(
# "fqdn",
# ReplaceUnique.build_placeholder(
# "fqdn",
# Org,
# App,
# ServiceVal,
# ),
# )
@property
def as_dict(self) -> ServiceYamlRead:
with self.path.open("rt") as f:
data_str = f.read()
for func in self.pre_render_funcs:
data_str = func(data_str)
data_dict: TypeYamlDict = yaml.safe_load(data_str) # pyright: ignore[reportAny]
if not isinstance(data_dict, MutableMapping):
raise TypeError
path_to_typed(ServiceYamlRead, data_dict, self.path)
return cast(ServiceYamlRead, cast(object, data_dict))
@property
def pre_render_funcs(self) -> Iterator[Callable[[str], str]]:
yield self.fqdn
yield self.replace_pre
# for service in self.services:
# yield service.replace_pre
@final
@dataclass(frozen=True, slots=True)
class VolumePath:
path: Path
@property
def as_k_v(self) -> tuple[str, VolYaml]:
return self.path.stem, self.as_dict
@property
def as_dict(self) -> VolYaml:
return cast(VolYaml, cast(object, read_yaml(self.path)))
@final
@dataclass(frozen=True, slots=True)
class ComposePaths:
services: frozenset[ServicePath]
volumes: frozenset[VolumePath]
@property
def render_funcs(self) -> Iterator[Callable[[str], str]]:
for service in self.services:
yield service.replace_post
@property
def volumes_k_v(self) -> Iterator[tuple[str, VolYaml]]:
for path in self.volumes:
yield path.as_k_v
# @property
# def render_funcs(self) -> Iterator[Callable[[str], str]]:
# for path in self.services:
# yield from path.render_funcs

View File

@@ -1,82 +0,0 @@
from collections.abc import Callable, Iterator
from dataclasses import dataclass, field
from os import sep
from pathlib import Path
from typing import Self, final
from docker_compose.cfg import DATA_ROOT
from docker_compose.cfg.org import App, Org, OrgData
from docker_compose.cfg.replace import ReplaceUnique
@final
@dataclass(frozen=True, slots=True)
class ComposeFileRendered:
path: Path
def write(self, data: str) -> None:
print(self.path)
self.path.parent.mkdir(parents=True, exist_ok=True)
with self.path.open("wt") as f:
_ = f.write(data)
# @final
# @dataclass(frozen=True, slots=True)
# class DataDirReplace(RecordCls[Path]):
# old = RecordName("data")
#
#
# @final
# @dataclass(frozen=True, slots=True)
# class DataDir:
# val: Path
# replace: DataDirReplace = field(init=False)
#
# def __post_init__(self) -> None:
# setter = super().__setattr__
# setter("replace", DataDirReplace(self.val))
#
# @classmethod
# def from_org(cls, org: OrgData) -> Self:
# cls(DATA_ROOT.joinpath(org.org.val, org.app.val))
@final
@dataclass(frozen=True, slots=True)
class DestPaths:
# data_root = Record("data_root", str(DATA_ROOT))
data_root = ReplaceUnique.auto_format("data", str(DATA_ROOT))
data_path = ReplaceUnique(
data_root.src,
sep.join((data_root.src, Org.src.fmt, App.src.fmt)),
)
data_dir: Path
env_file: Path = field(init=False)
compose_file: ComposeFileRendered = field(init=False)
def __post_init__(self) -> None:
setter = super(DestPaths, self).__setattr__
path_join = self.data_dir.joinpath
setter("env_file", path_join(".env"))
setter("compose_file", ComposeFileRendered(path_join("docker-compose.yml")))
@classmethod
def from_org(cls, org: OrgData) -> Self:
return cls.from_path(DATA_ROOT.joinpath(org.org.dest, org.app.dest))
@classmethod
def from_path(cls, path: Path) -> Self:
return cls(path)
# def mk_compose_dir(self) -> None:
# folder = self.data_dir
# if folder.exists():
# return
# folder.mkdir(parents=True)
@property
def pre_render_funcs(self) -> Iterator[Callable[[str], str]]:
yield self.data_path
@property
def render_funcs(self) -> Iterator[Callable[[str], str]]:
yield self.data_root

View File

@@ -1,56 +0,0 @@
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, ReplaceStatic
@final
@dataclass(frozen=True, slots=True)
class Pswd(ReplaceStatic):
src = ReplaceDynamic("pswd")
@final
@dataclass
class Env:
pswd = 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]]:
p = self.pswd
for k, v in self.data.items():
if self.pswd.src.fmt not in v:
yield k, v
continue
yield k, p(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,59 +0,0 @@
from collections.abc import Iterator
from dataclasses import dataclass
from typing import Callable, Self, final, override
from docker_compose.cfg.org_yaml import OrgDataYaml
from docker_compose.cfg.replace import ReplaceDynamic, ReplaceStatic, ReplaceUnique
@final
@dataclass(frozen=True, slots=True)
class Org(ReplaceStatic):
src = ReplaceDynamic("org")
@final
@dataclass(frozen=True, slots=True)
class App(ReplaceStatic):
src = ReplaceDynamic("name")
@final
@dataclass(frozen=True, slots=True)
class Url(ReplaceStatic):
src = ReplaceDynamic("url")
@property
@override
def dest(self) -> str:
val = super(Url, self).dest
if not val:
return val
return ".".join((val, "ccamper7", "net"))
@final
@dataclass(frozen=True, slots=True)
class OrgData:
org_app = ReplaceUnique.build_placeholder('dn', Org, App)
app: App
org: Org
url: Url
@classmethod
def from_dict(cls, app: str, org: str, data: OrgDataYaml) -> Self:
return cls(
App(app),
Org(org),
Url(data.get("url")),
)
@property
def render_funcs(self) -> Iterator[Callable[[str], str]]:
yield self.app
yield self.org
yield self.url
@property
def pre_render_funcs(self) -> Iterator[Callable[[str], str]]:
yield self.org_app

View File

@@ -1,70 +0,0 @@
from collections.abc import Callable
from dataclasses import dataclass, field
from typing import ClassVar, Self, final
def format_src(src: str) -> str:
return f"${{_{src.upper()}}}"
@final
@dataclass(frozen=True, slots=True)
class ReplaceUnique:
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)
@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
@dataclass(frozen=True, slots=True)
class ReplaceDynamic:
val: str
fmt: str
@classmethod
def factory(cls, val:str):
return cls(val, format_src(val))
def __call__(self, string: str) -> str:
return string.replace(self.fmt, self.val)
# def __str__(self) -> str:
# return self.val if isinstance(self.val, str) else self.val.fmt
# def build_placeholder(self, *args: "ReplaceDynamic"):
# data = ((rep.val.upper(), rep.fmt) for rep in chain((self,), args))
# src: tuple[str, ...]
# dest: tuple[str, ...]
# src, dest = zip(*data)
# return ReplaceUnique("_".join(src), "_".join(dest))
@dataclass(frozen=True, slots=True)
class ReplaceStatic:
src: ClassVar[ReplaceDynamic]
_dest: None | str | Callable[[], str]
def __call__(self, string: str) -> str:
return string.replace(self.src.fmt, self.dest)
@property
def dest(self) -> str:
if not self._dest:
return ""
if isinstance(self._dest, str):
return self._dest
return self._dest()
@classmethod
def two_stage(cls, dest: str) -> tuple[Self, ReplaceDynamic]:
dest_var = ReplaceDynamic(dest)
return cls(dest_var.fmt), dest_var

View File

@@ -1,101 +0,0 @@
from collections.abc import Iterator
from dataclasses import dataclass
from pathlib import Path
from typing import Self, cast, final
from docker_compose.cfg.compose_paths import ServicePath, VolumePath
from docker_compose.cfg.org_yaml import OrgYaml
from docker_compose.util.Ts import TypeYamlCompatibleDict, TypeYamlCompatibleRes
from docker_compose.util.yaml_util import read_yaml, write_yaml
YAML_EXTS = frozenset((".yml", ".yaml"))
class ComposeFileTemplate(Path):
def write_dict(self, data: TypeYamlCompatibleDict) -> None:
write_yaml(data, self)
def write(self, data: str) -> None:
with self.open("wt") as f:
_ = f.write(data)
class OrgFile(Path):
@property
def as_dict(self) -> OrgYaml:
return cast(OrgYaml, cast(object, read_yaml(self)))
class YamlDir(Path):
@property
def yaml_files(self) -> Iterator[Path]:
if not self:
raise FileNotFoundError(self)
for service in self.iterdir():
if service.suffix not in YAML_EXTS:
continue
yield service
def __bool__(self) -> bool:
return self.exists()
class CfgDir(YamlDir):
@property
def cfg_file(self) -> OrgFile:
for file in self.yaml_files:
if file.stem != "cfg":
continue
return OrgFile(file)
raise FileNotFoundError(self.joinpath("cfg.y(a)ml"))
class ServiceDir(YamlDir):
@property
def files(self) -> Iterator[ServicePath]:
for file in self.yaml_files:
yield ServicePath(file)
class VolumesDir(YamlDir):
@property
def files(self) -> Iterator[VolumePath]:
try:
for file in self.yaml_files:
yield VolumePath(file)
except FileNotFoundError:
return
class VolumeData(Path):
def write(self, data: TypeYamlCompatibleRes) -> None:
write_yaml(data, self)
@final
@dataclass(frozen=True, slots=True)
class SrcPaths:
cfg_dir: CfgDir
org_file: OrgFile
env_file: Path
service_dir: ServiceDir
vol_dir: VolumesDir
compose_file: ComposeFileTemplate
volume_data: VolumeData
@classmethod
def from_path(cls, src: Path) -> Self:
cfg_dir = CfgDir(src)
return cls(
cfg_dir,
cfg_dir.cfg_file,
src.joinpath(".env"),
ServiceDir(src.joinpath("services")),
VolumesDir(src.joinpath("volumes")),
ComposeFileTemplate(src.joinpath("docker-compose.yml")),
VolumeData(src.joinpath("volume_paths.yml")),
)
@property
def app(self) -> str:
return self.cfg_dir.stem

View File

@@ -1,74 +0,0 @@
from collections.abc import Iterator
from dataclasses import dataclass
from pathlib import Path
from typing import Self, cast
from docker_compose.cfg.cfg_paths import CfgData
from docker_compose.cfg.org import OrgData
from docker_compose.cfg.src_path import SrcPaths
from docker_compose.compose.compose_yaml import ComposeYaml
from docker_compose.compose.net import Net
from docker_compose.compose.services import Service
from docker_compose.compose.volume_yaml import VolYaml
from docker_compose.util.Ts import TypeYamlCompatibleDict
from docker_compose.util.yaml_util import to_yaml
@dataclass(slots=True)
class Compose:
cfg: CfgData
services: frozenset[Service]
networks: Net
volumes: dict[str, VolYaml]
# replace_args: ReplaceArgs
@classmethod
def from_path(cls, path: Path) -> Iterator[Self]:
return cls.from_src_path(SrcPaths.from_path(path))
@classmethod
def from_src_path(cls, src_paths: SrcPaths) -> Iterator[Self]:
for cfg in CfgData.from_src_paths(src_paths):
yield cls.from_cfg(cfg)
@classmethod
def from_cfg(cls, cfg_data: CfgData) -> Self:
# services = {
# path.stem: Service.from_dict(path, data)
# for path, data in cfg_data.compose_paths.services_k_v
# }
services = frozenset(
Service.from_path(path) for path in cfg_data.compose_paths.services
)
return cls(
cfg_data,
services,
Net.from_service_list(services),
dict(cfg_data.compose_paths.volumes_k_v),
)
# @property
# def app(self) -> str:
# return self.cfg.src_paths.app
@property
def as_dict(self) -> ComposeYaml:
return ComposeYaml(
name=OrgData.org_app.dest,
services={
service.service_name: service.as_dict for service in self.services
},
networks=self.networks.as_dict,
volumes=self.volumes,
)
@property
def as_template(self) -> str:
data = cast(TypeYamlCompatibleDict, cast(object, self.as_dict))
return self.cfg.pre_render(to_yaml(data))
def write_template(self):
self.cfg.src_paths.compose_file.write(self.as_template)
def __call__(self):
self.write_template()

View File

@@ -1,12 +0,0 @@
from typing import TypedDict
from docker_compose.compose.net_yaml import NetYaml
from docker_compose.compose.services_yaml import ServiceYamlWrite
from docker_compose.compose.volume_yaml import VolYaml
class ComposeYaml(TypedDict):
name: str
services: dict[str, ServiceYamlWrite]
networks: NetYaml | None
volumes: dict[str, VolYaml] | None

View File

@@ -1,62 +0,0 @@
from collections.abc import Iterator
from dataclasses import dataclass
from pathlib import Path
from typing import final, override
from docker_compose.cfg import ROOT
from docker_compose.compose.compose import Compose
@final
@dataclass(slots=True)
class Rendered(Compose):
# @property
# def org(self) -> str:
# return self.cfg.org_data.org
@property
def bind_vols(self) -> Iterator[Path]:
root = str(ROOT)
for app in self.services:
for vol in app.volumes:
path = self.cfg.render_all(vol.split(":", 1)[0])
if not path.startswith(root):
continue
path = Path(path)
if path.is_file():
continue
yield path
# @property
# def missing_bind_vols(self) -> Iterator[Path]:
# for path in self.bind_vols:
# if path.exists():
# continue
# yield path
def mk_bind_vols(self) -> None:
for path in self.bind_vols:
path.mkdir(parents=True, exist_ok=True)
def write_bind_vols(self) -> None:
self.cfg.src_paths.volume_data.write(map(str, self.bind_vols))
@property
def proxy_nets(self) -> Iterator[str]:
for net in self.networks.proxys:
yield self.cfg.render_all(net)
@property
def as_rendered(self) -> str:
return self.cfg.render(self.as_template)
def write(self) -> None:
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

@@ -0,0 +1,4 @@
from docker_compose.util.replace import Replace
DN = Replace.build_placeholder("dn", "org", "name")
FQDN = Replace.build_placeholder("fqdn", "org", "name", "service")

View File

@@ -0,0 +1,12 @@
from typing import TypedDict
from docker_compose.compose_data.net_yaml import NetYaml
from docker_compose.compose_data.services_yaml import ServiceYamlWrite
from docker_compose.compose_data.volume_yaml import VolYaml
class ComposeYaml(TypedDict):
name: str
services: dict[str, ServiceYamlWrite]
networks: NetYaml | None
volumes: dict[str, VolYaml] | None

View File

@@ -0,0 +1,64 @@
from collections.abc import Iterable, Iterator
from dataclasses import dataclass
from pathlib import Path
from typing import Self, cast, final, override
from docker_compose.compose_data import DN
from docker_compose.compose_data.compose_yaml import ComposeYaml
from docker_compose.compose_data.net import Net
from docker_compose.compose_data.service import Service
from docker_compose.compose_data.src_paths import SrcPaths
from docker_compose.compose_data.volume_yaml import VolYaml
from docker_compose.util.replace import Replace
from docker_compose.util.yaml_util import read_yaml, to_yaml
@final
@dataclass(slots=True)
class ComposeData:
name: str
services: dict[str, Service]
networks: Net
volumes: dict[str, VolYaml]
@override
def __str__(self) -> str:
rep = Replace.format_src("name", self.name)
return rep(to_yaml(self.as_dict)) # pyright: ignore[reportArgumentType]
@staticmethod
def get_services(paths: Iterable[Path]) -> Iterator[tuple[str, Service]]:
for path in paths:
service = Service.from_path(path)
yield service.service_name, service
@staticmethod
def get_volumes(paths: Iterable[Path]) -> Iterator[tuple[str, VolYaml]]:
for path in paths:
yield path.stem, cast(VolYaml, cast(object, read_yaml(path)))
@classmethod
def from_path(cls, path: Path) -> Self:
return cls.from_src_paths(SrcPaths.from_path(path))
@classmethod
def from_src_paths(cls, src_paths: SrcPaths) -> Self:
services = dict(cls.get_services(src_paths.service_files))
return cls(
src_paths.app_name,
services,
Net.from_service_list(services.values()),
dict(cls.get_volumes(src_paths.volume_files)),
)
@property
def as_dict(self) -> ComposeYaml:
return ComposeYaml(
name=DN.dest,
services={
service.service_name: service.as_dict
for service in self.services.values()
},
networks=self.networks.as_dict,
volumes=self.volumes,
)

View File

@@ -0,0 +1,17 @@
from dataclasses import dataclass
from pathlib import Path
from typing import Self, final
@final
@dataclass(frozen=True, slots=True)
class DestPaths:
compose_file: Path
bind_vol_path: Path
@classmethod
def from_path(cls, src: Path) -> Self:
return cls(
src.joinpath("docker-compose.yml"),
src.joinpath("bind_vols.yml"),
)

View File

@@ -0,0 +1,28 @@
from dataclasses import dataclass
from pathlib import Path
from typing import Self, final, override
from docker_compose.compose_data.data import ComposeData
from docker_compose.compose_data.dest_paths import DestPaths
@final
@dataclass(frozen=True, slots=True)
class Template:
compose_data: ComposeData
dest_path: DestPaths
@classmethod
def from_path(cls, path: Path) -> Self:
return cls(
ComposeData.from_path(path),
DestPaths.from_path(path),
)
def __call__(self) -> None:
with self.dest_path.compose_file.open("wt") as f:
_ = f.write(str(self.compose_data))
@override
def __str__(self) -> str:
return str(self.compose_data)

View File

@@ -1,23 +1,27 @@
from collections.abc import Iterable, Iterator from collections.abc import Iterable, Iterator
from dataclasses import dataclass, field from dataclasses import dataclass
from typing import Self, final from typing import Self, final
from docker_compose.cfg.org import OrgData from docker_compose.compose_data.net_yaml import NetArgsYaml, NetYaml
from docker_compose.compose.net_yaml import NetArgsYaml, NetYaml from docker_compose.compose_data.service import Service
from docker_compose.compose.services import Service from docker_compose.util.replace import Replace
@final @final
@dataclass(frozen=True, slots=True) @dataclass(frozen=True, slots=True)
class NetArgs: class NetArgs:
name: str name: str
full_name: str = field(init=False) full_name: str
external: bool = field(init=False) external: bool
def __post_init__(self): @classmethod
setter = super(NetArgs, self).__setattr__ def factory(cls, name: str):
setter("full_name", f"{OrgData.org_app.dest}_{self.name}") f = Replace.build_placeholder("_", "org", "name").dest
setter("external", self.name == "proxy") return cls(
name,
f"{f}_{name}",
name == "proxy",
)
@property @property
def as_dict(self) -> NetArgsYaml: def as_dict(self) -> NetArgsYaml:
@@ -38,6 +42,9 @@ class NetArgs:
class Net: class Net:
data: frozenset[NetArgs] data: frozenset[NetArgs]
def __iter__(self) -> Iterator[NetArgs]:
yield from self.data
@classmethod @classmethod
def from_service_list(cls, args: Iterable[Service]) -> Self: def from_service_list(cls, args: Iterable[Service]) -> Self:
return cls.from_list( return cls.from_list(
@@ -46,7 +53,7 @@ class Net:
@classmethod @classmethod
def from_list(cls, args: frozenset[str]) -> Self: def from_list(cls, args: frozenset[str]) -> Self:
return cls(frozenset(NetArgs(arg) for arg in args)) return cls(frozenset(NetArgs.factory(arg) for arg in args))
@property @property
def as_dict(self) -> NetYaml: def as_dict(self) -> NetYaml:

View File

@@ -1,21 +1,19 @@
from dataclasses import dataclass, field from collections.abc import Callable, Iterator
from dataclasses import dataclass
from pathlib import Path
from typing import Self, final, override from typing import Self, final, override
from docker_compose.cfg.compose_paths import ServicePath, ServiceVal import yaml
from docker_compose.cfg.org import App, Org, OrgData, Url
from docker_compose.cfg.replace import ReplaceDynamic, ReplaceStatic, ReplaceUnique from docker_compose.compose_data import DN, FQDN
from docker_compose.compose.services_yaml import ( from docker_compose.compose_data.services_yaml import (
HealthCheck, HealthCheck,
ServiceYamlRead, ServiceYamlRead,
ServiceYamlWrite, ServiceYamlWrite,
) )
from docker_compose.util.Ts import T_Primitive from docker_compose.util.replace import Replace
from docker_compose.util.Ts import T_Primitive, TypeYamlDict
from docker_compose.util.yaml_util import validate_typed_dict
@final
@dataclass(frozen=True, slots=True)
class ServiceVal(ReplaceStatic):
src = ReplaceDynamic("service")
@final @final
@@ -23,23 +21,15 @@ class ServiceVal(ReplaceStatic):
class Service: class Service:
_traefik_labels = frozenset( _traefik_labels = frozenset(
( (
f"traefik.http.routers.{OrgData.org_app.dest}.rule=Host(`{Url.src}`)", f"traefik.http.routers.{DN.src}.rule=Host(`{Replace.fmt('url')}`)",
f"traefik.http.routers.{OrgData.org_app.dest}.entrypoints=websecure", f"traefik.http.routers.{DN.src}.entrypoints=websecure",
f"traefik.docker.network={OrgData.org_app.dest}_proxy", f"traefik.docker.network={DN.src}_proxy",
f"traefik.http.routers.{OrgData.org_app.dest}.tls.certresolver=le", f"traefik.http.routers.{DN.src}.tls.certresolver=le",
) )
) )
_sec_opts = frozenset(("no-new-privileges:true",)) _sec_opts = frozenset(("no-new-privileges:true",))
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_name: str
# service_val: ServiceVal
# fqdn: ReplaceUnique
command: tuple[str, ...] command: tuple[str, ...]
entrypoint: tuple[str, ...] entrypoint: tuple[str, ...]
environment: dict[str, T_Primitive] environment: dict[str, T_Primitive]
@@ -56,17 +46,24 @@ class Service:
healthcheck: HealthCheck | None healthcheck: HealthCheck | None
ports: frozenset[str] ports: frozenset[str]
def __post_init__(self):
setter = super(Service, self).__setattr__
setter("service_rep", ServiceVal(self.service_name))
@override @override
def __hash__(self) -> int: def __hash__(self) -> int:
return hash(self.service_name) return hash(self.service_name)
@classmethod @classmethod
def from_path(cls, path: ServicePath) -> Self: def from_path(cls, path: Path) -> Self:
return cls.from_dict(path.path.stem, path.as_dict) with path.open("rt") as f:
return cls.from_txt(path.stem, f.read())
@classmethod
def from_txt(cls, name: str, data_str: str) -> Self:
for func in cls.get_pre_render_funcs(name):
data_str = func(data_str)
data_dict: TypeYamlDict = yaml.safe_load(data_str) # pyright: ignore[reportAny]
# if not isinstance(data_dict, MutableMapping):
# raise TypeError
data = validate_typed_dict(ServiceYamlRead, data_dict)
return cls.from_dict(name, data) # pyright: ignore[reportArgumentType]
@classmethod @classmethod
def from_dict(cls, name: str, data: ServiceYamlRead) -> Self: def from_dict(cls, name: str, data: ServiceYamlRead) -> Self:
@@ -77,6 +74,7 @@ class Service:
return cls( return cls(
# service_val, # service_val,
name, name,
# Replace.format_src_dest("service", name),
tuple(data.get("command", ())), tuple(data.get("command", ())),
tuple(data.get("entrypoint", ())), tuple(data.get("entrypoint", ())),
data.get("environment", {}), data.get("environment", {}),
@@ -96,6 +94,12 @@ class Service:
frozenset(data.get("ports", ())), frozenset(data.get("ports", ())),
) )
@classmethod
def get_pre_render_funcs(cls, name: str) -> Iterator[Callable[[str], str]]:
yield DN
yield FQDN
yield Replace.format_src_dest("service", name)
@property @property
def as_dict(self) -> ServiceYamlWrite: def as_dict(self) -> ServiceYamlWrite:
return ServiceYamlWrite( return ServiceYamlWrite(
@@ -109,7 +113,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.fqdn.dest, container_name=f"{DN.dest}_{self.service_name}",
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

@@ -0,0 +1,90 @@
from collections.abc import Iterator
from dataclasses import dataclass
from pathlib import Path
from typing import Self, final
# class ComposeFileTemplate(Path):
# def write_dict(self, data: TypeYamlCompatibleDict) -> None:
# write_yaml(data, self)
#
# def write(self, data: str) -> None:
# with self.open("wt") as f:
# _ = f.write(data)
#
#
# class OrgFile(Path):
# @property
# def as_dict(self) -> OrgYaml:
# return cast(OrgYaml, cast(object, read_yaml(self)))
# class YamlDir(Path):
# @property
# def yaml_files(self) -> Iterator[Path]:
# if not self:
# raise FileNotFoundError(self)
# for service in self.iterdir():
# if service.suffix not in YAML_EXTS:
# continue
# yield service
#
# def __bool__(self) -> bool:
# return self.exists()
#
# class CfgDir(YamlDir):
# @property
# def cfg_file(self) -> OrgFile:
# for file in self.yaml_files:
# if file.stem != "cfg":
# continue
# return OrgFile(file)
# raise FileNotFoundError(self.joinpath("cfg.y(a)ml"))
#
#
# class ServiceDir(YamlDir):
# @property
# def files(self) -> Iterator[ServicePath]:
# for file in self.yaml_files:
# yield ServicePath(file)
#
#
# class VolumesDir(YamlDir):
# @property
# def files(self) -> Iterator[VolumePath]:
# try:
# for file in self.yaml_files:
# yield VolumePath(file)
# except FileNotFoundError:
# return
# class VolumeData(Path):
# def write(self, data: TypeYamlCompatibleRes) -> None:
# write_yaml(data, self)
@final
@dataclass(frozen=True, slots=True)
class SrcPaths:
YAML_EXTS = frozenset((".yml", ".yaml"))
app_name: str
service_files: frozenset[Path]
volume_files: frozenset[Path]
@classmethod
def from_path(cls, src: Path) -> Self:
return cls(
src.stem,
frozenset(cls.get_yaml_files(src.joinpath("services"))),
frozenset(cls.get_yaml_files(src.joinpath("volumes"))),
)
@classmethod
def get_yaml_files(cls, path: Path) -> Iterator[Path]:
for service in path.iterdir():
if service.suffix not in cls.YAML_EXTS:
continue
yield service

44
src/docker_compose/env/data.py vendored Normal file
View File

@@ -0,0 +1,44 @@
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, override
from docker_compose.util.replace import Replace
@final
@dataclass
class EnvData:
line_valid = re.compile(r"^\s*(\w+)\s*=\s*(.+)\s*$")
pswd = Replace.format_src("pswd", partial(secrets.token_urlsafe, 12))
data: dict[str, str]
@override
def __str__(self) -> str:
return "\n".join(sorted(map("=".join, self.with_pass)))
@classmethod
def get_lines(cls, path: Path) -> Iterator[tuple[str, str]]:
with path.open(mode="rt") as f:
for line in f:
res = cls.line_valid.match(line)
if not res:
continue
yield res.group(1), res.group(2)
@classmethod
def from_path(cls, path: Path) -> Self:
return cls({k: v for k, v in cls.get_lines(path)})
@property
def with_pass(self) -> Iterator[tuple[str, str]]:
p = self.pswd
for k, v in self.data.items():
if self.pswd.src not in v:
yield k, v
continue
yield k, p(v)

98
src/docker_compose/env/main.py vendored Normal file
View File

@@ -0,0 +1,98 @@
from collections.abc import Iterator
from dataclasses import dataclass
from pathlib import Path
from typing import Self, final
from docker_compose.env.data import EnvData
from docker_compose.org.data import OrgData
@final
@dataclass(frozen=True, slots=True)
class Env:
env: EnvData
org_data: OrgData
def __call__(self):
with self.org_data.dest.open("wt") as f:
_ = f.write(str(self.env))
@classmethod
def from_path(cls, path: Path, org: OrgData) -> Self:
return cls(
EnvData.from_path(path),
org,
)
@property
def org(self) -> str:
return self.org_data.org.dest
@property
def app(self) -> str:
return self.org_data.app.dest
@final
@dataclass(frozen=True, slots=True)
class EnvByOrg:
data: dict[str, Env]
app: str
def __call__(self):
for obj in self:
obj()
def __iter__(self) -> Iterator[Env]:
yield from self.data.values()
@classmethod
def _from_path_sub(cls, path: Path) -> Iterator[tuple[str, Env]]:
env_data = EnvData.from_path(path)
for org in OrgData.from_path(path):
env = Env(env_data, org)
yield env.org, env
@classmethod
def from_path(cls, path: Path) -> Self:
return cls(
dict(cls._from_path_sub(path)),
OrgData.get_app(path),
)
#
# @property
# def app(self) -> str:
# return self.env.org_data.app.dest
@final
@dataclass(frozen=True, slots=True)
class EnvByApp:
data: dict[str, EnvByOrg]
def __iter__(self) -> Iterator[EnvByOrg]:
yield from self.data.values()
def __call__(self) -> None:
for obj in self:
obj()
@staticmethod
def _get_folders(path_: Path) -> Iterator[Path]:
for path in path_.iterdir():
if not path.is_dir():
continue
if path.stem == "traefik":
continue
yield path
@classmethod
def _from_path_sub(cls, path_: Path) -> Iterator[tuple[str, EnvByOrg]]:
for path in cls._get_folders(path_):
by_org = EnvByOrg.from_path(path)
yield by_org.app, by_org
@classmethod
def from_path(cls, path: Path) -> Self:
return cls(dict(cls._from_path_sub(path)))

0
README.md → src/docker_compose/org/__init__.py Executable file → Normal file
View File

View File

@@ -0,0 +1,45 @@
from collections.abc import Iterator
from dataclasses import dataclass
from pathlib import Path
from typing import Callable, Self, cast, final
from docker_compose import APP_ROOT
from docker_compose.org.org_yaml import OrgDataYaml, OrgYaml
from docker_compose.util.replace import Replace
from docker_compose.util.yaml_util import read_yaml
@final
@dataclass(frozen=True, slots=True)
class OrgData:
app: Replace
org: Replace
url: Replace
dest: Path
@classmethod
def from_dict(cls, app: str, org: str, data: OrgDataYaml) -> Self:
url = data.get("url")
return cls(
Replace.format_src("name", app),
Replace.format_src("org", org),
Replace.format_src(
"url", ".".join((url, "ccamper7", "net")) if url else None
),
APP_ROOT.joinpath(org, app),
)
@classmethod
def get_app(cls, path: Path) -> str:
return path.stem
@classmethod
def from_path(cls, path: Path) -> Iterator[Self]:
app = cls.get_app(path)
for org, data in cast(OrgYaml, cast(object, read_yaml(path))).items():
yield cls.from_dict(app, org, data)
def __iter__(self) -> Iterator[Callable[[str], str]]:
yield self.app
yield self.org
yield self.url

View File

@@ -0,0 +1,165 @@
from collections.abc import Iterator
from dataclasses import dataclass
from itertools import chain
from pathlib import Path
from typing import Self, final, override
from docker_compose import APP_ROOT, ROOT, TEMPLATE_ROOT
from docker_compose.compose_data.main import Template
from docker_compose.org.data import OrgData
from docker_compose.util.replace import Replace
from docker_compose.util.yaml_util import write_yaml
@final
@dataclass(frozen=True, slots=True)
class BindVols:
# data_rep = Replace("data", str(DATA_ROOT))
render: "Render"
def __call__(self):
# def mk_bind_vols(self) -> None:
for path in self:
path.mkdir(parents=True, exist_ok=True)
def __iter__(self) -> Iterator[Path]:
root = str(ROOT)
for app in self.render.template.compose_data.services.values():
for vol in app.volumes:
path = self.render.render(vol.split(":", 1)[0])
if not path.startswith(root):
continue
path = Path(path)
if not path.is_dir():
continue
yield path
@final
@dataclass(frozen=True, slots=True)
class Render:
data_rep = Replace("data", str(APP_ROOT))
template: Template
org_data: OrgData
@override
def __str__(self) -> str:
return self.render(str(self.template))
def __call__(self):
self.write(str(self))
@property
def bind_vols(self) -> BindVols:
return BindVols(self)
def render(self, txt: str) -> str:
for func in chain((self.data_rep,), self.org_data):
txt = func(txt)
return txt
@property
def proxy_nets(self) -> Iterator[str]:
for net in self.template.compose_data.networks:
if not net.external:
continue
yield self.render(net.full_name)
def write(self, data: str, render:bool=False):
with self.org_data.dest.open("wt") as f:
_ = f.write(self.render(data) if render else data)
@final
@dataclass(frozen=True, slots=True)
class RenderByOrg:
template: Template
renders: dict[str, Render]
def __iter__(self) -> Iterator[Render]:
yield from self.renders.values()
# yield render
def __call__(self) -> None:
self.template()
for render in self:
render()
self.write_bind_vol_data()
def write_bind_vol_data(self):
write_yaml(self.vols, self.template.dest_path.bind_vol_path)
def __getitem__(self, key: str) -> Render:
return self.renders[key]
def __bool__(self) -> bool:
return bool(self.renders)
@staticmethod
def from_path_sub(template: Template, path: Path) -> Iterator[tuple[str, Render]]:
for org in OrgData.from_path(path):
yield org.org.dest, Render(template, org)
@classmethod
def from_path(cls, path: Path) -> Self:
template = Template.from_path(path)
return cls(
template,
dict(cls.from_path_sub(template, path)),
)
@property
def app(self):
return self.template.compose_data.name
@property
def vols(self) -> Iterator[str]:
for render in self:
for path in render.bind_vols:
yield str(path)
@property
def proxy_nets(self) -> Iterator[str]:
for render in self:
yield from render.proxy_nets
@final
@dataclass(frozen=True, slots=True)
class RenderByApp:
renders: dict[str, RenderByOrg]
def __iter__(self) -> Iterator[RenderByOrg]:
yield from self.renders.values()
def __call__(self) -> None:
for obj in self:
obj()
@staticmethod
def _get_folders(path_: Path) -> Iterator[Path]:
for path in path_.iterdir():
if not path.is_dir():
continue
if path.stem == "traefik":
continue
yield path
@classmethod
def _from_path_sub(cls, path_: Path) -> Iterator[tuple[str, RenderByOrg]]:
for path in cls._get_folders(path_):
by_org = RenderByOrg.from_path(path)
yield by_org.app, by_org
@classmethod
def from_path(cls, path: Path) -> Self:
return cls(dict(cls._from_path_sub(path)))
@classmethod
def load_all(cls) -> Self:
return cls.from_path(TEMPLATE_ROOT)
@property
def proxy_nets(self) -> Iterator[str]:
for render in self:
yield from render.proxy_nets

View File

@@ -17,7 +17,7 @@ type TypePrim = T_Primitive | _PrimIters | TypePrimDict
# type T_TDict = MutableMapping[T_Primitive, T_Prim] # type T_TDict = MutableMapping[T_Primitive, T_Prim]
# data going to and from yaml # data going to and from YAML
type TypeYaml = T_Primitive | TypeYamlRes type TypeYaml = T_Primitive | TypeYamlRes
type TypeYamlRes = list[TypeYaml] | TypeYamlDict type TypeYamlRes = list[TypeYaml] | TypeYamlDict
class TypeYamlDict(Protocol): class TypeYamlDict(Protocol):

View File

@@ -0,0 +1,92 @@
from collections.abc import Callable
from dataclasses import dataclass
from typing import Self, final
type TypeDest = str | None | Callable[[], str]
@final
@dataclass(frozen=True, slots=True)
class Replace:
src: str
_dest: TypeDest
def __call__(self, string: str) -> str:
return string.replace(self.src, self.dest)
@classmethod
def format_src(cls, src: str, dest: TypeDest):
return cls(cls.fmt(src), dest)
@classmethod
def format_src_dest(cls, src: str, dest: str):
return cls(cls.fmt(src), cls.fmt(dest))
@classmethod
def from_str(cls, src: str) -> Self:
return cls.format_src(src, src)
@classmethod
def build_placeholder(cls, src: str, *dest: str) -> Self:
return cls.format_src(
src,
"_".join(map(cls.fmt, dest)),
)
@property
def dest(self) -> str:
if not self._dest:
return ""
if isinstance(self._dest, str):
return self._dest
return self._dest()
@staticmethod
def fmt(src: str) -> str:
return f"${{_{src.upper()}}}"
#
# @final
# @dataclass(frozen=True, slots=True)
# class ReplaceDynamic:
# val: str
# fmt: str
#
# @classmethod
# def factory(cls, val: str):
# return cls(val, format_src(val))
#
# def __call__(self, string: str) -> str:
# return string.replace(self.fmt, self.val)
#
# # def __str__(self) -> str:
# # return self.val if isinstance(self.val, str) else self.val.fmt
# # def build_placeholder(self, *args: "ReplaceDynamic"):
# # data = ((rep.val.upper(), rep.fmt) for rep in chain((self,), args))
# # src: tuple[str, ...]
# # dest: tuple[str, ...]
# # src, dest = zip(*data)
# # return ReplaceUnique("_".join(src), "_".join(dest))
#
#
# @dataclass(frozen=True, slots=True)
# class ReplaceStatic:
# src: ClassVar[ReplaceDynamic]
# _dest: None | str | Callable[[], str]
#
# def replace(self, string: str) -> str:
# return string.replace(self.src.fmt, self.dest)
#
# @property
# def dest(self) -> str:
# if not self._dest:
# return ""
# if isinstance(self._dest, str):
# return self._dest
# return self._dest()
#
# # @classmethod
# # def two_stage(cls, dest: str) -> tuple[Self, ReplaceDynamic]:
# # dest_var = ReplaceDynamic(dest)
# # return cls(dest_var.fmt), dest_var

View File

@@ -1,5 +1,5 @@
import re import re
from collections.abc import Iterator, MutableMapping, Sequence, Set from collections.abc import Iterator, MutableMapping, Set
from pathlib import Path from pathlib import Path
from typing import ( from typing import (
cast, cast,
@@ -48,7 +48,7 @@ class VerboseSafeDumper(yaml.SafeDumper):
def yaml_prep(data: TypeYamlCompatibleRes) -> TypeYamlCompatibleRes: 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, tuple):
return tuple(list_prep(data)) return tuple(list_prep(data))
res = tuple(list_prep(data)) res = tuple(list_prep(data))
try: try:
@@ -110,26 +110,33 @@ def read_typed_yaml[T: TypeYamlDict](
path: Path, path: Path,
) -> T: ) -> T:
with path.open("rt") as f: with path.open("rt") as f:
data: TypeYamlDict = yaml.safe_load(f) # pyright: ignore[reportAny] data: T = yaml.safe_load(f) # pyright: ignore[reportAny]
path_to_typed(type_, data, path) return path_to_typed(type_, data, path)
return cast(T, data)
def path_to_typed( def path_to_typed[T: TypeYamlDict](
type_: type[TypeYamlDict], type_: type[T],
data: TypeYamlDict, data: T,
path: Path, path: Path,
) -> None: ) -> T:
try: try:
validate_typed_dict(type_, data) return validate_typed_dict(type_, data)
except (KeyError, TypeError) as e: except (KeyError, TypeError) as e:
e.add_note(f"path: {path!s}") e.add_note(f"path: {path!s}")
raise e raise e
def validate_typed_dict( def validate_typed_dict[T: TypeYamlDict](
t: type[TypeYamlDict], t: type[T],
data: TypeYamlDict, data: T,
) -> T:
_validate_typed_dict(t, data)
return cast(T, cast(object, data))
def _validate_typed_dict[T: TypeYamlDict](
t: type[T],
data: T,
) -> None: ) -> None:
keys = frozenset(data.keys()) keys = frozenset(data.keys())
missing = t.__required_keys__.difference(keys) missing = t.__required_keys__.difference(keys)
@@ -142,7 +149,7 @@ def validate_typed_dict(
for key, val in data.items(): for key, val in data.items():
t2 = cast(type, cast(object, hints[key])) t2 = cast(type, cast(object, hints[key]))
if is_typeddict(t2): if is_typeddict(t2):
validate_typed_dict(t2, cast(TypeYamlDict, val)) _validate_typed_dict(t2, cast(TypeYamlDict, val))
continue continue
# try: # try:
@@ -154,7 +161,6 @@ def validate_typed_dict(
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
# valid = isinstance(val, get_types(t2)) # valid = isinstance(val, get_types(t2))
# except TypeError: # except TypeError:
# valid = isinstance(val, get_origin(t2)) # valid = isinstance(val, get_origin(t2))