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

sync

Browse Source
This commit is contained in:
Christian Camper
2025-12-18 00:00:53 -06:00
parent fa4339768f
commit 6aa6c9e4dd
47 changed files with 752 additions and 894 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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)

8
src/docker_compose/Ts.py Normal file
View File

@@ -0,0 +1,8 @@
from collections.abc import Mapping
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]

4
src/docker_compose/__init__.py
View File

@@ -1,7 +1,7 @@
from collections.abc import Iterable, Iterator
from compose.cfg import CFG_ROOT, TRAEFIK_PATH
from compose.compose.render import Rendered
from docker_compose.cfg import CFG_ROOT, TRAEFIK_PATH
from docker_compose.compose.render import Rendered
def load_all() -> Iterable[Rendered]:

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

@@ -0,0 +1,30 @@
from dataclasses import dataclass
from pathlib import Path
from typing import Self, final
from docker_compose.cfg.cfg_paths_yaml import CfgYaml
from docker_compose.cfg.org_data import OrgData
from docker_compose.cfg.src_path import SrcPaths
@final
@dataclass(frozen=True, slots=True)
class CfgData:
src_paths: SrcPaths
name: str
services: frozenset[Path]
volumes: frozenset[Path] | None
org_data: frozenset[OrgData]
@classmethod
def from_src_paths(cls, src_paths: SrcPaths) -> Self:
cfg_yaml = CfgYaml.from_src_paths(src_paths)
cfg_root = src_paths.cfg_dir.joinpath
vols = cfg_yaml.data.get("volumes")
return cls(
src_paths,
src_paths.cfg_dir.name,
frozenset(cfg_root(path) for path in cfg_yaml.data["services"]),
frozenset(cfg_root(path) for path in vols) if vols else None,
frozenset(cfg_yaml.orgs_data),
)

27
src/docker_compose/cfg/cfg_paths_yaml.py Normal file
View File

@@ -0,0 +1,27 @@
from collections.abc import Iterator
from dataclasses import dataclass
from typing import NotRequired, Self, TypedDict, final
from docker_compose.cfg.org_data import OrgData
from docker_compose.cfg.org_data_yaml import OrgDataYaml
from docker_compose.cfg.src_path import SrcPaths
from docker_compose.yaml import YamlWrapper
class CfgYamlData(TypedDict):
services: list[str]
volumes: NotRequired[list[str]]
orgs: list[OrgDataYaml]
@final
@dataclass(frozen=True, slots=True)
class CfgYaml(YamlWrapper[CfgYamlData]):
@classmethod
def from_src_paths(cls, src_paths: SrcPaths) -> Self:
return cls.from_path(src_paths.cfg_file)
@property
def orgs_data(self) -> Iterator[OrgData]:
for org in self.data["orgs"]:
yield OrgData.from_dict(org)

15
src/docker_compose/cfg/org_data.py Normal file
View File

@@ -0,0 +1,15 @@
from dataclasses import dataclass
from typing import Self, final
from docker_compose.cfg.org_data_yaml import OrgDataYaml
@final
@dataclass(frozen=True, slots=True)
class OrgData:
org: str
url: str | None
@classmethod
def from_dict(cls, data: OrgDataYaml) -> Self:
return cls(data["org"], data.get("url"))

6
src/docker_compose/cfg/org_data_yaml.py Normal file
View File

@@ -0,0 +1,6 @@
from typing import NotRequired, TypedDict
class OrgDataYaml(TypedDict):
org: str
url: NotRequired[str]

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

@@ -0,0 +1,30 @@
from dataclasses import dataclass
from pathlib import Path
from typing import Self, final
@final
@dataclass(frozen=True, slots=True)
class SrcPaths:
cfg_dir: Path
cfg_file: Path
env_file: Path
@classmethod
def from_path(cls, src: Path) -> Self:
return cls(
cfg_dir=src,
cfg_file=src.joinpath("cfg.yml"),
env_file=src.joinpath(".env"),
)
# 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)

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

@@ -0,0 +1,75 @@
from abc import ABCMeta
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.src_path import SrcPaths
from docker_compose.compose.compose_yaml import ComposeYaml, ComposeYamlData
from docker_compose.compose.net import Net
from docker_compose.compose.replace_args import ReplaceArgs
from docker_compose.compose.service import Service
from docker_compose.compose.volumes_yaml import VolYaml
@dataclass(frozen=True, slots=True)
class Compose(metaclass=ABCMeta):
cfg: CfgData
services: dict[str, Service]
networks: Net | None
volumes: dict[str, VolYaml] | None
replace_args: frozenset[ReplaceArgs]
@classmethod
def from_path(cls, path: Path):
return cls.from_src_path(SrcPaths.from_path(path))
@classmethod
def from_src_path(cls, src_paths: SrcPaths) -> Self:
return cls.from_cfg(CfgData.from_src_paths(src_paths))
@classmethod
def from_cfg(cls, cfg_data: CfgData) -> Self:
services = {path.stem: Service.from_path(path) for path in cfg_data.services}
vols = {path.stem: VolYaml.from_path(path) for path in cfg_data.services}
nets = Net.from_service_list(services.values())
return cls(
cfg_data,
services,
nets if nets else None,
vols if vols else None,
frozenset(
ReplaceArgs.from_cfg_data(cfg_data, org) for org in cfg_data.org_data
),
)
@property
def as_dict(self) -> ComposeYaml:
data = ComposeYamlData(
name=self.cfg.name,
services={k: v.as_dict.data for k, v in self.services.items()},
)
if self.networks:
data["networks"] = self.networks.as_dict
if self.volumes:
data["volumes"] = {k: v.data for k, v in self.volumes.items()}
return ComposeYaml(data)
@property
def as_yaml(self) -> str:
return self.as_dict.as_yaml
@property
def proxys(self) -> Iterator[str]:
proxy = self.networks
if proxy is None:
return
for net in proxy.data.values():
if not net.is_proxy:
return
yield net.name
# nets = frozenset(net.name for net in proxy.data.values() if net.is_proxy)
# return nets if nets else None

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

@@ -0,0 +1,20 @@
from dataclasses import dataclass
from typing import NotRequired, TypedDict, final
from docker_compose.compose.net_yaml import NetYaml
from docker_compose.compose.service_yaml_write import ServiceYamlWriteData
from docker_compose.compose.volumes_yaml import VolYamlData
from docker_compose.yaml import YamlWrapper
class ComposeYamlData(TypedDict):
name: str
services: dict[str, ServiceYamlWriteData]
networks: NotRequired[NetYaml]
volumes: NotRequired[dict[str, VolYamlData]]
@final
@dataclass(frozen=True, slots=True)
class ComposeYaml(YamlWrapper[ComposeYamlData]):
pass

35
src/docker_compose/compose/dest_path.py Normal file
View File

@@ -0,0 +1,35 @@
from dataclasses import dataclass
from pathlib import Path
from typing import Self, final
@final
@dataclass(frozen=True, slots=True)
class DestPaths:
data_dir: Path
env_file: Path
compose_file: Path
@classmethod
def from_path(cls, path: Path) -> Self:
return cls(
path,
path.joinpath(".env"),
path.joinpath("docker-docker_compose.yml"),
)
# @staticmethod
# def _mk_dir(path: Path) -> None:
# if path.exists():
# return
# path.mkdir(parents=True)
def mk_compose_dir(self) -> None:
if self.data_dir.exists():
return
self.data_dir.mkdir(parents=True)
# vols = self.bind_vols
# if vols is None:
# return
# for vol in vols:
# _mk_dir(vol)

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

@@ -0,0 +1,46 @@
from collections.abc import Iterable
from dataclasses import dataclass
from typing import final
from docker_compose.compose.net_args import NetArgs
from docker_compose.compose.net_yaml import NetYaml
from docker_compose.compose.service import Service
# @final
# @dataclass(frozen=True, slots=True)
# class Net:
# internal: NetArgs | None
# proxy: NetArgs | None
# @property
# def as_dict(self) -> NetYaml:
# yaml_dict = NetYaml()
# if self.internal is not None:
# yaml_dict["internal"] = self.internal.as_dict
# if self.proxy is not None:
# yaml_dict["proxy"] = self.proxy.as_dict
# return yaml_dict
@final
@dataclass
class Net:
data: dict[str, NetArgs]
@classmethod
def from_service_list(cls, args: Iterable[Service]):
return cls.from_list(
{net for service in args if service.networks for net in service.networks}
)
@classmethod
def from_list(cls, args: Iterable[str]):
return cls({net: NetArgs(net, NetArgs.is_proxy_check(net)) for net in args})
@property
def as_dict(self) -> NetYaml:
return {name: net_args.as_dict for name, net_args in self.data.items()}
def __bool__(self) -> bool:
return bool(self.data)

28
src/docker_compose/compose/net_args.py Normal file
View File

@@ -0,0 +1,28 @@
from dataclasses import dataclass
from typing import final
from docker_compose.compose.net_args_yaml import NetArgsYaml
@final
@dataclass(frozen=True, slots=True)
class NetArgs:
name: str
external: bool | None
@property
def as_dict(self) -> NetArgsYaml:
yaml_dict = NetArgsYaml(
name=self.name,
)
if self.external is not None:
yaml_dict["external"] = self.external
return yaml_dict
@staticmethod
def is_proxy_check(name: str) -> bool:
return name.endswith("proxy")
@property
def is_proxy(self) -> bool:
return self.is_proxy_check(self.name)

6
src/docker_compose/compose/net_args_yaml.py Normal file
View File

@@ -0,0 +1,6 @@
from typing import NotRequired, TypedDict
class NetArgsYaml(TypedDict):
name: str
external: NotRequired[bool]

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

@@ -0,0 +1,8 @@
from docker_compose.compose.net_args_yaml import NetArgsYaml
# class NetYaml(TypedDict):
# internal: NotRequired[NetArgsYaml]
# proxy: NotRequired[NetArgsYaml]
type NetYaml = dict[str, NetArgsYaml]

39
src/docker_compose/compose/render.py Normal file
View File

@@ -0,0 +1,39 @@
from collections.abc import Iterator
from dataclasses import dataclass
from pathlib import Path
from typing import final
from docker_compose.compose.compose import Compose
from docker_compose.compose.replace_args import ReplaceArgs
@final
@dataclass(frozen=True, slots=True)
class Rendered(Compose):
def mk_bind_vols(self) -> None:
for app_data in self.services.values():
if app_data.volumes is None:
continue
for vol in app_data.volumes:
for arg in self.replace_args:
path = arg.render_yaml(vol.split(":", 1)[0])
if not path.startswith("/"):
continue
path = Path(path)
if path.exists():
continue
path.mkdir(parents=True)
@property
def proxy_nets(self) -> Iterator[str]:
for net in self.proxys:
for re in self.replace_args:
yield re.org_name.replace(net)
def write_all(self) -> None:
self.mk_bind_vols()
for arg in self.replace_args:
arg.write_yaml(self.as_yaml)
def write(self, args: ReplaceArgs) -> None:
args.write_yaml(self.as_yaml)

73
src/docker_compose/compose/replace_args.py Normal file
View File

@@ -0,0 +1,73 @@
from dataclasses import dataclass
from functools import reduce
from shutil import copyfile
from typing import Self, final
from docker_compose.cfg import DATA_ROOT
from docker_compose.cfg.cfg_paths import CfgData
from docker_compose.cfg.org_data import OrgData
from docker_compose.compose.dest_path import DestPaths
from docker_compose.compose.val_obj import (
DataDir,
NameVal,
OrgVal,
Record,
Url,
)
@final
@dataclass(frozen=True, slots=True)
class ReplaceArgs:
cfg: CfgData
org: Record[OrgVal]
name: Record[NameVal]
org_name: Record[NameVal]
data: Record[DataDir]
url: Record[Url]
dest_paths: DestPaths
# noinspection PyMissingTypeHints
def __iter__(self):
yield self.org
yield self.name
yield self.org_name
yield self.data
yield self.url
@classmethod
def from_cfg_data(cls, cfg_data: CfgData, org_data: OrgData) -> Self:
_org = OrgVal(org_data.org)
_name = NameVal(cfg_data.name)
org_name = NameVal(f"{_org.str}_{_name.str}") if _org.is_valid() else _name
data_dir = DATA_ROOT.joinpath(_org.str, _name.str)
return cls(
cfg_data,
Record("org", _org),
Record("name", _name),
Record("org_name", org_name),
Record("data", DataDir(data_dir)),
Record("url", Url(org_data.url)),
DestPaths.from_path(data_dir),
)
def mk_compose_env(self) -> None:
src = self.cfg.src_paths.env_file
dest = self.dest_paths.env_file
if src.exists() and not dest.exists():
_ = copyfile(src, dest)
def render_yaml(self, yaml: str) -> str:
return reduce(lambda s, f: f.replace(s), self, yaml)
def write_yaml(self, yaml: str) -> None:
self.dest_paths.mk_compose_dir()
with self.dest_paths.compose_file.open("wt") as f:
_ = f.write(self.render_yaml(yaml))
# def mk_vol_dir(self, path: str):
# p = Path(self.render_yaml(path))
# if p.exists():
# return
# p.mkdir(parents=True)

80
src/docker_compose/compose/service.py Normal file
View File

@@ -0,0 +1,80 @@
from abc import ABCMeta
from dataclasses import dataclass
from pathlib import Path
from typing import Self, final
from docker_compose.compose.service_yaml_read import (
ServiceYamlRead,
)
from docker_compose.compose.service_yaml_write import (
ServiceYamlWrite,
ServiceYamlWriteData,
)
from docker_compose.compose.val_obj import Record
from docker_compose.Ts import T_Primitive
@final
@dataclass(frozen=True, slots=True)
class Service(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[str] | None
restart: str
security_opt: frozenset[str]
user: str | None
volumes: frozenset[str] | None
@classmethod
def from_path(cls, path: Path) -> Self:
return cls.from_dict(ServiceYamlRead.from_path(path))
@classmethod
def from_dict(cls, data: ServiceYamlRead):
command = data.data.get("command")
entrypoint = data.data.get("entrypoint")
volumes = data.data.get("volumes")
nets = data.data.get("networks")
return cls(
None if not command else tuple(command),
Record.get_replace_name("org_name"),
tuple(entrypoint) if entrypoint else None,
data.data.get("environment"),
data.data["image"],
data.labels,
data.data.get("logging"),
frozenset(nets) if nets else None,
"unless-stopped",
data.sec_opts,
data.data.get("user"),
frozenset(volumes) if volumes else None,
)
@property
def as_dict(self) -> ServiceYamlWrite:
data = ServiceYamlWriteData(
container_name=self.container_name,
image=self.image,
restart=self.restart,
security_opt=sorted(self.security_opt),
)
if self.command is not None:
data["command"] = list(self.command)
if self.entrypoint is not None:
data["entrypoint"] = list(self.entrypoint)
if self.environment is not None:
data["environment"] = self.environment
if self.labels is not None:
data["labels"] = sorted(self.labels)
if self.logging is not None:
data["logging"] = self.logging
if self.user is not None:
data["user"] = self.user
if self.volumes is not None:
data["volumes"] = sorted(self.volumes)
return ServiceYamlWrite(data)

60
src/docker_compose/compose/service_yaml_read.py Normal file
View File

@@ -0,0 +1,60 @@
from dataclasses import dataclass
from typing import Literal, NotRequired, TypedDict
from docker_compose.compose.val_obj import Record
from docker_compose.Ts import T_Primitive
from docker_compose.yaml import YamlWrapper
type T_NetAbc = str | Literal["proxy", "internal"]
class ServiceYamlReadData(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]]
@dataclass(frozen=True, slots=True)
class ServiceYamlRead(YamlWrapper[ServiceYamlReadData]):
@property
def sec_opts(self) -> frozenset[str]:
sec_opts = frozenset(
"no-new-privileges:true",
)
sec = self.data.get("security_opt")
if not sec:
return sec_opts
return sec_opts.union(sec)
@property
def labels(self) -> frozenset[str] | None:
org_name = Record.get_replace_name("org_name")
url = Record.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 = self.data.get("labels")
if not labels:
return
if "traefik.enable=true" not in labels:
return frozenset(labels)
return traefik_labels.union(labels)
@property
def nets(self) -> frozenset[str] | None:
nets = self.data.get("networks")
if nets is None:
return
return frozenset(nets)

15
src/docker_compose/compose/service_yaml_write.py Normal file
View File

@@ -0,0 +1,15 @@
from dataclasses import dataclass
from typing import NotRequired
from docker_compose.compose.service_yaml_read import ServiceYamlReadData
from docker_compose.yaml import YamlWrapper
class ServiceYamlWriteData(ServiceYamlReadData):
container_name: NotRequired[str]
restart: NotRequired[str]
@dataclass(frozen=True, slots=True)
class ServiceYamlWrite(YamlWrapper[ServiceYamlWriteData]):
pass

81
src/docker_compose/compose/val_obj.py Normal file
View File

@@ -0,0 +1,81 @@
from abc import ABCMeta, abstractmethod
from dataclasses import dataclass
from pathlib import Path
from typing import final, override
@dataclass(frozen=True, slots=True)
class RecordVal(metaclass=ABCMeta):
@property
@abstractmethod
def str(self) -> str:
pass
@final
@dataclass(frozen=True, slots=True)
class Record[T: RecordVal]:
name: str
val: T
def replace(self, string: str) -> str:
return string.replace(self.replace_name, self.val.str)
@property
def replace_name(self) -> str:
return self.get_replace_name(self.name)
@staticmethod
def get_replace_name(string: str) -> str:
return f"${{_{string.upper()}}}"
@final
@dataclass(frozen=True, slots=True)
class OrgVal(RecordVal):
val: str | None
@property
@override
def 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(RecordVal):
val: str
@property
@override
def str(self) -> str:
return self.val
@final
@dataclass(frozen=True, slots=True)
class DataDir(RecordVal):
path: Path
@property
@override
def str(self) -> str:
return str(self.path)
@final
@dataclass(frozen=True, slots=True)
class Url(RecordVal):
sub_url: str | None
@property
@override
def str(self) -> str:
if self.sub_url is None:
return ""
return ".".join([self.sub_url, "ccamper7", "net"])

25
src/docker_compose/compose/volumes_yaml.py Normal file
View File

@@ -0,0 +1,25 @@
from dataclasses import dataclass
from typing import final
from docker_compose.Ts import T_YamlDict
from docker_compose.yaml import YamlWrapper
type VolYamlData = dict[str, T_YamlDict]
@final
@dataclass(frozen=True, slots=True)
class VolYaml(YamlWrapper[VolYamlData]):
pass
# def vols_from_path(path: Path) -> VolYamlData:
# return cast(VolYamlData, read_yml(path))
# def vols_yaml_factory(self) -> Iterator[tuple[str, VolDataYaml]]:
# vols = self.volumes
# if vols is None:
# return
# for path in vols:
# yield path.stem, cast(VolDataYaml, read_yml(path))

2
src/docker_compose/util.py
View File

@@ -1,7 +1,7 @@
from collections.abc import Mapping
from typing import Any, cast
from compose.Ts import T_PrimDict, T_Primitive, T_PrimVal
from docker_compose.Ts import T_PrimDict, T_Primitive, T_PrimVal
def merge_dicts[T: Mapping[Any, Any]](dict1: T, dict2: T) -> T:

42
src/docker_compose/yaml.py Normal file
View File

@@ -0,0 +1,42 @@
import re
from collections.abc import ItemsView, Iterator, KeysView
from dataclasses import dataclass
from pathlib import Path
from typing import ClassVar, Protocol, cast, override
import yaml
class ProtoMapping[K, V: object](Protocol):
def __getitem__(self, key: K, /) -> V: ...
def __iter__(self) -> Iterator[K]: ...
def __len__(self) -> int: ...
def __contains__(self, key: object, /) -> bool: ...
def keys(self) -> KeysView[K]: ...
def items(self) -> ItemsView[K, V]: ...
class TTypedyamldict(ProtoMapping[str, object], Protocol):
__required_keys__: ClassVar[frozenset[str]]
__optional_keys__: ClassVar[frozenset[str]]
class VerboseSafeDumper(yaml.SafeDumper):
@override
def ignore_aliases(self, data: object) -> bool:
return True
@dataclass(frozen=True, slots=True)
class YamlWrapper[T: ProtoMapping[str, object]]:
data: T
@classmethod
def from_path(cls, path: Path):
with path.open("rt") as f:
return cls(cast(T, yaml.safe_load(f)))
@property
def as_yaml(self) -> str:
_yaml = yaml.dump(self.data, Dumper=VerboseSafeDumper)
return re.sub(r"(^\s*-)", r" \g<1>", _yaml, flags=re.MULTILINE)