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-15 01:03:25 -06:00
parent cef96db21a
commit 9d3ce193e8
4 changed files with 117 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -3,9 +3,9 @@ from typing import cast
from compose.cfg.entity import CfgData, CfgDataYaml, OrgData from compose.cfg.entity import CfgData, CfgDataYaml, OrgData
from compose.compose.entity import VolYaml from compose.compose.entity import VolYaml
from compose.service.entity import Service from compose.service.entity import Service, T_Compose
from compose.service.factory import services_yaml_factory from compose.service.factory import services_yaml_factory
from compose.util import read_yml from compose.util import get_replace_name, read_yml
def cfg_get_orgs(data: CfgDataYaml) -> Iterator[OrgData]: def cfg_get_orgs(data: CfgDataYaml) -> Iterator[OrgData]:
@@ -16,10 +16,63 @@ def cfg_get_orgs(data: CfgDataYaml) -> Iterator[OrgData]:
) )
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]]: def cfg_get_services(cfg_data: CfgData) -> Iterator[tuple[str, Service]]:
for path in cfg_data.services: for path in cfg_data.services:
_dict = services_yaml_factory(path) data = services_yaml_factory(path)
yield path.stem, Service.from_dict(_dict) # 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]]: def cfg_get_volumes(cfg_data: CfgData) -> Iterator[tuple[str, VolYaml]]:

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

@@ -1,11 +1,10 @@
from dataclasses import asdict, dataclass from dataclasses import asdict, dataclass
from typing import Literal, NotRequired, Self, TypedDict, final from typing import Literal, NotRequired, TypedDict, final
from compose.cfg import T_YamlDict from compose.cfg import T_YamlDict
from compose.cfg.entity import CfgData from compose.cfg.entity import CfgData
from compose.net.entities import Net, NetTraefik, NetYaml from compose.net.entities import Net, NetTraefik, NetYaml
from compose.service.entity import Service, ServiceYaml, TraefikService from compose.service.entity import Service, ServiceYaml, TraefikService
from compose.service.get import services_get_networks
from compose.util import to_yaml from compose.util import to_yaml
type VolYaml = dict[str, T_YamlDict] type VolYaml = dict[str, T_YamlDict]
@@ -26,25 +25,25 @@ class Compose:
networks: Net | None networks: Net | None
volumes: VolYaml | None volumes: VolYaml | None
@classmethod # @classmethod
def from_dict(cls, cfg: CfgData, data: ComposeYaml) -> Self: # def from_dict(cls, cfg: CfgData, data: ComposeYaml) -> Self:
# services = dict[str, ComposeService]() # # services = dict[str, ComposeService]()
services = dict(_get_services_dict(data)) # services = dict(_get_services_dict(data))
# vols = frozenset(_get_volumes_dict(data)) # # vols = frozenset(_get_volumes_dict(data))
return cls( # return cls(
cfg, # cfg,
services, # services,
services_get_networks(services.values()), # services_get_networks(services.values()),
data.get("volumes"), # data.get("volumes"),
) # )
def as_yaml(self) -> str: def as_yaml(self) -> str:
return to_yaml(asdict(self)) return to_yaml(asdict(self))
def _get_services_dict(data: ComposeYaml): # def _get_services_dict(data: ComposeYaml):
for k, v in data["services"].items(): # for k, v in data["services"].items():
yield k, Service.from_dict(v) # yield k, Service.from_dict(v)
# def _get_volumes_dict(data: ComposeYaml) -> Iterator[VolYaml]: # def _get_volumes_dict(data: ComposeYaml) -> Iterator[VolYaml]:

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

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

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

@@ -1,9 +1,8 @@
from abc import ABCMeta, abstractmethod from abc import ABCMeta, abstractmethod
from dataclasses import dataclass from dataclasses import dataclass
from typing import Literal, NotRequired, Self, TypedDict, TypeVar, overload, override from typing import Literal, NotRequired, TypedDict, TypeVar, overload, override
from compose.cfg import T_Primitive from compose.cfg import T_Primitive
from compose.util import get_replace_name
type T_NetAbc = str | Literal["proxy", "internal"] type T_NetAbc = str | Literal["proxy", "internal"]
TCo_NetABC = TypeVar("TCo_NetABC", bound=T_NetAbc, covariant=True) TCo_NetABC = TypeVar("TCo_NetABC", bound=T_NetAbc, covariant=True)
@@ -57,35 +56,35 @@ class ServiceAbc[T_net: T_NetAbc, T_Yaml: T_Compose](metaclass=ABCMeta):
user: str | None user: str | None
volumes: frozenset[str] | None volumes: frozenset[str] | None
@classmethod # @classmethod
def from_dict(cls, data: T_Yaml) -> Self: # def from_dict(cls, data: T_Yaml) -> Self:
command = data.get("command") # command = data.get("command")
volumes = data.get("volumes") # volumes = data.get("volumes")
entry = data.get("entrypoint") # entry = data.get("entrypoint")
name = data.get("container_name") # name = data.get("container_name")
if name is None: # if name is None:
raise KeyError # raise KeyError
return cls( # return cls(
tuple(command) if command else None, # tuple(command) if command else None,
name, # name,
tuple(entry) if entry else None, # tuple(entry) if entry else None,
data.get("environment"), # data.get("environment"),
data["image"], # data["image"],
_get_labels(data), # _get_labels(data),
None, # None,
cls.get_nets(data), # cls.get_nets(data),
"unless-stopped", # "unless-stopped",
_get_sec_opts(data), # _get_sec_opts(data),
data.get("user"), # data.get("user"),
frozenset(volumes) if volumes else None, # frozenset(volumes) if volumes else None,
) # )
def is_valid(self) -> bool: # def is_valid(self) -> bool:
attrs = (self.container_name, self.restart, self.logging) # attrs = (self.container_name, self.restart, self.logging)
for attr in attrs: # for attr in attrs:
if attr is None: # if attr is None:
return False # return False
return True # return True
@staticmethod @staticmethod
@abstractmethod @abstractmethod
@@ -127,36 +126,3 @@ def _get_nets(
if nets is None: if nets is None:
return return
return frozenset(nets) return frozenset(nets)
def _get_sec_opts(
data: T_Compose,
) -> frozenset[str]:
sec_opts = frozenset(
"no-new-privileges:true",
)
sec = data.get("security_opt")
if not sec:
return sec_opts
return sec_opts.union(sec)
def _get_labels(
data: T_Compose,
) -> frozenset[str] | None:
org_name = get_replace_name("org_name")
url = get_replace_name("url")
traefik_labels = frozenset(
(
f"traefik.http.routers.{org_name}.rule=Host(`{url}`)",
f"traefik.http.routers.{org_name}.entrypoints=websecure",
f"traefik.docker.network={org_name}_proxy",
f"traefik.http.routers.{org_name}.tls.certresolver=le",
)
)
labels = data.get("labels")
if not labels:
return
if "traefik.enable=true" not in labels:
return frozenset(labels)
return traefik_labels.union(labels)