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
2026-01-23 15:15:49 -06:00
parent b04a997561
commit 0c5686b3a7
9 changed files with 120 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
src/docker_compose/domain/compose/compose.py
View File

@@ -14,11 +14,13 @@ from docker_compose.domain.compose.volume_files import VolumeFile
if TYPE_CHECKING:
from docker_compose.domain.paths.src import SrcPaths
class ComposeDict(TypedDict):
name:str
services:dict[str,ServiceWriteDict]
networks:dict[str,NetworkDictSub]
volumes:dict[str,Any]
name: str
services: dict[str, ServiceWriteDict]
networks: dict[str, NetworkDictSub]
volumes: dict[str, Any]
@final
class Compose(Slots):
@@ -44,15 +46,12 @@ class Compose(Slots):
for network in service.networks:
yield network.as_dict
@property
def as_dict(self) -> ComposeDict:
return {
'name':self.name,
"name": self.name,
"services": dict(ChainMap(*(s.as_dict for s in self.services))),
"networks": dict(ChainMap(
*(self.networks)
)),
"networks": dict(ChainMap(*(self.networks))),
"volumes": dict(ChainMap(*(vol.as_dict for vol in self.volumes))),
}

139
src/docker_compose/domain/compose/service/service.py
View File

@@ -21,8 +21,12 @@ from docker_compose.util import ReplaceStr
if TYPE_CHECKING:
from docker_compose.domain.compose.compose import Compose
class DependsOnDict(TypedDict):
condition: Literal['service_started', 'service_healthy', 'service_completed_successfully']
condition: Literal[
"service_started", "service_healthy", "service_completed_successfully"
]
class HealthCheck(TypedDict):
test: tuple[str, ...]
@@ -31,79 +35,95 @@ class HealthCheck(TypedDict):
retries: int | None
start_period: str | None
class ServiceReadKeys(StrEnum):
image='image'
restart='restart'
image = "image"
restart = "restart"
class ServiceReadDict(TypedDict):
image:str
restart:NotRequired[str]
image: str
restart: NotRequired[str]
user: NotRequired[str]
shm_size: NotRequired[str]
depends_on: NotRequired[str|dict[str, DependsOnDict]]
command: NotRequired[tuple[str,...]]
entrypoint: NotRequired[tuple[str,...]]
depends_on: NotRequired[str | dict[str, DependsOnDict]]
command: NotRequired[tuple[str, ...]]
entrypoint: NotRequired[tuple[str, ...]]
environment: NotRequired[dict[str, str]]
labels: NotRequired[tuple[str,...]]
logging: NotRequired[tuple[str,...]]
networks: NotRequired[tuple[str,...]]
security_opts: NotRequired[tuple[str,...]]
volumes: NotRequired[tuple[str,...]]
ports: NotRequired[tuple[str,...]]
labels: NotRequired[tuple[str, ...]]
logging: NotRequired[tuple[str, ...]]
networks: NotRequired[tuple[str, ...]]
security_opts: NotRequired[tuple[str, ...]]
volumes: NotRequired[tuple[str, ...]]
ports: NotRequired[tuple[str, ...]]
healthcheck: NotRequired[HealthCheck]
class ServiceWriteDict(TypedDict):
container_name:str
image:str
restart:str
user: str|None
shm_size: str|None
depends_on: str|dict[str, DependsOnDict]|None
command: tuple[str,...]
entrypoint: tuple[str,...]
container_name: str
image: str
restart: str
user: str | None
shm_size: str | None
depends_on: str | dict[str, DependsOnDict] | None
command: tuple[str, ...]
entrypoint: tuple[str, ...]
environment: dict[str, str]
labels: tuple[str,...]
logging: tuple[str,...]
networks: tuple[str,...]
security_opts: tuple[str,...]
volumes: tuple[str,...]
ports: tuple[str,...]
healthcheck: HealthCheck|None
labels: tuple[str, ...]
logging: tuple[str, ...]
networks: tuple[str, ...]
security_opts: tuple[str, ...]
volumes: tuple[str, ...]
ports: tuple[str, ...]
healthcheck: HealthCheck | None
@final
class Service(Slots):
_traefik_labels:tuple[str,...] = (
_traefik_labels: tuple[str, ...] = (
"traefik.enable=true",
f"traefik.http.routers.{DN.repl}.rule=Host(`{ReplaceStr.fmt('url')}`)",
f"traefik.http.routers.{DN.repl}.entrypoints=websecure",
f"traefik.docker.network={DN.repl}_proxy",
f"traefik.http.routers.{DN.repl}.tls.certresolver=le",
)
_sec_opts:tuple[str,...] = ("no-new-privileges:true",)
_sec_opts: tuple[str, ...] = ("no-new-privileges:true",)
def __init__(self, compose:Compose, path:Path) -> None:
self.compose :Final[Compose]= compose
self.service_name:Final[str] = path.stem
def __init__(self, compose: Compose, path: Path) -> None:
self.compose: Final[Compose] = compose
self.service_name: Final[str] = path.stem
data: Final[ServiceReadDict] = self.load(path)
self.container_name:Final[str] = f"{DN.repl}_{self.service_name}"
self.container_name: Final[str] = f"{DN.repl}_{self.service_name}"
self.image:Final[str] = data['image']
self.user:Final[str | None] = data.get('user')
self.shm_size:Final[str | None] = data.get('shm_size')
self.restart:Final[str] = data.get('restart',"unless-stopped")
self.depends_on:Final[str | dict[str, DependsOnDict] | None] = data.get('depends_on')
self.command:Final[tuple[str, ...]] = self.string_lists(data, 'command')
self.entrypoint:Final[tuple[str, ...]] = self.string_lists(data,'entrypoint')
self.environment:Final[dict[str, str]] = self.string_dict(data.get('environment',{}) )
self.labels_raw:Final[tuple[str, ...]] = self.string_lists(data,'labels')
self.logging:Final[tuple[str, ...]] =self.string_lists( data,'logging')
self.networks:Final[tuple[Network, ...]] = tuple(Network(self, s) for s in data.get('networks', ()) )
self.security_opt:Final[tuple[str, ...]] = self.string_lists(data,'security_opts')
self.volumes:Final[tuple[Volumes, ...]] = tuple(Volumes(self, s) for s in data.get('volumes', ()) )
self.ports:Final[tuple[Port, ...]] = tuple(Port(s) for s in data.get('ports', ()))
self.healthcheck:Final[HealthCheck | None] = data.get('healthcheck')
self.image: Final[str] = data["image"]
self.user: Final[str | None] = data.get("user")
self.shm_size: Final[str | None] = data.get("shm_size")
self.restart: Final[str] = data.get("restart", "unless-stopped")
self.depends_on: Final[str | dict[str, DependsOnDict] | None] = data.get(
"depends_on"
)
self.command: Final[tuple[str, ...]] = self.string_lists(data, "command")
self.entrypoint: Final[tuple[str, ...]] = self.string_lists(data, "entrypoint")
self.environment: Final[dict[str, str]] = self.string_dict(
data.get("environment", {})
)
self.labels_raw: Final[tuple[str, ...]] = self.string_lists(data, "labels")
self.logging: Final[tuple[str, ...]] = self.string_lists(data, "logging")
self.networks: Final[tuple[Network, ...]] = tuple(
Network(self, s) for s in data.get("networks", ())
)
self.security_opt: Final[tuple[str, ...]] = self.string_lists(
data, "security_opts"
)
self.volumes: Final[tuple[Volumes, ...]] = tuple(
Volumes(self, s) for s in data.get("volumes", ())
)
self.ports: Final[tuple[Port, ...]] = tuple(
Port(s) for s in data.get("ports", ())
)
self.healthcheck: Final[HealthCheck | None] = data.get("healthcheck")
@property
def as_dict(self) -> dict[str, ServiceWriteDict]:
@@ -112,10 +132,10 @@ class Service(Slots):
container_name=self.container_name,
image=self.image,
restart=self.restart,
user = self.user,
user=self.user,
shm_size=self.shm_size,
depends_on=self.depends_on,
command = self.command,
command=self.command,
entrypoint=self.entrypoint,
environment=self.environment,
labels=self.labels,
@@ -128,10 +148,9 @@ class Service(Slots):
)
}
def __iter__(self) -> Generator[ReplaceStr]:
yield FQDN
yield DN
yield DN
yield ReplaceStr(ReplaceStr.fmt("service"), self.service_name)
def __call__(self, data: str) -> str:
@@ -141,10 +160,10 @@ class Service(Slots):
def labels(self) -> tuple[str, ...]:
if "traefik.enable=true" not in self.labels_raw:
return self.labels_raw
return tuple(chain(self.labels_raw,self._traefik_labels))
return tuple(chain(self.labels_raw, self._traefik_labels))
def string_lists(self, data:ServiceReadDict, key:str) -> tuple[str, ...]:
return tuple(self.string_lists_sub(data.get(key,())))
def string_lists(self, data: ServiceReadDict, key: str) -> tuple[str, ...]:
return tuple(self.string_lists_sub(data.get(key, ())))
@staticmethod
def string_lists_sub(data: Iterable[str]) -> Iterator[str]:
@@ -152,11 +171,11 @@ class Service(Slots):
yield s.strip()
@staticmethod
def string_dict(data:dict[str,str])->dict[str,str]:
return {k.strip():v.strip() for k,v in data.items()}
def string_dict(data: dict[str, str]) -> dict[str, str]:
return {k.strip(): v.strip() for k, v in data.items()}
@staticmethod
def load(path: Path) -> ServiceReadDict:
with path.open("rt") as f:
data =yaml.safe_load(f) # pyright: ignore[reportAny]
data = yaml.safe_load(f) # pyright: ignore[reportAny]
return TypeAdapter(ServiceReadDict).validate_python(data)

7
src/docker_compose/domain/compose/service/volumes.py
View File

@@ -13,11 +13,12 @@ if TYPE_CHECKING:
@final
class Volumes(Slots):
sep = ":"
def __init__(self, service:Service, raw:str) -> None:
def __init__(self, service: Service, raw: str) -> None:
src, dest = (s.strip() for s in raw.split(self.sep))
self.service: Final[Service] = service
self._src: Final[str]=src
self.dest: Final[str]= dest
self._src: Final[str] = src
self.dest: Final[str] = dest
@property
def src(self) -> str:

6
src/docker_compose/domain/compose/volume_files.py
View File

@@ -7,8 +7,8 @@ from autoslot import Slots
@final
class VolumeFile(Slots):
def __init__(self, path:Path) -> None:
self.name:Final = path.stem
def __init__(self, path: Path) -> None:
self.name: Final = path.stem
self.data: Final = self.load(path)
@staticmethod
@@ -18,4 +18,4 @@ class VolumeFile(Slots):
@property
def as_dict(self) -> dict[str, dict[str, Any]]:
return {self.name:self.data}
return {self.name: self.data}

11
src/docker_compose/domain/env/env_data.py vendored
View File

@@ -10,13 +10,12 @@ from docker_compose.domain.env.env_row import EnvRow
if TYPE_CHECKING:
from docker_compose.domain.paths.src import SrcPaths
@final
class EnvData(Slots):
def __init__(self, src_paths:SrcPaths) -> None:
self.src_paths:Final = src_paths
self.data:Final = tuple(self.lines)
def __init__(self, src_paths: SrcPaths) -> None:
self.src_paths: Final = src_paths
self.data: Final = tuple(self.lines)
@property
def lines(self) -> Generator[EnvRow]:
@@ -29,5 +28,3 @@ class EnvData(Slots):
@property
def as_list(self):
return tuple(str(row) for row in self.data)

12
src/docker_compose/domain/paths/dest.py
View File

@@ -7,7 +7,6 @@ from autoslot import Slots
from docker_compose import APP_ROOT
from docker_compose.domain.paths import FILES
from docker_compose.domain.paths.org import OrgData
if TYPE_CHECKING:
from docker_compose.domain.paths.org import OrgData
@@ -15,9 +14,8 @@ if TYPE_CHECKING:
@final
class DestPath(Slots):
def __init__(self, org_data:OrgData) -> None:
self.org_data:Final[OrgData] = org_data
self.base_path:Final[Path] = APP_ROOT.joinpath(*self.org_data)
self.compose_path:Final[Path] = self.base_path.joinpath(FILES.COMPOSE)
self.env_path :Final[Path]= self.base_path.joinpath(FILES.ENV)
def __init__(self, org_data: OrgData) -> None:
self.org_data: Final[OrgData] = org_data
self.base_path: Final[Path] = APP_ROOT.joinpath(*self.org_data)
self.compose_path: Final[Path] = self.base_path.joinpath(FILES.COMPOSE)
self.env_path: Final[Path] = self.base_path.joinpath(FILES.ENV)

24
src/docker_compose/domain/paths/src.py
View File

@@ -13,18 +13,20 @@ from docker_compose.domain.paths.org import OrgData, Orgs
@final
class SrcPaths(Slots):
def __init__(self, path:Path) -> None:
self.path:Final=path
def __init__(self, path: Path) -> None:
self.path: Final = path
self.compose:Final= Compose(self)
self.cfg:Final[dict[Orgs,OrgData]]= {cast(Orgs,obj.org): obj for obj in OrgData.from_src_path(self)}
self.env:Final=EnvData(self)
self.compose_file :Final= self.path.joinpath(FILES.COMPOSE)
self.bind_vol_path:Final= self.path.joinpath(FILES.BIND_VOLS)
self.service_files:Final=tuple(self.get_yaml_files(FILES.SERVICES))
self.volume_files:Final= tuple(self.get_yaml_files(FILES.VOLUMES))
self.cfg_file:Final= self.path.joinpath(FILES.CFG)
self.env_file:Final= self.path.joinpath(FILES.ENV)
self.compose: Final = Compose(self)
self.cfg: Final[dict[Orgs, OrgData]] = {
cast(Orgs, obj.org): obj for obj in OrgData.from_src_path(self)
}
self.env: Final = EnvData(self)
self.compose_file: Final = self.path.joinpath(FILES.COMPOSE)
self.bind_vol_path: Final = self.path.joinpath(FILES.BIND_VOLS)
self.service_files: Final = tuple(self.get_yaml_files(FILES.SERVICES))
self.volume_files: Final = tuple(self.get_yaml_files(FILES.VOLUMES))
self.cfg_file: Final = self.path.joinpath(FILES.CFG)
self.env_file: Final = self.path.joinpath(FILES.ENV)
def get_yaml_files(self, folder: str) -> Iterator[Path]:
for service in self.path.joinpath(folder).iterdir():

4
src/docker_compose/domain/render/bind_vols.py
View File

@@ -14,8 +14,8 @@ if TYPE_CHECKING:
@final
class BindVols(Slots):
def __init__(self, render:Render) -> None:
self.render:Final = render
def __init__(self, render: Render) -> None:
self.render: Final = render
def __call__(self):
for path in self:

6
src/docker_compose/domain/render/render.py
View File

@@ -5,7 +5,6 @@ from typing import TYPE_CHECKING, Final, final, override
from autoslot import Slots
from docker_compose.domain.compose.compose import Compose
from docker_compose.domain.paths.org import OrgData
from docker_compose.domain.render.bind_vols import BindVols
if TYPE_CHECKING:
@@ -14,8 +13,8 @@ if TYPE_CHECKING:
@final
class Render(Slots):
def __init__(self, org_data:OrgData) -> None:
self.org_data:Final[OrgData] =org_data
def __init__(self, org_data: OrgData) -> None:
self.org_data: Final[OrgData] = org_data
self.bind_vols = BindVols(self)
@property
@@ -29,4 +28,3 @@ class Render(Slots):
def __call__(self):
with self.org_data.dest.compose_path.open("wt") as f:
_ = f.write(str(self))