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

working

Browse Source
This commit is contained in:
Christian Camper
2026-01-08 23:04:58 -06:00
parent ca23b44a25
commit 38b6807e70
64 changed files with 1327 additions and 1059 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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]

33
src/compose/cfg/entity.py
View File

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

30
src/compose/cfg/factory.py
View File

@@ -1,30 +0,0 @@
from pathlib import Path
from typing import cast
from compose.cfg.entity import CfgData, CfgDataYaml, OrgDataYaml
from compose.cfg.get import cfg_get_orgs
from compose.src_path.entity import SrcPaths
from compose.src_path.get import src_path_get_services, src_path_get_volumes
from compose.util import read_yml, validate_typed_dict
def cfg_data_yml_factory(file: Path) -> CfgDataYaml:
data = cast(CfgDataYaml, read_yml(file))
validate_typed_dict(CfgDataYaml, data, file)
orgs_key = "orgs"
for org in data[orgs_key]:
validate_typed_dict(OrgDataYaml, org, file, (orgs_key,))
return data
def cfg_data_factory(src_paths: SrcPaths) -> CfgData:
data = cfg_data_yml_factory(src_paths.cfg_file)
vols = frozenset(src_path_get_volumes(src_paths, data))
return CfgData(
src_paths,
src_paths.cfg_dir.name,
frozenset(src_path_get_services(src_paths, data)),
vols if vols else None,
frozenset(cfg_get_orgs(data)),
)

83
src/compose/cfg/get.py
View File

@@ -1,83 +0,0 @@
from collections.abc import Iterator
from typing import cast
from compose.cfg.entity import CfgData, CfgDataYaml, OrgData
from compose.compose.entity import VolYaml
from compose.service.entity import Service, T_Compose
from compose.service.factory import services_yaml_factory
from compose.util import get_replace_name, read_yml
def cfg_get_orgs(data: CfgDataYaml) -> Iterator[OrgData]:
for org_data in data["orgs"]:
yield OrgData(
org_data["org"],
org_data.get("url"),
)
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)
def cfg_get_services(cfg_data: CfgData) -> Iterator[tuple[str, Service]]:
for path in cfg_data.services:
data = services_yaml_factory(path)
# yield path.stem, Service.from_dict(data)
# @classmethod
command = data.get("command")
volumes = data.get("volumes")
entry = data.get("entrypoint")
service = Service(
tuple(command) if command else None,
get_replace_name("org_name"),
tuple(entry) if entry else None,
data.get("environment"),
data["image"],
_get_labels(data),
None,
Service.get_nets(data),
"unless-stopped",
_get_sec_opts(data),
data.get("user"),
frozenset(volumes) if volumes else None,
)
yield path.stem, service
def cfg_get_volumes(cfg_data: CfgData) -> Iterator[tuple[str, VolYaml]]:
vols = cfg_data.volumes
if vols is None:
return
for path in vols:
yield path.stem, cast(VolYaml, read_yml(path))

66
src/compose/compose/entity.py
View File

@@ -1,66 +0,0 @@
from dataclasses import asdict, dataclass
from typing import Literal, NotRequired, TypedDict, final
from compose.cfg import T_YamlDict
from compose.cfg.entity import CfgData
from compose.net.entities import Net, NetTraefik, NetYaml
from compose.service.entity import Service, ServiceYaml, TraefikService
from compose.util import to_yaml
type VolYaml = dict[str, T_YamlDict]
class ComposeYaml(TypedDict):
name: str
services: dict[str, ServiceYaml]
networks: NotRequired[NetYaml]
volumes: NotRequired[dict[str, T_YamlDict]]
@final
@dataclass(frozen=True, slots=True)
class Compose:
cfg: CfgData
services: dict[str, Service]
networks: Net | None
volumes: VolYaml | None
# @classmethod
# def from_dict(cls, cfg: CfgData, data: ComposeYaml) -> Self:
# # services = dict[str, ComposeService]()
# services = dict(_get_services_dict(data))
# # vols = frozenset(_get_volumes_dict(data))
# return cls(
# cfg,
# services,
# services_get_networks(services.values()),
# data.get("volumes"),
# )
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)
# def _get_volumes_dict(data: ComposeYaml) -> Iterator[VolYaml]:
# vols = data.get("volumes")
# if vols is None:
# return
# for k, v in vols.items():
# yield {k: v}
@final
@dataclass(frozen=True, slots=True)
class TraefikCompose:
cfg: CfgData
services: dict[Literal["traefik"], TraefikService]
networks: NetTraefik
volumes: None
def as_yaml(self) -> str:
return to_yaml(asdict(self))

38
src/compose/compose/factory.py
View File

@@ -1,38 +0,0 @@
from compose.cfg.entity import CfgData
from compose.cfg.get import cfg_get_services, cfg_get_volumes
from compose.compose.entity import Compose, VolYaml
# from compose.service.factory import get_traefik_service
from compose.service.get import services_get_networks
def compose_factory(cfg_data: CfgData) -> Compose:
services = dict(cfg_get_services(cfg_data))
vols: VolYaml | None = dict(cfg_get_volumes(cfg_data))
return Compose(
cfg_data,
services,
services_get_networks(services.values()),
vols if vols else None,
)
# def traefik_compose_factory(renders: Iterable[Rendered]) -> TraefikCompose:
# src_paths = src_paths_factory(TRAEFIK_PATH)
# cfg = cfg_data_factory(src_paths)
# # cfg = CfgData(
# # src_paths,
# # 'traefik',
# # frozenset((TRAEFIK_PATH.joinpath('traefik'),)),
# # None,
# # )
# 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/compose/get.py
View File

@@ -1,26 +0,0 @@
from collections.abc import Iterator
from compose.compose.entity import Compose, TraefikCompose
from compose.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

11
src/compose/dest_path/entity.py
View File

@@ -1,11 +0,0 @@
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/compose/dest_path/factory.py
View File

@@ -1,17 +0,0 @@
from compose.cfg import DATA_ROOT
from compose.dest_path.entity import DestPaths
from compose.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"),
)

45
src/compose/net/entities.py
View File

@@ -1,45 +0,0 @@
from dataclasses import dataclass
from typing import NotRequired, 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_dict(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/compose/net/factory.py
View File

@@ -1,21 +0,0 @@
from compose.net.entities import Net, NetArgs
from compose.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/compose/rendered/entity.py
View File

@@ -1,18 +0,0 @@
from dataclasses import dataclass
from pathlib import Path
from typing import final
from compose.dest_path.entity import DestPaths
from compose.src_path.entity import SrcPaths
from compose.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

26
src/compose/rendered/factory.py
View File

@@ -1,26 +0,0 @@
from functools import reduce
from compose.dest_path.factory import dest_paths_factory
from compose.rendered.entity import Rendered
from compose.template.entity import Template
from compose.template.get import template_get_proxy, template_get_vols
from compose.template.util import replace
def rendered_factory(template: Template) -> Rendered:
yml = template.compose.as_yaml()
if template.replace_args is not None:
yml = reduce(
lambda s, f: replace(f, 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/compose/rendered/get.py
View File

@@ -1,13 +0,0 @@
from collections.abc import Iterable, Iterator
from compose.net.entities import NetArgs
from compose.net.factory import net_args_factory
from compose.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/compose/rendered/util.py
View File

@@ -1,38 +0,0 @@
from pathlib import Path
from shutil import copyfile
from compose.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)

128
src/compose/service/entity.py
View File

@@ -1,128 +0,0 @@
from abc import ABCMeta, abstractmethod
from dataclasses import dataclass
from typing import Literal, NotRequired, TypedDict, TypeVar, overload, override
from compose.cfg import T_Primitive
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: NotRequired[str]
entrypoint: NotRequired[list[str]]
environment: NotRequired[dict[str, T_Primitive]]
image: str
labels: NotRequired[list[str]]
logging: NotRequired[dict[str, str]]
networks: NotRequired[list[T_net]]
restart: NotRequired[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](metaclass=ABCMeta):
command: tuple[str, ...] | None
container_name: str
entrypoint: tuple[str, ...] | None
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")
# entry = data.get("entrypoint")
# name = data.get("container_name")
# if name is None:
# raise KeyError
# return cls(
# tuple(command) if command else None,
# name,
# tuple(entry) if entry else None,
# 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
@staticmethod
@abstractmethod
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)

11
src/compose/service/factory.py
View File

@@ -1,11 +0,0 @@
from pathlib import Path
from typing import cast
from compose.service.entity import ServiceYaml
from compose.util import read_yml, validate_typed_dict
def services_yaml_factory(path: Path):
data = cast(ServiceYaml, read_yml(path))
validate_typed_dict(ServiceYaml, data, path)
return data

26
src/compose/service/get.py
View File

@@ -1,26 +0,0 @@
from collections.abc import Iterable, Iterator
from typing import Literal
from compose.net.entities import Net
from compose.net.factory import net_factory_re
from compose.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/compose/src_path/entity.py
View File

@@ -1,19 +0,0 @@
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"),
)

15
src/compose/src_path/get.py
View File

@@ -1,15 +0,0 @@
from compose.cfg.entity import CfgDataYaml
from compose.src_path.entity import SrcPaths
def src_path_get_services(src_paths: SrcPaths, data: CfgDataYaml):
for path in data["services"]:
yield src_paths.cfg_dir.joinpath(path)
def src_path_get_volumes(src_paths: SrcPaths, data: CfgDataYaml):
vols = data.get("volumes")
if vols is None:
return
for path in vols:
yield src_paths.cfg_dir.joinpath(path)

40
src/compose/template/entity.py
View File

@@ -1,40 +0,0 @@
from collections.abc import Iterator
from dataclasses import dataclass
from typing import final
from compose.compose.entity import Compose, TraefikCompose
from compose.template.val_obj import (
DataDir,
NameVal,
OrgVal,
RecordCls,
T_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) -> Iterator[T_RecordCls]:
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

44
src/compose/template/factory.py
View File

@@ -1,44 +0,0 @@
from collections.abc import Iterator
from compose.cfg import DATA_ROOT
from compose.cfg.entity import CfgData, OrgData
from compose.compose.entity import Compose, TraefikCompose
from compose.compose.get import compose_get_volumes
from compose.template.entity import ReplaceArgs, Template
from compose.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
for org_data in cfg_data.orgs:
args = replace_args_factory(
cfg_data,
org_data,
)
yield Template(
compose,
args,
vols,
)

38
src/compose/template/get.py
View File

@@ -1,38 +0,0 @@
from collections.abc import Iterable
from functools import partial
from pathlib import Path
from compose.net.entities import Net
from compose.template.entity import Template
from compose.template.util import replace
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 partial(replace, r_args.data)
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 replace(r_args.name, net)

5
src/compose/template/util.py
View File

@@ -1,5 +0,0 @@
from compose.template.val_obj import T_RecordCls
def replace(record: T_RecordCls, string: str) -> str:
return str.replace(string, record.name, record.val.to_str())

100
src/compose/template/val_obj.py
View File

@@ -1,100 +0,0 @@
from dataclasses import dataclass
from pathlib import Path
from typing import Protocol, final
from compose.util import get_replace_name
# class RecordVal(ABC):
# @abstractmethod
# def to_str(self) -> str:
# pass
@final
class RecordVal(Protocol):
def to_str(self) -> str: ...
# TCo_RecordVal = TypeVar(
# "TCo_RecordVal",
# bound=RecordVal,
# covariant=True,
# )
# TCon_RecordVal = TypeVar(
# "TCon_RecordVal",
# bound=RecordVal,
# contravariant=True,
# )
@final
class T_RecordCls(Protocol):
# name: str
# val: RecordVal
@property
def name(self) -> str: ...
@property
def val(self) -> RecordVal: ...
@final
@dataclass(frozen=True, slots=True)
class RecordCls[T: RecordVal]:
name: str
val: T
# @final
# class RecordClsProto(Protocol):
# name:str
# val: RecordVal
# def replace(self:RecordCls[TCo_RecordVal], string: str) -> str:
# return str.replace(string, self.name, self.val.to_str())
@final
@dataclass(frozen=True, slots=True)
class OrgVal:
val: str | None
def to_str(self) -> str:
if self.val is None:
return "personal"
return self.val
def is_valid(self) -> bool:
return self.val is not None
@final
@dataclass(frozen=True, slots=True)
class NameVal:
val: str
def to_str(self) -> str:
return self.val
@final
@dataclass(frozen=True, slots=True)
class DataDir:
path: Path
def to_str(self) -> str:
return str(self.path)
@final
@dataclass(frozen=True, slots=True)
class Url:
sub_url: str | None
def to_str(self) -> str:
if self.sub_url is None:
return ""
return ".".join([self.sub_url, "ccamper7", "net"])
def Record[T: RecordVal](name: str, val: T) -> RecordCls[T]:
return RecordCls(get_replace_name(name), val)

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

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

@@ -0,0 +1,36 @@
from collections.abc import Iterator
from docker_compose.cfg import CFG_ROOT, TRAEFIK_PATH
from docker_compose.compose.net_yaml import NetArgsYaml
from docker_compose.compose.rendered import Rendered
from docker_compose.util.yaml_util import to_yaml
def load_all() -> Iterator[Rendered]:
for path in CFG_ROOT.iterdir():
if path == TRAEFIK_PATH:
continue
yield from Rendered.from_path(path)
def render_all() -> Iterator[str]:
for rendered in load_all():
rendered.write()
rendered.write_bind_vols()
rendered.mk_bind_vols()
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
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()

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

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

62
src/docker_compose/cfg/cfg_paths.py Normal file
View File

@@ -0,0 +1,62 @@
from collections.abc import Iterator
from dataclasses import dataclass
from itertools import chain
from shutil import copyfile
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.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.from_iters(
src_paths.service_dir.files,
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) -> None:
src = self.src_paths.env_file
dest = self.dest_paths.env_file
if src.exists() and not dest.exists():
_ = copyfile(src, dest)

112
src/docker_compose/cfg/compose_paths.py Normal file
View File

@@ -0,0 +1,112 @@
from collections.abc import Callable, Iterable, Iterator, MutableMapping
from dataclasses import dataclass, field
from pathlib import Path
from typing import Self, cast, final, override
import yaml
from docker_compose.cfg.org import OrgData
from docker_compose.cfg.record import Record, RecordCls, RecordName
from docker_compose.compose.services_yaml import ServiceYamlRead
from docker_compose.compose.volume_yaml import VolYaml
from docker_compose.util.Ts import T_YamlRW
from docker_compose.util.yaml_util import path_to_typed, read_yaml
@final
@dataclass(frozen=True, slots=True)
class ServiceValNew:
val: Path
@override
def __str__(self) -> str:
return str(RecordName(self.val.stem))
@property
def replace(self) -> Record[Self, str]:
return Record(self, self.val.stem)
@final
@dataclass(frozen=True, slots=True)
class ServiceVal(RecordCls[ServiceValNew]):
old = RecordName("service")
@property
def stage_two(self):
return self.new.replace
@final
@dataclass(frozen=True, slots=True)
class ServicePath:
fqdn = Record[RecordName, str](
RecordName("fqdn"),
f"{OrgData.org_app.old!s}_{ServiceVal.old!s}",
)
path: Path
replace: ServiceVal = field(init=False)
def __post_init__(self):
setter = super(ServicePath, self).__setattr__
setter("replace", ServiceVal(ServiceValNew(self.path)))
@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: T_YamlRW = 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 data_dict # pyright: ignore[reportReturnType]
@property
def pre_render_funcs(self) -> Iterator[Callable[[str], str]]:
yield self.fqdn
yield self.replace
@property
def render_funcs(self) -> Iterator[Callable[[str], str]]:
yield self.replace.stage_two
@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]
@classmethod
def from_iters(cls, services: Iterable[ServicePath], volumes: Iterable[VolumePath]):
return cls(
frozenset(services),
frozenset(volumes),
)
@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

81
src/docker_compose/cfg/dest_path.py Normal file
View File

@@ -0,0 +1,81 @@
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 AppVal, OrgData, OrgVal
from docker_compose.cfg.record import Record, RecordName
@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[RecordName, Path](RecordName("data_root"), DATA_ROOT)
data_path = Record[RecordName, str](
RecordName("data"),
sep.join((str(data_root.old), str(OrgVal.old), str(AppVal.old))),
)
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.val, org.app.val))
@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

104
src/docker_compose/cfg/org.py Normal file
View File

@@ -0,0 +1,104 @@
from collections.abc import Iterator
from dataclasses import dataclass, field
from typing import Callable, ClassVar, Self, final, override
from docker_compose.cfg.org_yaml import OrgDataYaml
from docker_compose.cfg.record import Record, RecordCls, RecordName
@final
@dataclass(frozen=True, slots=True)
class OrgVal(RecordCls[str]):
old: ClassVar[RecordName] = RecordName("org")
@final
@dataclass(frozen=True, slots=True)
class Org:
val: str
replace: OrgVal = field(init=False)
def __post_init__(self) -> None:
setter = super(Org, self).__setattr__
setter("replace", OrgVal(self.val))
@final
@dataclass(frozen=True, slots=True)
class AppVal(RecordCls[str]):
old: ClassVar[RecordName] = RecordName("name")
@final
@dataclass(frozen=True, slots=True)
class App:
val: str
replace: AppVal = field(init=False)
def __post_init__(self) -> None:
setter = super(App, self).__setattr__
setter("replace", AppVal(self.val))
@final
@dataclass(frozen=True, slots=True)
class UrlValNew:
val: str | None
@override
def __str__(self) -> str:
if not self.val:
return ""
return ".".join((self.val, "ccamper7", "net"))
@final
@dataclass(frozen=True, slots=True)
class UrlVal(RecordCls[UrlValNew]):
old = RecordName("url")
@final
@dataclass(frozen=True, slots=True)
class Url:
val: str | None
replace: UrlVal = field(init=False)
def __post_init__(self) -> None:
setter = super(Url, self).__setattr__
setter("replace", UrlVal(UrlValNew(self.val)))
@final
@dataclass(frozen=True, slots=True)
class OrgData:
org_app = Record[RecordName, str](
RecordName(f"{OrgVal.old.val}_{AppVal.old.val}"),
f"{OrgVal.old!s}_{AppVal.old!s}",
)
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.replace
yield self.org.replace
yield self.url.replace
@property
def pre_render_funcs(self) -> Iterator[Callable[[str], str]]:
yield self.org_app
# def render_yaml(self, yaml: str) -> str:
# for func in self.render_funcs:
# yaml = func(yaml)
# return yaml

9
src/docker_compose/cfg/org_yaml.py Normal file
View File

@@ -0,0 +1,9 @@
from typing import Literal, NotRequired, TypedDict
class OrgDataYaml(TypedDict):
# org: str
url: NotRequired[str]
type OrgYaml = dict[Literal["ccamper7", "c4", "stryten"], OrgDataYaml]

38
src/docker_compose/cfg/record.py Normal file
View File

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

101
src/docker_compose/cfg/src_path.py Normal file
View File

@@ -0,0 +1,101 @@
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 T_YamlDict, T_YamlRW
from docker_compose.util.yaml_util import read_yaml, write_yaml
YAML_EXTS = frozenset((".yml", ".yaml"))
class ComposeFileTemplate(Path):
def write_dict(self, data: T_YamlDict) -> 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: T_YamlRW) -> 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

0
src/docker_compose/compose/__init__.py Normal file
View File

69
src/docker_compose/compose/compose.py Normal file
View File

@@ -0,0 +1,69 @@
from collections.abc import Iterator
from dataclasses import dataclass
from pathlib import Path
from typing import Self
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.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=str(OrgData.org_app.old),
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:
return self.cfg.pre_render(to_yaml(self.as_dict))
def write_template(self):
self.cfg.src_paths.compose_file.write(self.as_template)

12
src/docker_compose/compose/compose_yaml.py Normal file
View File

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

61
src/docker_compose/compose/net.py Normal file
View File

@@ -0,0 +1,61 @@
from collections.abc import Iterable, Iterator
from dataclasses import dataclass
from typing import Self, final, override
from docker_compose.cfg.org import OrgData
from docker_compose.compose.net_yaml import NetArgsYaml, NetYaml
from docker_compose.compose.services import Service
@final
@dataclass(frozen=True, slots=True)
class NetArgs:
name: str
@property
def as_dict(self) -> NetArgsYaml:
yaml_dict = NetArgsYaml(
name=str(self),
)
if self.external:
yaml_dict["external"] = self.external
return yaml_dict
# @property
# def as_key_dict(self) -> tuple[str, NetArgsYaml]:
# return str(self), self.as_dict
@override
def __str__(self) -> str:
return f"{OrgData.org_app.old!s}_{self.name}"
@property
def external(self) -> bool:
return self.name == "proxy"
@final
@dataclass
class Net:
data: frozenset[NetArgs]
@classmethod
def from_service_list(cls, args: Iterable[Service]) -> Self:
return cls.from_list(
frozenset(net for service in args for net in service.networks)
)
@classmethod
def from_list(cls, args: frozenset[str]) -> Self:
return cls(frozenset(NetArgs(arg) for arg in args))
@property
def as_dict(self) -> NetYaml:
return {net.name: net.as_dict for net in self.data}
@property
def proxys(self) -> Iterator[str]:
for net in self.data:
if not net.external:
continue
yield str(net)[:-6]

9
src/docker_compose/compose/net_yaml.py Normal file
View File

@@ -0,0 +1,9 @@
from typing import NotRequired, TypedDict
class NetArgsYaml(TypedDict):
name: str
external: NotRequired[bool]
type NetYaml = dict[str, NetArgsYaml]

54
src/docker_compose/compose/rendered.py Normal file
View File

@@ -0,0 +1,54 @@
from collections.abc import Iterator
from dataclasses import dataclass
from pathlib import Path
from typing import final
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)

102
src/docker_compose/compose/services.py Normal file
View File

@@ -0,0 +1,102 @@
from dataclasses import dataclass
from typing import Self, final, override
from docker_compose.cfg.compose_paths import ServicePath, ServiceVal
from docker_compose.cfg.org import OrgData, UrlVal
from docker_compose.compose.services_yaml import (
HealthCheck,
ServiceYamlRead,
ServiceYamlWrite,
)
from docker_compose.util.Ts import T_Primitive
@final
@dataclass(frozen=True, slots=True)
class Service:
_traefik_labels = frozenset(
(
f"traefik.http.routers.{OrgData.org_app.old!s}.rule=Host(`{UrlVal.old!s}`)",
f"traefik.http.routers.{OrgData.org_app.old!s}.entrypoints=websecure",
f"traefik.docker.network={OrgData.org_app.old!s}_proxy",
f"traefik.http.routers.{OrgData.org_app.old!s}.tls.certresolver=le",
)
)
@override
def __hash__(self) -> int:
return hash(self.service_name)
@property
def service_name(self) -> str:
return self.service_val.new.val.stem
_sec_opts = frozenset(("no-new-privileges:true",))
# service_name: str
service_val: ServiceVal
command: tuple[str, ...]
entrypoint: tuple[str, ...]
environment: dict[str, T_Primitive]
image: str
labels: frozenset[str]
logging: dict[str, str]
networks: frozenset[str]
restart: str
security_opt: frozenset[str]
user: str | None
volumes: frozenset[str]
shm_size: str | None
depends_on: frozenset[str]
healthcheck: HealthCheck | None
ports: frozenset[str]
@classmethod
def from_path(cls, path: ServicePath) -> Self:
return cls.from_dict(path.replace, path.as_dict)
@classmethod
def from_dict(cls, service_val: ServiceVal, data: ServiceYamlRead) -> Self:
# helper = ServiceYamlProps(data)
labels = frozenset(data.get("labels", ()))
# ports = (f'"{p}"' for p in data.get("ports", ()))
return cls(
service_val,
tuple(data.get("command", ())),
tuple(data.get("entrypoint", ())),
data.get("environment", {}),
data["image"],
cls._traefik_labels.union(labels)
if "traefik.enable=true" in labels
else labels,
data.get("logging", {}),
frozenset(data.get("networks", ())),
"unless-stopped",
cls._sec_opts.union(data.get("security_opt", [])),
data.get("user"),
frozenset(data.get("volumes", ())),
data.get("shm_size"),
frozenset(data.get("depends_on", ())),
data.get("healthcheck"),
frozenset(data.get("ports", ())),
)
@property
def as_dict(self) -> ServiceYamlWrite:
return ServiceYamlWrite(
command=self.command,
entrypoint=self.entrypoint,
environment=self.environment,
image=self.image,
labels=self.labels,
logging=self.logging,
networks=self.networks,
security_opt=self.security_opt,
user=self.user,
volumes=self.volumes,
container_name=self.service_val(ServicePath.fqdn.new),
restart=self.restart,
shm_size=self.shm_size,
depends_on=self.depends_on,
healthcheck=self.healthcheck,
ports=self.ports,
)

47
src/docker_compose/compose/services_yaml.py Normal file
View File

@@ -0,0 +1,47 @@
from typing import NotRequired, TypedDict
from docker_compose.util.Ts import T_Primitive
class HealthCheck(TypedDict):
test: list[str] | str
interval: NotRequired[str]
timeout: NotRequired[str]
retries: NotRequired[int]
start_period: NotRequired[str]
class ServiceYamlRead(TypedDict):
command: NotRequired[list[str]]
entrypoint: NotRequired[list[str]]
environment: NotRequired[dict[str, T_Primitive]]
image: str
labels: NotRequired[list[str]]
logging: NotRequired[dict[str, str]]
networks: NotRequired[list[str]]
security_opt: NotRequired[list[str]]
user: NotRequired[str]
volumes: NotRequired[list[str]]
shm_size: NotRequired[str]
depends_on: NotRequired[list[str]]
healthcheck: NotRequired[HealthCheck]
ports: NotRequired[list[str]]
class ServiceYamlWrite(TypedDict):
command: tuple[str, ...]
entrypoint: tuple[str, ...]
environment: dict[str, T_Primitive]
image: str
labels: frozenset[str]
logging: dict[str, str]
networks: frozenset[str]
security_opt: frozenset[str]
user: str | None
volumes: frozenset[str]
container_name: str
restart: str
shm_size: str | None
depends_on: frozenset[str]
healthcheck: HealthCheck | None
ports: frozenset[str]

3
src/docker_compose/compose/volume_yaml.py Normal file
View File

@@ -0,0 +1,3 @@
from docker_compose.util.Ts import T_YamlDict
type VolYaml = dict[str, T_YamlDict]

0
src/docker_compose/compose/volumes.py Normal file
View File

99
src/docker_compose/util/Ts.py Normal file
View File

@@ -0,0 +1,99 @@
from collections.abc import ItemsView, Iterator, KeysView, MutableMapping, Set
from types import GenericAlias, UnionType
from typing import (
ClassVar,
Never,
Protocol,
TypeAliasType,
cast,
get_args,
get_origin,
overload,
)
class TypedYamlDict[K: object, V: object](Protocol):
def __getitem__(self, key: str | K, /) -> V: ...
# def __setitem__(self, key: str, value: V, /) -> V: ...
def __delitem__(self, key: Never | K, /) -> None: ...
def __contains__(self, key: K, /) -> bool: ...
def __iter__(self) -> Iterator[K]: ...
def __len__(self) -> int: ...
def keys(self) -> KeysView[K]: ...
def items(self) -> ItemsView[K, V]: ...
def pop(self, key: Never | K, /) -> V: ...
# def popitem(self) -> tuple[K, V]: ...
# def clear(self) -> None: ...
__required_keys__: ClassVar[frozenset[str]]
__optional_keys__: ClassVar[frozenset[str]]
# class Test(TypedDict):
# var: str
#
#
# x = Test(var="test")
#
#
# def is_typed_dict_test(obj: TypedYamlDict[object, object]) -> None:
# print(obj)
# pass
#
#
# is_typed_dict_test(x)
type T_Primitive = None | bool | int | str
type T_PrimIters = tuple[T_Prim, ...] | list[T_Prim] | Set[T_Prim] | Iterator[T_Prim]
type T_PrimDict = MutableMapping[T_Primitive, T_Prim]
type T_Prim = T_Primitive | T_PrimIters | T_PrimDict
type T_YamlIters = tuple[T_Yaml, ...] | list[T_Yaml] | Set[T_Yaml] | Iterator[T_Yaml]
type T_YamlDict = MutableMapping[str, T_Yaml]
type T_YamlRW = T_YamlIters | T_YamlDict
type T_Yaml = T_Primitive | T_YamlRW
type T_YamlPostDict = TypedYamlDict[str, T_YamlPost]
type T_YamlPostRes = tuple[T_YamlPost, ...] | T_YamlPostDict
type T_YamlPost = T_Primitive | T_YamlPostRes
def get_union_types(annotations: UnionType) -> Iterator[type]:
for annotation in get_args(annotations): # pyright: ignore[reportAny]
if isinstance(annotation, TypeAliasType):
annotation = annotation.__value__ # pyright: ignore[reportAny]
if isinstance(annotation, UnionType):
yield from get_union_types(annotation)
continue
yield get_types(annotation) # pyright: ignore[reportAny]
@overload
def get_types(annotation: UnionType) -> tuple[type]:
pass
@overload
def get_types(annotation: GenericAlias) -> type:
pass
@overload
def get_types(annotation: TypeAliasType) -> type | tuple[type, ...]:
pass
def get_types(
annotation: TypeAliasType | GenericAlias | UnionType,
) -> type | tuple[type, ...]:
if isinstance(annotation, TypeAliasType):
annotation = annotation.__value__ # pyright: ignore[reportAny]
if isinstance(annotation, GenericAlias):
return get_origin(annotation)
if isinstance(annotation, UnionType):
return tuple(get_union_types(annotation))
return cast(type, annotation) # pyright: ignore[reportInvalidCast]

29
src/docker_compose/util/__init__.py Normal file
View File

@@ -0,0 +1,29 @@
# from collections.abc import Iterator, Mapping
# from typing import Any, cast
#
# from docker_compose.util.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
# ) -> Iterator[tuple[T_Primitive, T_PrimVal]]:
# 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)))
#

157
src/docker_compose/util/yaml_util.py Normal file
View File

@@ -0,0 +1,157 @@
import re
from collections.abc import Iterator, MutableMapping, Set
from pathlib import Path
from typing import cast, get_type_hints, is_typeddict, override
import yaml
from docker_compose.util.Ts import (
T_YamlDict,
T_YamlIters,
T_YamlPost,
T_YamlPostDict,
T_YamlPostRes,
T_YamlRW,
TypedYamlDict,
get_types,
)
# class TypedYamlDict[K: object, V: object](Protocol):
# def __getitem__(self, key: K, /) -> V: ...
# # def __setitem__(self, key: K, value: V, /) -> V: ...
# def __delitem__(self, key: K, /) -> V: ...
# def __contains__(self, key: K, /) -> bool: ...
# def __iter__(self) -> Iterator[K]: ...
# def __len__(self) -> int: ...
# def keys(self) -> KeysView[K]: ...
# def items(self) -> ItemsView[K, V]: ...
# def pop(self, key: K, /) -> V: ...
#
# # def popitem(self) -> tuple[K, V]: ...
#
# # def clear(self) -> None: ...
#
# __required_keys__: ClassVar[frozenset[str]]
# __optional_keys__: ClassVar[frozenset[str]]
class VerboseSafeDumper(yaml.SafeDumper):
@override
def ignore_aliases(self, data: object) -> bool:
return True
def yaml_prep(data: T_YamlRW) -> T_YamlPostRes:
if isinstance(data, MutableMapping):
return dict_prep(data)
if isinstance(data, (tuple, list)):
return tuple(list_prep(data))
res = tuple(list_prep(data))
try:
return tuple(sorted(res)) # pyright: ignore[reportArgumentType, reportUnknownArgumentType, reportUnknownVariableType]
except TypeError:
return res
def list_prep(data: T_YamlIters) -> Iterator[T_YamlPost]:
for v in data:
if isinstance(v, (MutableMapping, tuple, list, Set, Iterator)):
yield yaml_prep(v)
continue
if v:
yield v
continue
if isinstance(v, bool):
yield v
continue
def dict_prep(data: T_YamlDict) -> T_YamlPostDict:
keys = tuple(data.keys())
for k in keys:
v = data[k]
if isinstance(v, (MutableMapping, tuple, list, Set, Iterator)):
data[k] = v = yaml_prep(v) # pyright: ignore[reportArgumentType]
if v:
continue
if isinstance(v, bool):
continue
del data[k]
return cast(T_YamlPostDict, cast(object, data))
def to_yaml(data: T_YamlRW) -> str:
dict_ = yaml_prep(data)
res = yaml.dump(dict_, Dumper=VerboseSafeDumper)
res = re.sub(r"(^\s?-)", r" \g<1>", res, flags=re.MULTILINE)
return re.sub(r"(\W*?)(\d+:\d+)", r'\g<1>"\g<2>"', res, flags=re.MULTILINE)
def write_yaml(
data: T_YamlRW,
path: Path,
) -> None:
with path.open("wt") as f:
_ = f.write(to_yaml(data))
def read_yaml(path: Path) -> T_YamlPostRes:
with path.open("rt") as f:
return yaml.safe_load(f) # pyright: ignore[reportAny]
def read_typed_yaml[T: TypedYamlDict[object, object]](
type_: type[T],
path: Path,
) -> T:
with path.open("rt") as f:
data: T_YamlDict = yaml.safe_load(f) # pyright: ignore[reportAny]
path_to_typed(type_, data, path)
return cast(T, data) # pyright: ignore[reportInvalidCast]
def path_to_typed(
type_: type[TypedYamlDict[object, object]],
data: T_YamlDict,
path: Path,
) -> None:
try:
validate_typed_dict(type_, data)
except (KeyError, TypeError) as e:
e.add_note(f"path: {path!s}")
raise e
def validate_typed_dict(
t: type[TypedYamlDict[object, object]],
data: T_YamlDict,
) -> None:
keys = frozenset(data.keys())
missing = t.__required_keys__.difference(keys)
if missing:
raise KeyError(f"missing required key(s): {', '.join(missing)}")
extra = keys.difference(t.__required_keys__, t.__optional_keys__)
if extra:
raise KeyError(f"extra key(s): {', '.join(map(str, extra))}")
hints = get_type_hints(t)
for key, val in data.items():
t2 = hints[key] # pyright: ignore[reportAny]
if is_typeddict(t2): # pyright: ignore[reportAny]
validate_typed_dict(t2, cast(T_YamlDict, val)) # pyright: ignore[reportAny]
continue
# try:
if not isinstance(val, get_types(t2)): # pyright: ignore[reportAny]
raise TypeError(
f"key: {key} expected *{type(t2).__name__}*, got *{type(val).__name__}*" # pyright: ignore[reportAny]
)
# valid = isinstance(val, get_types(t2))
# except TypeError:
# valid = isinstance(val, get_origin(t2))
# if not valid:
# raise TypeError(
# f"key: {key} expected *{type(t2).__name__}*, got *{type(val).__name__}*"
# )
# yield key, val