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

sync

Browse Source
This commit is contained in:
Christian Camper
2025-12-18 00:00:05 -06:00
parent ca23b44a25
commit fa4339768f
15 changed files with 181 additions and 138 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
.idea/.gitignore generated vendored Normal file
View File

@@ -0,0 +1,5 @@
# Default ignored files
/shelf/
/workspace.xml
# Editor-based HTTP Client requests
/httpRequests/

11
.idea/compose_gen_uv.iml generated Normal file
View File

@@ -0,0 +1,11 @@
<?xml version="1.0" encoding="UTF-8"?>
<module type="PYTHON_MODULE" version="4">
<component name="NewModuleRootManager">
<content url="file://$MODULE_DIR$">
<sourceFolder url="file://$MODULE_DIR$/src" isTestSource="false" />
<excludeFolder url="file://$MODULE_DIR$/.venv" />
</content>
<orderEntry type="jdk" jdkName="Python 3.13 (compose_gen_uv)" jdkType="Python SDK" />
<orderEntry type="sourceFolder" forTests="false" />
</component>
</module>

7
.idea/dictionaries/project.xml generated Normal file
View File

@@ -0,0 +1,7 @@
<component name="ProjectDictionaryState">
<dictionary name="project">
<words>
<w>traefik</w>
</words>
</dictionary>
</component>

10
.idea/inspectionProfiles/Project_Default.xml generated Normal file
View File

@@ -0,0 +1,10 @@
<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="PyMissingTypeHintsInspection" enabled="true" level="WEAK WARNING" enabled_by_default="true">
<option name="m_onlyWhenTypesAreKnown" value="false" />
</inspection_tool>
<inspection_tool class="PyUnnecessaryCastInspection" enabled="true" level="WEAK WARNING" enabled_by_default="true" />
</profile>
</component>

6
.idea/inspectionProfiles/profiles_settings.xml generated Normal file
View File

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

7
.idea/misc.xml generated Normal file
View File

@@ -0,0 +1,7 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="Black">
<option name="sdkName" value="Python 3.13 (compose_gen_uv)" />
</component>
<component name="ProjectRootManager" version="2" project-jdk-name="Python 3.13 (compose_gen_uv)" project-jdk-type="Python SDK" />
</project>

8
.idea/modules.xml generated Normal file
View File

@@ -0,0 +1,8 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="ProjectModuleManager">
<modules>
<module fileurl="file://$PROJECT_DIR$/.idea/compose_gen_uv.iml" filepath="$PROJECT_DIR$/.idea/compose_gen_uv.iml" />
</modules>
</component>
</project>

9
.idea/ruff.xml generated Normal file
View File

@@ -0,0 +1,9 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="RuffConfigService">
<option name="globalRuffExecutablePath" value="/opt/pycharm_venv/bin/ruff" />
<option name="runRuffOnSave" value="true" />
<option name="useRuffImportOptimizer" value="true" />
<option name="useRuffServer" value="true" />
</component>
</project>

6
.idea/vcs.xml generated Normal file
View File

@@ -0,0 +1,6 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="VcsDirectoryMappings">
<mapping directory="" vcs="Git" />
</component>
</project>

35
src/compose/__init__.py
View File

@@ -1,35 +0,0 @@
from collections.abc import Iterable, Iterator
from compose.cfg import CFG_ROOT, TRAEFIK_PATH
from compose.cfg.factory import cfg_data_factory
from compose.compose.factory import compose_factory
from compose.rendered.entity import Rendered
from compose.rendered.factory import rendered_factory
from compose.rendered.util import write
from compose.src_path.entity import src_paths_factory
from compose.template.factory import template_factory
def load_all() -> Iterable[Rendered]:
for dir in CFG_ROOT.iterdir():
paths = src_paths_factory(dir)
cfg = cfg_data_factory(paths)
parsed = compose_factory(cfg)
for template in template_factory(parsed):
yield rendered_factory(template)
def render_all() -> Iterator[Rendered]:
for rendered in load_all():
write(rendered)
yield rendered
if __name__ == "__main__":
renders = render_all()
src_paths = src_paths_factory(TRAEFIK_PATH)
cfg_data = cfg_data_factory(src_paths)
traefik = compose_factory(cfg_data)
for template in template_factory(traefik):
rendered = rendered_factory(template)
write(rendered)

26
src/compose/cfg/__init__.py
View File

@@ -1,26 +0,0 @@
from collections.abc import Mapping
from pathlib import Path
type nested_list = list[str | nested_list]
type T_Primitive = bool | int | str
type T_PrimVal = T_Primitive | list[T_Primitive] | T_PrimDict
type T_PrimDict = Mapping[T_Primitive, T_PrimVal]
type T_YamlVals = T_Primitive | list[T_Primitive | T_YamlDict] | T_YamlDict
type T_YamlDict = Mapping[str, T_YamlVals]
CFG_ROOT = Path("/data/cfg")
DATA_ROOT = Path("/data")
TRAEFIK_PATH = Path("/data/traefik")
# TCo_YamlVals = TypeVar(
# "TCo_YamlVals",
# bound=T_Primitive | list[T_Primitive | T_YamlDict] | T_YamlDict,
# covariant=True,
# )
# type TCo_YamlDict = dict[str, TCo_YamlVals]
# TCo_YamlDict = TypeVar("TCo_YamlDict", bound=dict[str, T_YamlVals], covariant=True)
# class HasServices(TypedDict):
# services: dict[str, ComposeService]

77
src/compose/util.py
View File

@@ -1,77 +0,0 @@
import re
from collections.abc import KeysView, Mapping
from pathlib import Path
from typing import Any, ClassVar, Protocol, cast, override
import yaml
from compose.cfg import T_PrimDict, T_Primitive, T_PrimVal, T_YamlDict
class VerboseSafeDumper(yaml.SafeDumper):
@override
def ignore_aliases(self, data: Any) -> bool: # pyright: ignore[reportExplicitAny, reportAny]
return True
def merge_dicts[T: Mapping[Any, Any]](dict1: T, dict2: T) -> T:
def _merge_dicts(dict1: T_PrimDict, dict2: T_PrimDict):
s1 = frozenset(dict1.keys())
s2 = frozenset(dict2.keys())
for k in s1.difference(s2):
yield k, dict1[k]
for k in s2.difference(s1):
yield k, dict2[k]
for k in s1.intersection(s2):
v1 = dict1[k]
v2 = dict2[k]
if isinstance(v1, dict) and isinstance(v2, dict):
yield k, dict[T_Primitive, T_PrimVal](_merge_dicts(v1, v2))
continue
if isinstance(v1, list) and isinstance(v2, list):
yield k, list(frozenset(v1).union(v2))
continue
raise Exception("merge error")
return cast(T, dict(_merge_dicts(dict1, dict2)))
def read_yml(path: Path) -> T_YamlDict:
with path.open("rt") as f:
return cast(T_YamlDict, yaml.safe_load(f))
def to_yaml(data: T_YamlDict) -> str:
_yaml = yaml.dump(data, Dumper=VerboseSafeDumper)
return re.sub(r"(^\s*-)", r" \g<1>", _yaml, flags=re.MULTILINE)
def get_replace_name(name: str) -> str:
return f"${{_{name.upper()}}}"
class T_TypedDict(Protocol):
__required_keys__: ClassVar[frozenset[str]]
def keys(self) -> KeysView[str]: ...
def validate_typed_dict(
typed_dict: type[T_TypedDict],
data: T_TypedDict,
path: Path | None = None,
pre: tuple[str, ...] | None = None,
) -> None:
req = typed_dict.__required_keys__.difference(data.keys())
if not req:
return
if pre is None:
keys = (f'"{key}"' for key in req)
else:
key_pre = ".".join(pre)
keys = (f'"{key_pre}.{key}"' for key in req)
msg = f"key(s) ({', '.join(keys)}) not found"
if path is not None:
msg = f"{msg} in file {path!s}"
print(msg)
raise KeyError

21
src/docker_compose/__init__.py Normal file
View File

@@ -0,0 +1,21 @@
from collections.abc import Iterable, Iterator
from compose.cfg import CFG_ROOT, TRAEFIK_PATH
from compose.compose.render import Rendered
def load_all() -> Iterable[Rendered]:
for _dir in CFG_ROOT.iterdir():
yield Rendered.from_path(_dir)
def render_all() -> Iterator[str]:
for rendered in load_all():
rendered.write_all()
yield from rendered.proxy_nets
if __name__ == "__main__":
# renders = render_all()
nets = frozenset(render_all())
traefik = Rendered.from_path(TRAEFIK_PATH)

5
src/docker_compose/cfg/__init__.py Normal file
View File

@@ -0,0 +1,5 @@
from pathlib import Path
CFG_ROOT = Path("/data/cfg")
DATA_ROOT = Path("/data")
TRAEFIK_PATH = Path("/data/traefik")

86
src/docker_compose/util.py Normal file
View File

@@ -0,0 +1,86 @@
from collections.abc import Mapping
from typing import Any, cast
from compose.Ts import T_PrimDict, T_Primitive, T_PrimVal
def merge_dicts[T: Mapping[Any, Any]](dict1: T, dict2: T) -> T:
def _merge_dicts(_dict1: T_PrimDict, _dict2: T_PrimDict):
s1 = frozenset(_dict1.keys())
s2 = frozenset(_dict2.keys())
for k in s1.difference(s2):
yield k, _dict1[k]
for k in s2.difference(s1):
yield k, _dict2[k]
for k in s1.intersection(s2):
v1 = _dict1[k]
v2 = _dict2[k]
if isinstance(v1, dict) and isinstance(v2, dict):
yield k, dict[T_Primitive, T_PrimVal](_merge_dicts(v1, v2))
continue
if isinstance(v1, list) and isinstance(v2, list):
yield k, list(frozenset(v1).union(v2))
continue
raise Exception("merge error")
return cast(T, dict(_merge_dicts(dict1, dict2)))
# class T_TypedDict(Protocol):
# __required_keys__: ClassVar[frozenset[str]]
# def keys(self) -> KeysView[str]: ...
# def read_yml(path: Path):
# with path.open("rt") as f:
# return yaml.safe_load(f)
# def to_yaml(data: T_YamlDict) -> str:
# _yaml = yaml.dump(data, Dumper=VerboseSafeDumper)
# return re.sub(r"(^\s*-)", r" \g<1>", _yaml, flags=re.MULTILINE)
# def get_replace_name(name: str) -> str:
# return f"${{_{name.upper()}}}"
# def validate_typed_dict(
# # typed_dict: type[T_TypedDict],
# data: T_TypedDict,
# path: Path | None = None,
# pre: tuple[str, ...] | None = None,
# ) -> None:
# req = type(data).__required_keys__.difference(data.keys())
# if not req:
# return
# if pre is None:
# keys = (f'"{key}"' for key in req)
# else:
# key_pre = ".".join(pre)
# keys = (f'"{key_pre}.{key}"' for key in req)
# msg = f"key(s) ({', '.join(keys)}) not found"
# if path is not None:
# msg = f"{msg} in file {path!s}"
# print(msg)
# raise KeyError
# def to_typed_dict[T:T_TypedDict](typed_dict:type[T] ,data: Mapping[str, Any]) -> T:
# missing = typed_dict.__required_keys__.difference(data)
# if missing:
# msg = f"key(s) ({', '.join(map("{}".format, missing))}) not found"
# raise KeyError(msg)
# _dict = typed_dict()
# for key in typed_dict.__required_keys__:
# val = data[key]
# if not isinstance(val, typed_dict.__annotations__[key]):
# msg = f'invalid type for {type(data).__name__}[{key}]\nexpected {typed_dict.__annotations__[key]} got {type(val).__name__}'
# raise TypeError()
# _dict[key] = val
# for key, key_type in BackupData.__annotations__.items():
# if key not in data:
# raise ValueError(f"Key: {key} is not available in data.")
# result[key] = key_type(data[key])
# return result