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-14 18:14:07 -06:00
parent 9d45d5db88
commit a0b6e00a31
38 changed files with 969 additions and 932 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
src/cfg/__init__.py Normal file
View File

@@ -0,0 +1,26 @@
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]

32
src/cfg/entity.py Normal file
View File

@@ -0,0 +1,32 @@
from dataclasses import dataclass
from pathlib import Path
from typing import NotRequired, TypedDict, final
from src_path.entity import SrcPaths
class OrgDataYaml(TypedDict):
org: NotRequired[str]
url: NotRequired[str]
class CfgDataYaml(TypedDict):
name: str
files: list[str]
orgs: list[OrgDataYaml]
@final
@dataclass(frozen=True, slots=True)
class OrgData:
org: str | None
url: str | None
@final
@dataclass(frozen=True, slots=True)
class CfgData:
src_paths: SrcPaths
name: str
files: frozenset[Path]
orgs: frozenset[OrgData] | None

18
src/cfg/factory.py Normal file
View File

@@ -0,0 +1,18 @@
from typing import cast
from cfg.entity import CfgData, CfgDataYaml
from cfg.get import cfg_get_orgs
from src_path.entity import SrcPaths
from src_path.get import src_path_get_files
from util import read_yml
def cfg_data_factory(cfg_dir: SrcPaths) -> CfgData:
data = cast(CfgDataYaml, read_yml(cfg_dir.cfg_file))
return CfgData(
cfg_dir,
data["name"],
frozenset(src_path_get_files(cfg_dir, data)),
frozenset(cfg_get_orgs(data)),
)

20
src/cfg/get.py Normal file
View File

@@ -0,0 +1,20 @@
from collections.abc import Iterator
from typing import cast
from cfg.entity import CfgData, CfgDataYaml, OrgData
from service.entity import Service, ServiceYaml
from util import read_yml
def cfg_get_orgs(data: CfgDataYaml) -> Iterator[OrgData]:
for org_data in data["orgs"]:
yield OrgData(
org_data.get("org"),
org_data.get("url"),
)
def cfg_get_services(cfg_data: CfgData) -> Iterator[tuple[str, Service]]:
for path in cfg_data.files:
_dict = cast(ServiceYaml, read_yml(path))
yield path.stem, Service.from_dict(_dict)

250
src/compose.py
View File

@@ -1,250 +0,0 @@
from collections.abc import Collection, Iterator
from dataclasses import dataclass
from functools import reduce
from pathlib import Path
from shutil import copyfile
from typing import Literal, NotRequired, Protocol, cast, final
from Ts import (
CfgData,
Compose,
ComposeNet,
ComposeNetArgs,
HasServices,
OrgData,
TraefikComposeDict,
TraefikNet,
TraefikNetName,
)
from util import merge_dicts, read_yml, to_yaml
from val_objs import RecordCls
# CFG_ROOT = Path("/data/cfg")
# def replace(rec: RecordCls, string: str) -> str:
# return str.replace(string, rec.name, rec.val.to_str())
# @final
# @dataclass(frozen=True, slots=True)
# class ComposeBuild:
# root_path: Path
# # def __init__(self, path: str) -> None:
# # self.env_path = self.path.joinpath(".env")
# def compose_build_factory(root_path: str):
# path = Path("/data/cfg").joinpath(root_path)
# return ComposeBuild(path)
# CFG_PATH = Path("/data/cfg")
def compose_treafik(proxy_data: Iterator[tuple[str, str]]):
root_path = CFG_PATH.joinpath("treafik")
cfg_data = CfgData(
name="treafik",
files=["treafik.yml"],
orgs=[OrgData(org="util", url="treafik")],
)
services = build_compose(root_path, cfg_data)["services"]
networks: TraefikNet = dict()
for name, proxy in proxy_data:
networks[name] = TraefikNetName(name=proxy)
traefik_compose = TraefikComposeDict(
name="traefik",
services=services,
networks=networks,
)
mk_volumes = tuple(get_volumes(traefik_compose))
og_yaml = to_yaml(traefik_compose)
org = cfg_data["orgs"][0]
# def build_compose(compose_rw: ComposeRWData, cfg_data: CfgData) -> Compose:
# compose_dict = get_compose_dict(compose_rw, cfg_data)
# insert_defaults(compose_dict)
# insert_traefik_labels(compose_dict)
# return compose_dict
# def write_compose(write_data: ComposeData):
# write(write_data)
# mk_compose_dir(write_data)
# mk_compose_env(write_data.rw)
# def gen_compose(path: str) -> Iterator[tuple[str, str]]:
# # compose_build = compose_build_factory(path)
# root_path = CFG_PATH.joinpath(path)
# cfg_data = get_config_data(root_path)
# compose_dict = build_compose(root_path, cfg_data)
# mk_volumes = tuple(get_volumes(compose_dict))
# proxy_net = set_networks(compose_dict)
# og_yaml = to_yaml(compose_dict)
# for _dict in cfg_data["orgs"]:
# compose_rw = compose_rw_factory(
# _dict.get("org"), cfg_data["name"], _dict["url"]
# )
# write(compose_rw, og_yaml)
# mk_compose_dir(compose_rw, mk_volumes)
# mk_compose_env(root_path, compose_rw.data_dir.to_path())
# if proxy_net is None:
# continue
# for net in proxy_net:
# def sub():
# for kv in net:
# yield replace(compose_rw.org_name, kv)
# yield tuple[str, str](sub())
# def get_config_data(cfg_dir: Path) -> CfgData:
# cfg_path = cfg_dir.joinpath("cfg.yml")
# return cast(CfgData, read_yml(cfg_path))
# def get_compose_dict(compose_rw: ComposeRWData, cfg_data: CfgData) -> Compose:
# def sub():
# for file in cfg_data["files"]:
# path = compose_rw.data_dir.joinpath(file)
# yield cast(Compose, read_yml(path))
# return reduce(merge_dicts, sub(), cast(Compose, {})) # pyright: ignore[reportInvalidCast]
# @final
# class TraefikBuild:
# def insert_defaults(data: Compose) -> None:
# data["name"] = "${_ORG_NAME}"
# for app_data in data["services"].values():
# app_data["restart"] = "unless-stopped"
# sec_opts = {"no-new-privileges:true"}
# app_data["security_opt"] = list(
# sec_opts.union(app_data.get("security_opt", set()))
# )
# def insert_traefik_labels(data: Compose) -> None:
# for app_data in data["services"].values():
# traefik_labels = {
# "traefik.http.routers.${_ORG_NAME}.rule=Host(`${_URL}`)",
# "traefik.http.routers.${_ORG_NAME}.entrypoints=websecure",
# "traefik.docker.network=${_ORG_NAME}_proxy",
# "traefik.http.routers.${_ORG_NAME}.tls.certresolver=le",
# }
# if "labels" not in app_data:
# continue
# if "traefik.enable=true" not in app_data["labels"]:
# continue
# app_data["labels"] = list(traefik_labels.union(app_data["labels"]))
# def get_volumes(data: HasServices) -> Iterator[str]:
# for app_data in data["services"].values():
# if "volumes" not in app_data:
# return
# for vol in app_data["volumes"]:
# if not vol.startswith(r"${_DATA}"):
# continue
# yield vol.split(":", 1)[0]
# def set_networks(data: Compose) -> tuple[str, str] | None:
# def sub() -> Iterator[Literal["proxy", "internal"]]:
# for app_data in data["services"].values():
# if "networks" not in app_data:
# continue
# yield from app_data["networks"]
# nets = set(sub())
# net = ComposeNet()
# if "proxy" in nets:
# proxy = ComposeNetArgs(name="${_ORG_NAME}_proxy", external=True)
# net["proxy"] = proxy
# ret = "${_ORG_NAME}", "${_ORG_NAME}_proxy"
# else:
# ret = None
# if "internal" in nets:
# net["internal"] = ComposeNetArgs(name="${_ORG_NAME}")
# if not net:
# return
# data["networks"] = net
# return ret
# def compose_rw_factory(org_raw: str | None, name_raw: str, sub_url: str):
# _org = OrgVal(org_raw)
# _name = NameVal(name_raw)
# # org = Record("ORG", _org)
# # name = Record("NAME", _name)
# org_name = (
# NameVal(f"{_org.to_str()}_{_name.to_str()}") if _org.is_valid() else _name
# )
# # org_name = Record("ORG_NAME", org_name)
# data_dir = Path("/data").joinpath(_org.to_str(), _name.to_str())
# # cfg_dir = CFG_PATH.joinpath(org_name.val.to_str())
# # data = Record("DATA", DataDir(data_dir))
# # url = Record("URL", Url(sub_url))
# # Path("/data").joinpath(_org.to_str(), _name.to_str())
# # compose_path = data_dir.to_path().joinpath("docker-compose.yml")
# # env_path = data_dir.to_path().joinpath(".env")
# return ComposeRWData(
# Record("ORG", _org),
# Record("NAME", _name),
# Record("ORG_NAME", org_name),
# Record("DATA", DataDir(data_dir)),
# Record("URL", Url(sub_url)),
# data_dir,
# CFG_PATH.joinpath(org_name.to_str()),
# )
# def __init__(self, org: str | None, name: str, sub_url: str) -> None:
# super().__init__()
# self._org = OrgVal(org)
# self._name = NameVal(name)
# self.org = Record("ORG", self._org)
# self.name = Record("NAME", self._name)
# org_name = (
# NameVal(f"{self.org.val.to_str()}_{self.name.val.to_str()}")
# if self._org.is_valid()
# else self._name
# )
# self.org_name = Record("ORG_NAME", org_name)
# self.data_dir = DataDir(
# self._org,
# self._name,
# )
# self.data = Record("DATA", self.data_dir)
# self.url = Record("URL", Url(sub_url))
# self.compose_path = self.data_dir.to_path().joinpath("docker-compose.yml")
# self.env_path = self.data_dir.to_path().joinpath(".env")
# def mk_compose_dir(compose_data: ComposeData) -> None:
# for vol in compose_data.volumes:
# mk_path = Path(replace(compose_data.rw.data, vol))
# if mk_path.exists():
# continue
# mk_path.mkdir(parents=True)
# def mk_compose_env(compose_data: ComposeRWData) -> None:
# root_env_path = compose_data.cfg_dir.joinpath(".env")
# dest_env_path = compose_data.data_dir.joinpath(".env")
# if root_env_path.exists() and not dest_env_path.exists():
# _ = copyfile(root_env_path, dest_env_path)
# def write(compose_data: ComposeData) -> None:
# string = reduce(lambda s, rec: replace(rec, s), compose_data.rw, compose_data.yaml)
# compose_path = compose_data.rw.data_dir.joinpath("docker-compose.yml")
# if not compose_path.exists():
# compose_path.parent.mkdir(parents=True)
# with compose_path.open("wt") as f:
# _ = f.write(string)

163
src/compose/compose_struct.py
View File

@@ -1,163 +0,0 @@
from collections.abc import Mapping
from dataclasses import dataclass, asdict
from typing import Literal, NotRequired, Self, TypedDict, final
from util import to_yaml
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 = dict[str, T_YamlVals]
class ComposeNetArgsYaml(TypedDict):
name: str
external: NotRequired[bool]
@final
@dataclass(frozen=True, slots=True)
class ComposeNetArgs:
name: str
external: bool | None
@classmethod
def from_dict(cls, data: ComposeNetArgsYaml) -> Self:
return cls(
data["name"],
data.get("external"),
)
class ComposeNetYaml(TypedDict):
internal: NotRequired[ComposeNetArgsYaml]
proxy: NotRequired[ComposeNetArgsYaml]
@final
@dataclass(frozen=True, slots=True)
class ComposeNet:
internal: ComposeNetArgs | None
proxy: ComposeNetArgs | None
@classmethod
def from_class(cls, data: ComposeNetYaml) -> Self:
internal = data.get("internal")
if internal is not None:
internal = ComposeNetArgs.from_dict(internal)
proxy = data.get("proxy")
if proxy is not None:
proxy = ComposeNetArgs.from_dict(proxy)
return cls(internal, proxy)
class ComposeServiceYaml(TypedDict):
command: NotRequired[list[str]]
container_name: str
entrypoint: list[str]
environment: NotRequired[dict[str, T_Primitive]]
image: str
labels: NotRequired[list[str]]
logging: dict[str, str]
networks: NotRequired[list[Literal["proxy", "internal"]]]
restart: str
security_opt: NotRequired[list[str]]
user: NotRequired[str]
volumes: NotRequired[list[str]]
@final
@dataclass(slots=True)
class ComposeService:
command: list[str] | None
container_name: str | None
entrypoint: list[str]
environment: dict[str, T_Primitive] | None
image: str
labels: set[str] | None
logging: dict[str, str] | None
networks: set[Literal["proxy", "internal"]] | None
restart: str | None
security_opt: set[str] | None
user: str | None
volumes: set[str] | None
@classmethod
def from_dict(cls, data: ComposeServiceYaml) -> Self:
labels = data.get("labels")
if labels is not None:
labels = set(labels)
sec = data.get("security_opt")
if sec is not None:
sec = set(sec)
vols = data.get("volumes")
if vols is not None:
vols = set(vols)
return cls(
data.get("command"),
None, # data['container_name'],
data["entrypoint"],
data.get("environment"),
data["image"],
labels,
None,
data.get("netwoks"),
None,
sec,
data.get("user"),
vols,
)
def is_valid(self) -> bool:
attrs = (self.container_name, self.restart, self.logging)
for attr in attrs:
if attr is None:
return False
return True
class ComposeYaml(TypedDict):
name: str
networks: NotRequired[ComposeNetYaml]
services: dict[str, ComposeServiceYaml]
@final
@dataclass(slots=True)
class Compose:
name: str
networks: ComposeNet | None
services: dict[str, ComposeService]
@classmethod
def from_dict(cls, data: ComposeYaml) -> Self:
services = dict[str, ComposeService]()
for k, v in data["services"].items():
services[k] = ComposeService.from_dict(v)
return cls(
data["name"],
None,
services,
)
def as_yaml(self) -> str:
return to_yaml(asdict(self))
class TraefikNetName(TypedDict):
name: str
type TraefikNet = dict[str, TraefikNetName]
class TraefikComposeDict(TypedDict):
name: str
networks: TraefikNet
services: dict[str, ComposeServiceYaml]
class HasServices(TypedDict):
services: dict[str, ComposeServiceYaml]

265
src/compose/entities.py
View File

@@ -1,265 +0,0 @@
from collections.abc import Iterator
from .compose_struct import Compose, ComposeService, ComposeServiceYaml, ComposeYaml
from dataclasses import dataclass
from functools import reduce
from pathlib import Path
from typing import Literal, NotRequired, Self, TypedDict, cast, final
from util import merge_dicts, read_yml, to_yaml
from .compose_struct import ComposeNet, ComposeNetArgs, HasServices
from .val_objs import DataDir, NameVal, OrgVal, Record, RecordCls, RecordVal, Url
def replace(rec: RecordCls[RecordVal], string: str) -> str:
return str.replace(string, rec.name, rec.val.to_str())
@final
@dataclass(frozen=True, slots=True)
class SrcPaths:
data_dir: Path
cfg_file: Path
env_file: Path
def src_path_factory(src: str) -> SrcPaths:
root = Path("/data/cfg")
dir = root.joinpath(src)
return SrcPaths(
data_dir=dir,
cfg_file=dir.joinpath("cfg.yml"),
env_file=dir.joinpath(".env"),
)
class OrgDataYaml(TypedDict):
org: NotRequired[str]
url: NotRequired[str]
class CfgDataYaml(TypedDict):
name: str
files: list[str]
orgs: list[OrgDataYaml]
@final
@dataclass(frozen=True, slots=True)
class OrgData:
org: str | None
url: str | None
@final
@dataclass(frozen=True, slots=True)
class CfgData:
name: str
files: tuple[Path, ...]
orgs: tuple[OrgData, ...]
def config_data_factory(cfg_dir: SrcPaths) -> CfgData:
data = cast(CfgDataYaml, read_yml(cfg_dir.cfg_file))
def sub_files():
for path in data["files"]:
yield cfg_dir.data_dir.joinpath(path)
def sub_orgs():
for org_data in data["orgs"]:
yield OrgData(
org_data.get("org"),
org_data.get("url"),
)
return CfgData(
data["name"],
tuple(sub_files()),
tuple(sub_orgs()),
)
@final
@dataclass(frozen=True, slots=True)
class ComposeTemplate:
org: RecordCls[OrgVal]
name: RecordCls[NameVal]
org_name: RecordCls[NameVal]
data: RecordCls[DataDir]
url: RecordCls[Url]
def __iter__(self):
yield self.org
yield self.name
yield self.org_name
yield self.data
yield self.url
def compose_template_factory(cfg_data: CfgData) -> Iterator[ComposeTemplate]:
# def sub():
for org_data in cfg_data.orgs:
_org = OrgVal(org_data.org)
_name = NameVal(cfg_data.name)
org_name = (
NameVal(f"{_org.to_str()}_{_name.to_str()}") if _org.is_valid() else _name
)
data_dir = Path("/data").joinpath(_org.to_str(), _name.to_str())
yield ComposeTemplate(
Record("ORG", _org),
Record("NAME", _name),
Record("ORG_NAME", org_name),
Record("DATA", DataDir(data_dir)),
Record("URL", Url(org_data.url)),
# data_dir,
# CFG_PATH.joinpath(org_name.to_str()),
)
# return tuple(sub())
@final
@dataclass(frozen=True, slots=True)
class DestPaths:
data_dir: Path
env_file: Path
compose_file: Path
def dest_paths_factory(compose_template: ComposeTemplate) -> DestPaths:
data_dir = compose_template.data.val.path
return DestPaths(
data_dir, data_dir.joinpath(".env"), data_dir.joinpath("docker-compose.yml")
)
@final
@dataclass(frozen=True, slots=True)
class ParsedCompose:
@final
@dataclass(frozen=True, slots=True)
class DestData:
paths: DestPaths
templates: ComposeTemplate
volumes: tuple[Path, ...]
def get_volumes_raw(data: HasServices) -> Iterator[str]:
for app_data in data["services"].values():
if "volumes" not in app_data:
return
for vol in app_data["volumes"]:
if not vol.startswith(r"${_DATA}"):
continue
yield vol.split(":", 1)[0]
def dest_data_factory(cfg_data: CfgData, compose: Compose) -> Iterator[DestData]:
# def sub():
vols_raw = tuple(get_volumes_raw(compose))
# templates = tuple(compose_template_factory(cfg_data))
# for template in templates:
# for vol in vols_raw:
for template in compose_template_factory(cfg_data):
paths = dest_paths_factory(template)
def vol_sub():
for vol in vols_raw:
yield Path(replace(template.data.val, vol))
yield DestData(
paths,
template,
tuple(vol_sub()),
)
# return tuple(sub())
@final
@dataclass(frozen=True, slots=True)
class ComposeData:
src_paths: SrcPaths
dest_data: tuple[DestData, ...]
Compose: Compose
proxy_net: tuple[str, str] | None
yaml: str
def get_compose(cfg_data: CfgData) -> Compose:
def insert_defaults(data: Compose) -> None:
sec_opts = "no-new-privileges:true"
data.name = "${_ORG_NAME}"
for app_data in data.services.values():
app_data.restart = "unless-stopped"
if app_data.security_opt is None:
app_data.security_opt = {sec_opts}
else:
app_data.security_opt.add(sec_opts)
def insert_traefik_labels(data: Compose) -> None:
for app_data in data.services.values():
traefik_labels = {
"traefik.http.routers.${_ORG_NAME}.rule=Host(`${_URL}`)",
"traefik.http.routers.${_ORG_NAME}.entrypoints=websecure",
"traefik.docker.network=${_ORG_NAME}_proxy",
"traefik.http.routers.${_ORG_NAME}.tls.certresolver=le",
}
if app_data.labels is None:
continue
if "traefik.enable=true" not in app_data.labels:
continue
app_data.labels.update(traefik_labels)
def sub():
for path in cfg_data.files:
_dict = cast(ComposeServiceYaml, read_yml(path))
yield ComposeService.from_dict(_dict)
compose = reduce(merge_dicts, sub(), cast(Compose, {})) # pyright: ignore[reportInvalidCast]
defaults = (insert_defaults, insert_traefik_labels)
for func in defaults:
func(compose)
return compose
def set_networks(data: Compose) -> tuple[str, str] | None:
def sub() -> Iterator[Literal["proxy", "internal"]]:
for app_data in data["services"].values():
if "networks" not in app_data:
continue
yield from app_data["networks"]
nets = set(sub())
net = ComposeNet()
if "proxy" in nets:
proxy = ComposeNetArgs(name="${_ORG_NAME}_proxy", external=True)
net["proxy"] = proxy
ret = "${_ORG_NAME}", "${_ORG_NAME}_proxy"
else:
ret = None
if "internal" in nets:
net["internal"] = ComposeNetArgs(name="${_ORG_NAME}")
if not net:
return
data["networks"] = net
return ret
def compose_data_factory(src: str):
src_path = src_path_factory(src)
cfg_data = config_data_factory(src_path)
compose = get_compose(cfg_data)
dest_data = dest_data_factory(cfg_data, compose)
return ComposeData(
src_path,
tuple(dest_data),
compose,
set_networks(compose),
to_yaml(compose),
)

52
src/compose/entity.py Normal file
View File

@@ -0,0 +1,52 @@
from dataclasses import asdict, dataclass
from typing import Literal, NotRequired, Self, TypedDict, final
from cfg.entity import CfgData
from net.entities import Net, NetTraefik, NetYaml
from service.entity import Service, ServiceYaml, TraefikService
from service.get import services_get_networks
from util import to_yaml
class ComposeYaml(TypedDict):
name: str
networks: NotRequired[NetYaml]
services: dict[str, ServiceYaml]
@final
@dataclass(frozen=True, slots=True)
class Compose:
cfg: CfgData
networks: Net | None
services: dict[str, Service]
@classmethod
def from_dict(cls, cfg: CfgData, data: ComposeYaml) -> Self:
# services = dict[str, ComposeService]()
services = dict(get_services_dict(data))
return cls(
cfg,
services_get_networks(services.values()),
services,
)
def as_yaml(self) -> str:
return to_yaml(asdict(self))
def get_services_dict(data: ComposeYaml):
for k, v in data["services"].items():
yield k, Service.from_dict(v)
@final
@dataclass(frozen=True, slots=True)
class TraefikCompose:
cfg: CfgData
networks: NetTraefik
services: dict[Literal["traefik"], TraefikService]
def as_yaml(self) -> str:
return to_yaml(asdict(self))

37
src/compose/factory.py Normal file
View File

@@ -0,0 +1,37 @@
from collections.abc import Iterable
from cfg import TRAEFIK_PATH
from cfg.entity import CfgData
from cfg.factory import cfg_data_factory
from cfg.get import cfg_get_services
from compose.entity import Compose, TraefikCompose
from net.entities import NetTraefik
from rendered.entity import Rendered
from rendered.get import rendered_get_nets
from service.entity import TraefikService
from service.factory import get_traefik_service
from service.get import services_get_networks
from src_path.entity import src_paths_factory
def compose_factory(cfg_data: CfgData) -> Compose:
services = dict(cfg_get_services(cfg_data))
return Compose(
cfg_data,
services_get_networks(services.values()),
services,
)
def traefik_compose_factory(renders: Iterable[Rendered]) -> TraefikCompose:
src_paths = src_paths_factory(TRAEFIK_PATH)
cfg = cfg_data_factory(src_paths)
service = get_traefik_service()
nets: NetTraefik = dict(rendered_get_nets(renders))
service["networks"] = list(nets.keys())
return TraefikCompose(
cfg,
nets,
{"traefik": TraefikService.from_dict(service)},
)

26
src/compose/get.py Normal file
View File

@@ -0,0 +1,26 @@
from collections.abc import Iterator
from compose.entity import Compose, TraefikCompose
from util import get_replace_name
def compose_get_volumes(compose: Compose | TraefikCompose) -> Iterator[str]:
if isinstance(compose, TraefikCompose):
return
f = get_replace_name("DATA")
for app_data in compose.services.values():
if app_data.volumes is None:
return
for vol in app_data.volumes:
if not vol.startswith(f):
continue
yield vol.split(":", 1)[0]
# def compose_get_volumes(compose: Compose | TraefikCompose):
# return _volumes_sub(compose)
# # vols = set(_volumes_sub(compose))
# # if not vols:
# # return None
# # return vols

35
src/compose/service.py
View File

@@ -1,35 +0,0 @@
from functools import reduce
from pathlib import Path
from shutil import copyfile
from .entities import ComposeData, replace
def mk_dir(path: Path):
if path.exists():
return
path.mkdir(parents=True)
def mk_compose_dir(compose_data: ComposeData) -> None:
for dest in compose_data.dest_data:
mk_dir(dest.paths.data_dir)
for vol in dest.volumes:
mk_dir(vol)
def mk_compose_env(compose_data: ComposeData) -> None:
src = compose_data.src_paths.env_file
for dest_data in compose_data.dest_data:
dest = dest_data.paths.env_file
if src.exists() and not dest.exists():
_ = copyfile(src, dest)
def write(compose_data: ComposeData) -> None:
for dest_data in compose_data.dest_data:
string = reduce(
lambda s, rec: replace(rec, s), dest_data.templates, compose_data.yaml
)
with dest_data.paths.compose_file.open("wt") as f:
_ = f.write(string)

11
src/dest_path/entity.py Normal file
View File

@@ -0,0 +1,11 @@
from dataclasses import dataclass
from pathlib import Path
from typing import final
@final
@dataclass(frozen=True, slots=True)
class DestPaths:
data_dir: Path
env_file: Path
compose_file: Path

17
src/dest_path/factory.py Normal file
View File

@@ -0,0 +1,17 @@
from cfg import DATA_ROOT
from dest_path.entity import DestPaths
from template.entity import Template
def dest_paths_factory(template: Template) -> DestPaths:
r_args = template.replace_args
path = (
DATA_ROOT.joinpath(template.compose.cfg.name)
if r_args is None
else r_args.data.val.path
)
return DestPaths(
path,
path.joinpath(".env"),
path.joinpath("docker-compose.yml"),
)

46
src/main.py
View File

@@ -1,23 +1,33 @@
from typing import Any, Generator
from collections.abc import Iterable, Iterator
from cfg import CFG_ROOT
from cfg.factory import cfg_data_factory
from compose.factory import compose_factory, traefik_compose_factory
from rendered.entity import Rendered
from rendered.factory import rendered_factory
from rendered.util import write
from src_path.entity import src_paths_factory
from template.factory import template_factory
from collections.abc import Iterator
from Ts import TraefikComposeDict, TraefikNet, TraefikNetName
from compose import gen_compose
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__":
paths: tuple[str, ...] = ("gitea", "opencloud", "jellyfin", "immich")
def sub() -> Iterator[tuple[str, str]]:
for path in paths:
yield from gen_compose(path)
# yield name, TraefikNetName(name=proxy)
networks: TraefikNet = dict(sub())
traefik = ComposeBuild("traefik").build()
traefik_compose = TraefikComposeDict(
name="traefik",
networks=networks,
)
renders = render_all()
traefik = traefik_compose_factory(renders)
for template in template_factory(traefik):
rendered = rendered_factory(template)
write(rendered)

45
src/net/entities.py Normal file
View File

@@ -0,0 +1,45 @@
from dataclasses import dataclass
from typing import NotRequired, Self, TypedDict, final
type NetTraefik = dict[str, NetArgs]
class NetArgsYaml(TypedDict):
name: str
external: NotRequired[bool]
@final
@dataclass(frozen=True, slots=True)
class NetArgs:
name: str
external: bool | None
@classmethod
def from_dict(cls, data: NetArgsYaml) -> Self:
return cls(
data["name"],
data.get("external"),
)
class NetYaml(TypedDict):
internal: NotRequired[NetArgsYaml]
proxy: NotRequired[NetArgsYaml]
@final
@dataclass(frozen=True, slots=True)
class Net:
internal: NetArgs | None
proxy: NetArgs | None
@classmethod
def from_class(cls, data: NetYaml) -> Self:
internal = data.get("internal")
if internal is not None:
internal = NetArgs.from_dict(internal)
proxy = data.get("proxy")
if proxy is not None:
proxy = NetArgs.from_dict(proxy)
return cls(internal, proxy)

21
src/net/factory.py Normal file
View File

@@ -0,0 +1,21 @@
from net.entities import Net, NetArgs
from util import get_replace_name
def net_args_factory(name: str, external: bool | None = None) -> NetArgs:
return NetArgs(name, external if external else None)
def net_factory(name: str, _internal: bool, _proxy: bool) -> Net:
return Net(
net_args_factory(name, _internal) if _internal else None,
net_args_factory(f"{name}_proxy", _proxy) if _proxy else None,
)
def net_factory_re(_internal: bool, _proxy: bool) -> Net:
return net_factory(
get_replace_name("name"),
_internal,
_proxy,
)

18
src/rendered/entity.py Normal file
View File

@@ -0,0 +1,18 @@
from dataclasses import dataclass
from pathlib import Path
from typing import final
from dest_path.entity import DestPaths
from src_path.entity import SrcPaths
from template.entity import Template
@final
@dataclass(frozen=True, slots=True)
class Rendered:
template: Template
src_paths: SrcPaths
dest_paths: DestPaths
volumes: frozenset[Path] | None
proxy_net: str | None
data: str

25
src/rendered/factory.py Normal file
View File

@@ -0,0 +1,25 @@
from functools import reduce
from dest_path.factory import dest_paths_factory
from rendered.entity import Rendered
from template.entity import Template
from template.get import template_get_proxy, template_get_vols
def rendered_factory(template: Template) -> Rendered:
yml = template.compose.as_yaml()
if template.replace_args is not None:
yml = reduce(
lambda s, f: f.replace(s),
template.replace_args,
yml,
)
vols = frozenset(template_get_vols(template))
return Rendered(
template,
template.compose.cfg.src_paths,
dest_paths_factory(template),
vols if vols else None,
template_get_proxy(template),
yml,
)

13
src/rendered/get.py Normal file
View File

@@ -0,0 +1,13 @@
from collections.abc import Iterable, Iterator
from net.entities import NetArgs
from net.factory import net_args_factory
from rendered.entity import Rendered
def rendered_get_nets(renders: Iterable[Rendered]) -> Iterator[tuple[str, NetArgs]]:
for render in renders:
net = render.proxy_net
if net is None:
continue
yield net, net_args_factory(f"{net}_proxy")

38
src/rendered/util.py Normal file
View File

@@ -0,0 +1,38 @@
from pathlib import Path
from shutil import copyfile
from rendered.entity import Rendered
def _mk_dir(path: Path) -> None:
if path.exists():
return
path.mkdir(parents=True)
def _mk_compose_dir(rendered: Rendered) -> None:
_mk_dir(rendered.dest_paths.data_dir)
vols = rendered.volumes
if vols is None:
return
for vol in vols:
_mk_dir(vol)
def _mk_compose_env(rendered: Rendered) -> None:
src = rendered.src_paths.env_file
dest = rendered.dest_paths.env_file
if src.exists() and not dest.exists():
_ = copyfile(src, dest)
def write_raw(path: Path, data: str) -> None:
with path.open("wt") as f:
_ = f.write(data)
def write(rendered: Rendered) -> None:
funcs = (_mk_compose_dir, _mk_compose_env)
for func in funcs:
func(rendered)
write_raw(rendered.dest_paths.compose_file, rendered.data)

158
src/service/entity.py Normal file
View File

@@ -0,0 +1,158 @@
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Literal, NotRequired, Self, TypedDict, TypeVar, overload, override
from cfg import T_Primitive
from util import get_replace_name
type T_NetAbc = str | Literal["proxy", "internal"]
TCo_NetABC = TypeVar("TCo_NetABC", bound=T_NetAbc, covariant=True)
class ServiceYamlAbc[T_net: T_NetAbc](TypedDict):
command: NotRequired[list[str]]
container_name: str
entrypoint: list[str]
environment: NotRequired[dict[str, T_Primitive]]
image: str
labels: NotRequired[list[str]]
logging: dict[str, str]
networks: NotRequired[list[T_net]]
restart: str
security_opt: NotRequired[list[str]]
user: NotRequired[str]
volumes: NotRequired[list[str]]
TCo_ServiceYaml = TypeVar(
"TCo_ServiceYaml",
bound=ServiceYamlAbc[T_NetAbc],
covariant=True,
)
class ServiceYaml(ServiceYamlAbc[Literal["proxy", "internal"]]):
pass
class TraefikServiceYaml(ServiceYamlAbc[str]):
pass
type T_Compose = ServiceYaml | TraefikServiceYaml
@dataclass(frozen=True, slots=True)
class ServiceAbc[T_net: T_NetAbc, T_Yaml: T_Compose](ABC):
command: tuple[str, ...] | None
container_name: str
entrypoint: tuple[str, ...]
environment: dict[str, T_Primitive] | None
image: str
labels: frozenset[str] | None
logging: dict[str, str] | None
networks: frozenset[T_net] | None
restart: str
security_opt: frozenset[str]
user: str | None
volumes: frozenset[str] | None
@classmethod
def from_dict(cls, data: T_Yaml) -> Self:
command = data.get("command")
volumes = data.get("volumes")
return cls(
tuple(command) if command else None,
data["container_name"],
tuple(data["entrypoint"]),
data.get("environment"),
data["image"],
_get_labels(data),
None,
cls.get_nets(data),
"unless-stopped",
_get_sec_opts(data),
data.get("user"),
frozenset(volumes) if volumes else None,
)
def is_valid(self) -> bool:
attrs = (self.container_name, self.restart, self.logging)
for attr in attrs:
if attr is None:
return False
return True
@abstractmethod
@staticmethod
def get_nets(data: T_Yaml) -> frozenset[T_net] | None:
pass
# return self._get_nets(data)
class Service(ServiceAbc[Literal["proxy", "internal"], ServiceYaml]):
@override
@staticmethod
def get_nets(data: ServiceYaml) -> frozenset[Literal["proxy", "internal"]] | None:
return _get_nets(data)
class TraefikService(ServiceAbc[str, TraefikServiceYaml]):
@override
@staticmethod
def get_nets(data: TraefikServiceYaml) -> frozenset[str] | None:
return _get_nets(data)
@overload
def _get_nets(
data: ServiceYaml,
) -> frozenset[Literal["proxy", "internal"]] | None:
pass
@overload
def _get_nets(data: TraefikServiceYaml) -> frozenset[str] | None:
pass
def _get_nets(
data: ServiceYaml | TraefikServiceYaml,
) -> frozenset[str] | frozenset[Literal["proxy", "internal"]] | None:
nets = data.get("networks")
if nets is None:
return
return frozenset(nets)
def _get_sec_opts(
data: T_Compose,
) -> frozenset[str]:
sec_opts = frozenset(
"no-new-privileges:true",
)
sec = data.get("security_opt")
if not sec:
return sec_opts
return sec_opts.union(sec)
def _get_labels(
data: T_Compose,
) -> frozenset[str] | None:
org_name = get_replace_name("org_name")
url = get_replace_name("url")
traefik_labels = frozenset(
(
f"traefik.http.routers.{org_name}.rule=Host(`{url}`)",
f"traefik.http.routers.{org_name}.entrypoints=websecure",
f"traefik.docker.network={org_name}_proxy",
f"traefik.http.routers.{org_name}.tls.certresolver=le",
)
)
labels = data.get("labels")
if not labels:
return
if "traefik.enable=true" not in labels:
return frozenset(labels)
return traefik_labels.union(labels)

10
src/service/factory.py Normal file
View File

@@ -0,0 +1,10 @@
from typing import cast
from cfg import TRAEFIK_PATH
from service.entity import TraefikServiceYaml
from util import read_yml
def get_traefik_service():
path = TRAEFIK_PATH.joinpath("service.yml")
return cast(TraefikServiceYaml, read_yml(path))

26
src/service/get.py Normal file
View File

@@ -0,0 +1,26 @@
from collections.abc import Iterable, Iterator
from typing import Literal
from net.entities import Net
from net.factory import net_factory_re
from service.entity import Service
def _networks_sub(
services: Iterable[Service],
) -> Iterator[Literal["proxy", "internal"]]:
for app_data in services:
networks = app_data.networks
if networks is None:
continue
yield from networks
def services_get_networks(services: Iterable[Service]) -> Net | None:
networks = frozenset(_networks_sub(services))
if not networks:
return None
return net_factory_re(
"internal" in networks,
"proxy" in networks,
)

19
src/src_path/entity.py Normal file
View File

@@ -0,0 +1,19 @@
from dataclasses import dataclass
from pathlib import Path
from typing import final
@final
@dataclass(frozen=True, slots=True)
class SrcPaths:
cfg_dir: Path
cfg_file: Path
env_file: Path
def src_paths_factory(src: Path) -> SrcPaths:
return SrcPaths(
cfg_dir=src,
cfg_file=src.joinpath("cfg.yml"),
env_file=src.joinpath(".env"),
)

7
src/src_path/get.py Normal file
View File

@@ -0,0 +1,7 @@
from cfg.entity import CfgDataYaml
from src_path.entity import SrcPaths
def src_path_get_files(src_paths: SrcPaths, data: CfgDataYaml):
for path in data["files"]:
yield src_paths.cfg_dir.joinpath(path)

32
src/template/entity.py Normal file
View File

@@ -0,0 +1,32 @@
from dataclasses import dataclass
from typing import final
from compose.entity import Compose, TraefikCompose
from template.val_obj import DataDir, NameVal, OrgVal, RecordCls, Url
@final
@dataclass(frozen=True, slots=True)
class ReplaceArgs:
org: RecordCls[OrgVal]
name: RecordCls[NameVal]
org_name: RecordCls[NameVal]
data: RecordCls[DataDir]
url: RecordCls[Url]
def __iter__(self):
yield self.org
yield self.name
yield self.org_name
yield self.data
yield self.url
@final
@dataclass(frozen=True, slots=True)
class Template:
compose: Compose | TraefikCompose
replace_args: ReplaceArgs | None
# dest_paths: DestPaths
volumes: frozenset[str] | None
# proxy_net: str | None

49
src/template/factory.py Normal file
View File

@@ -0,0 +1,49 @@
from collections.abc import Iterator
from cfg import DATA_ROOT
from cfg.entity import CfgData, OrgData
from compose.entity import Compose, TraefikCompose
from compose.get import compose_get_volumes
from template.entity import ReplaceArgs, Template
from template.val_obj import DataDir, NameVal, OrgVal, Record, Url
def replace_args_factory(cfg_data: CfgData, org_data: OrgData) -> ReplaceArgs:
_org = OrgVal(org_data.org)
_name = NameVal(cfg_data.name)
org_name = (
NameVal(f"{_org.to_str()}_{_name.to_str()}") if _org.is_valid() else _name
)
data_dir = DATA_ROOT.joinpath(_org.to_str(), _name.to_str())
return ReplaceArgs(
Record("org", _org),
Record("name", _name),
Record("org_name", org_name),
Record("data", DataDir(data_dir)),
Record("url", Url(org_data.url)),
)
def template_factory(compose: Compose | TraefikCompose) -> Iterator[Template]:
cfg_data = compose.cfg
vols = frozenset(compose_get_volumes(compose))
if not vols:
vols = None
orgs = cfg_data.orgs
if orgs is None:
yield Template(compose, None, vols)
return
for org_data in orgs:
args = replace_args_factory(
cfg_data,
org_data,
)
yield Template(
compose,
args,
vols,
)

36
src/template/get.py Normal file
View File

@@ -0,0 +1,36 @@
from collections.abc import Iterable
from pathlib import Path
from net.entities import Net
from template.entity import Template
def template_get_vols(template: Template) -> Iterable[Path]:
def _lambda(x: str) -> str:
return x
vols = template.volumes
if not vols:
return
r_args = template.replace_args
re = _lambda if r_args is None else r_args.data.replace
for vol in vols:
yield Path(re(vol))
def template_get_proxy(template: Template) -> str | None:
proxy = template.compose.networks
if proxy is None:
return
if not isinstance(proxy, Net):
return
if proxy.proxy is None:
return
if proxy.internal is None:
net = proxy.proxy.name.replace("_proxy", "")
else:
net = proxy.internal.name
r_args = template.replace_args
if r_args is None:
return net
return r_args.name.replace(net)

49
src/compose/val_objs.py → src/template/val_obj.py
View File

@@ -1,10 +1,27 @@
from abc import ABC, abstractmethod
from dataclasses import dataclass
from pathlib import Path
from typing import Protocol, final
from typing import TypeVar, final, override
from util import get_replace_name
class RecordVal(Protocol):
def to_str(self) -> str: ...
class RecordVal(ABC):
@abstractmethod
def to_str(self) -> str:
pass
TCo_RecordVal = TypeVar(
"TCo_RecordVal",
bound=RecordVal,
covariant=True,
)
TCon_RecordVal = TypeVar(
"TCon_RecordVal",
bound=RecordVal,
contravariant=True,
)
@final
@@ -13,12 +30,21 @@ class RecordCls[T: RecordVal]:
name: str
val: T
# @final
# class RecordClsProto(Protocol):
# name:str
# val: RecordVal
def replace(self, string: str) -> str:
return str.replace(string, self.name, self.val.to_str())
@final
@dataclass(frozen=True, slots=True)
class OrgVal:
class OrgVal(RecordVal):
val: str | None
@override
def to_str(self) -> str:
if self.val is None:
return "personal"
@@ -30,36 +56,35 @@ class OrgVal:
@final
@dataclass(frozen=True, slots=True)
class NameVal:
class NameVal(RecordVal):
val: str
@override
def to_str(self) -> str:
return self.val
@final
@dataclass(frozen=True, slots=True)
class DataDir:
class DataDir(RecordVal):
path: Path
@override
def to_str(self) -> str:
return str(self.path)
@final
@dataclass(frozen=True, slots=True)
class Url:
class Url(RecordVal):
sub_url: str | None
@override
def to_str(self) -> str:
if self.sub_url is None:
return ""
return ".".join([self.sub_url, "ccamper7", "net"])
# def get_replace_var(string: str) -> str:
# return f"${{_{name}}}"
def Record[T: RecordVal](name: str, val: T) -> RecordCls[T]:
return RecordCls(f"${{_{name}}}", val)
return RecordCls(get_replace_name(name), val)

0
src/treafik.py
View File

18
src/util.py
View File

@@ -5,7 +5,7 @@ from typing import Any, cast, override
import yaml
from compose.compose_struct import T_PrimDict, T_PrimVal, T_Primitive
from cfg import T_PrimDict, T_Primitive, T_PrimVal, T_YamlDict
class VerboseSafeDumper(yaml.SafeDumper):
@@ -16,8 +16,8 @@ class VerboseSafeDumper(yaml.SafeDumper):
def merge_dicts[T: Mapping[Any, Any]](dict1: T, dict2: T) -> T:
def _merge_dicts(dict1: T_PrimDict, dict2: T_PrimDict):
s1 = set(dict1.keys())
s2 = set(dict2.keys())
s1 = frozenset(dict1.keys())
s2 = frozenset(dict2.keys())
for k in s1.difference(s2):
yield k, dict1[k]
for k in s2.difference(s1):
@@ -29,18 +29,22 @@ def merge_dicts[T: Mapping[Any, Any]](dict1: T, dict2: T) -> T:
yield k, dict[T_Primitive, T_PrimVal](_merge_dicts(v1, v2))
continue
if isinstance(v1, list) and isinstance(v2, list):
yield k, list(set(v1).union(v2))
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) -> Mapping[Any, Any]: # pyright: ignore[reportExplicitAny]
def read_yml(path: Path) -> T_YamlDict:
with path.open("rt") as f:
return yaml.safe_load(f) # pyright: ignore[reportAny]
return cast(T_YamlDict, yaml.safe_load(f))
def to_yaml(data: Mapping[Any, Any]) -> str: # pyright: ignore[reportExplicitAny]
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()}}}"