This commit is contained in:
2026-01-23 15:00:14 -06:00
parent 13793f2d70
commit b04a997561
22 changed files with 424 additions and 596 deletions

View File

@@ -9,7 +9,11 @@
"editor.defaultFormatter": "charliermarsh.ruff"
},
"cSpell.words": [
"autoslot",
"ccamper",
"certresolver",
"funcs",
"STRYTEN",
"traefik",
"websecure"
]

View File

@@ -6,6 +6,7 @@ readme = "README.md"
authors = [{ name = "Christian Camper", email = "ccamper7@gmail.com" }]
requires-python = ">=3.13"
dependencies = [
"autoslot>=2025.11.1",
"basedpyright>=1.37.1",
"loguru>=0.7.3",
"pydantic>=2.12.5",
@@ -24,3 +25,11 @@ build-backend = "uv_build"
[tool.basedpyright]
reportExplicitAny = "none"
reportImportCycles = "none"
executionEnvironments = [
{ root = "src" }
]
exclude = [
".*",
"typings",
]

View File

@@ -28,6 +28,3 @@ TRAEFIK_PATH = TEMPLATE_ROOT.joinpath("traefik")
# SQLModel.metadata.create_all(ENGINE)
#
def fmt_replace_str(src: str) -> str:
return f"${{_{src.upper()}}}"

View File

@@ -1,70 +1,60 @@
from __future__ import annotations
from collections import ChainMap
from collections.abc import Generator, MutableMapping
from typing import TYPE_CHECKING, Any, final, override
from collections.abc import Iterator
from typing import TYPE_CHECKING, Any, Final, TypedDict, final, override
import yaml
from pydantic import Field, RootModel, computed_field, model_serializer
from pydantic.dataclasses import dataclass
from autoslot import Slots
from docker_compose.domain.compose.service.service import Service
from docker_compose.domain.compose.service.networks import NetworkDict, NetworkDictSub
from docker_compose.domain.compose.service.service import Service, ServiceWriteDict
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]
@final
@dataclass(slots=True)
class Compose:
src_paths: SrcPaths
name: str = Field(init=False)
services: tuple[Service, ...] = Field(init=False)
volumes: tuple[VolumeFile, ...] = Field(init=False)
def __post_init__(self):
self.name = self.src_paths.path.stem
self.services = tuple(self.service_files)
self.volumes = tuple(self.volume_files)
class Compose(Slots):
def __init__(self, src_paths: SrcPaths) -> None:
self.src_paths: Final[SrcPaths] = src_paths
self.name: Final[str] = self.src_paths.path.stem
self.services: Final[tuple[Service, ...]] = tuple(self.service_files)
self.volumes: Final[tuple[VolumeFile, ...]] = tuple(self.volume_files)
@property
def service_files(self):
def service_files(self) -> Iterator[Service]:
for path in self.src_paths.service_files:
yield Service.from_path(self, path)
yield Service(self, path)
@property
def volume_files(self):
def volume_files(self) -> Iterator[VolumeFile]:
for path in self.src_paths.volume_files:
yield VolumeFile.from_path(path)
yield VolumeFile(path)
@property
def networks_sub(self) -> Generator[dict[str, Any]]:
def networks(self) -> Iterator[NetworkDict]:
for service in self.services:
for network in service.networks:
yield network.as_dict(False)
yield network.as_dict
@computed_field
@property
def networks(self) -> MutableMapping[str, Any]:
return ChainMap(*self.networks_sub)
# @classmethod
# def from_path(cls, path: Path) -> Self:
# src_paths = SrcPaths(path)
# return cls(
# path.stem,
# tuple(map(Service.from_path, src_paths.service_files)),
# tuple(map(VolumeFile.from_path, src_paths.volume_files)),
# )
@property
def as_dict(self) -> dict[Any, Any]:
return RootModel[Compose](self).model_dump(exclude_none=True) # pyright: ignore[reportAny]
@model_serializer(mode="plain")
def dump(self) -> dict[str, Any]:
return {self.name: ChainMap(*(s.as_dict for s in self.services))}
def as_dict(self) -> ComposeDict:
return {
'name':self.name,
"services": dict(ChainMap(*(s.as_dict for s in self.services))),
"networks": dict(ChainMap(
*(self.networks)
)),
"volumes": dict(ChainMap(*(vol.as_dict for vol in self.volumes))),
}
@override
def __str__(self) -> str:

View File

@@ -1,15 +1,16 @@
from docker_compose import APP_ROOT, fmt_replace_str
from docker_compose import APP_ROOT
from docker_compose.domain.paths import ReStrings
from docker_compose.util import ReplaceStr
DN = ReplaceStr(
fmt_replace_str("dn"),
"_".join(fmt_replace_str(s) for s in ("org", "name")),
ReplaceStr.fmt("dn"),
"_".join(ReplaceStr.fmt(s) for s in (ReStrings.ORG, ReStrings.APP)),
)
FQDN = ReplaceStr(
fmt_replace_str("fqdn"),
"_".join(fmt_replace_str(s) for s in ("org", "name", "services")),
ReplaceStr.fmt("fqdn"),
"_".join(ReplaceStr.fmt(s) for s in (ReStrings.ORG, ReStrings.APP, ReStrings.SERVICE)),
)
DATA_DN = ReplaceStr(
fmt_replace_str("data"),
str(APP_ROOT.joinpath(*(fmt_replace_str(s) for s in ("org", "name")))),
ReplaceStr.fmt("data"),
str(APP_ROOT.joinpath(*(ReplaceStr.fmt(s) for s in (ReStrings.ORG, ReStrings.APP)))),
)

View File

@@ -1,42 +0,0 @@
from __future__ import annotations
from typing import TYPE_CHECKING, Any, cast, final
from pydantic import (
ConfigDict,
Field,
SerializerFunctionWrapHandler,
field_serializer,
model_serializer,
)
from pydantic.dataclasses import dataclass
if TYPE_CHECKING:
from docker_compose.domain.compose.service.service import Service
@final
@dataclass(slots=True, config=ConfigDict(str_strip_whitespace=True))
class Environment:
sep = "="
key: str
val: str
service: Service = Field(init=False)
@field_serializer("val", mode="plain")
def get_val(self, v: str) -> str:
return self.service(v)
# @field_validator("key", "val", mode="after")
# @classmethod
# def val_dump(cls, val: str) -> str:
# return val.strip()
@model_serializer(mode="wrap")
def model_serial(
self,
handler: SerializerFunctionWrapHandler,
# info: SerializationInfo,
) -> str:
data = cast(dict[str, Any], handler(self))
return f"{data['key']}{self.sep}{data['val']}"

View File

@@ -1,26 +0,0 @@
from typing import final
from pydantic import ConfigDict, field_validator
from pydantic.dataclasses import dataclass
@final
@dataclass(slots=True, config=ConfigDict(str_strip_whitespace=True))
class HealthCheck:
test: tuple[str, ...]
interval: str | None
timeout: str | None
retries: int | None
start_period: str | None
@field_validator("test", mode="after")
@classmethod
def test_validator(cls, v: tuple[str, ...]) -> tuple[str, ...]:
return tuple(s.strip() for s in v)
# @field_validator("interval", "timeout", "start_period", mode="after")
# @classmethod
# def string_validator(cls, v: str | None) -> str | None:
# if not v:
# return
# return v.strip()

View File

@@ -1,52 +1,30 @@
from __future__ import annotations
from typing import TYPE_CHECKING, Any, cast, final
from typing import TYPE_CHECKING, Final, TypedDict, final
from pydantic import (
ConfigDict,
Field,
RootModel,
SerializationInfo,
SerializerFunctionWrapHandler,
model_serializer,
)
from pydantic.dataclasses import dataclass
from autoslot import Slots
from docker_compose.domain.compose.service import DN
from docker_compose.domain.compose.service.service import Service
if TYPE_CHECKING:
from docker_compose.domain.compose.service.service import Service
class NetworkDictSub(TypedDict):
name: str
external: bool
type NetworkDict = dict[str, NetworkDictSub]
@final
@dataclass(slots=True, config=ConfigDict(str_strip_whitespace=True))
class Network:
val: str
service: Service = Field(init=False)
class Network(Slots):
def __init__(self, service:Service, val:str) -> None:
self.service: Final[Service] = service
self.val: Final[str] = val.strip()
self.name:Final[str] = f"{DN.repl}_{self.val}"
self.external :Final[bool]= "proxy" in self.val
@property
def name(self):
return f"{DN.repl}_{self.val}"
@property
def external(self):
return "proxy" in self.val
@model_serializer(mode="wrap")
def serialize_model(
self,
handler: SerializerFunctionWrapHandler,
info: SerializationInfo,
) -> str | dict[str, Any]:
context = cast(dict[str, Any] | None, info.context)
data = cast(dict[str, Any], handler(self))
if context:
context = cast(bool, context.get("full"))
if context is None:
return cast(str, data["val"])
if not data["external"] or context:
data.pop("external", None)
return {data.pop("val"): data}
def as_dict(self, context: bool = False) -> dict[str, Any]:
return RootModel[Network](self).model_dump(context={"full": context}) # pyright: ignore[reportAny]
def as_dict(self) -> NetworkDict:
return {self.name: NetworkDictSub(name=self.name, external=self.external)}

View File

@@ -1,21 +1,16 @@
from typing import Self, final
from typing import Final, final, override
from pydantic import model_serializer
from pydantic.dataclasses import dataclass
from autoslot import Slots
@final
@dataclass(slots=True)
class Port:
class Port(Slots):
sep = ":"
def __init__(self, raw:str) -> None:
src, dest = (int(s) for s in raw.split(self.sep))
self.src: Final[int] = src
self.dest: Final[int]=dest
src: int
dest: int
@classmethod
def from_string(cls, string: str) -> Self:
return cls(*(int(s) for s in string.split(cls.sep)))
@model_serializer(mode="plain")
def serialize_model(self) -> str:
@override
def __str__(self) -> str:
return f"{self.src}{self.sep}{self.dest}"

View File

@@ -1,26 +1,18 @@
from __future__ import annotations
from collections.abc import Generator, Iterable
from dataclasses import InitVar
from collections.abc import Generator, Iterable, Iterator
from enum import StrEnum
from functools import reduce
from itertools import chain
from pathlib import Path
from typing import TYPE_CHECKING, Any, Self, cast, final
from typing import TYPE_CHECKING, Final, Literal, NotRequired, TypedDict, final
import yaml
from pydantic import (
Field,
RootModel,
SerializerFunctionWrapHandler,
computed_field,
field_validator,
model_serializer,
)
from pydantic.dataclasses import dataclass
from autoslot import Slots
from pydantic import TypeAdapter
from docker_compose import fmt_replace_str
from docker_compose.domain.compose.compose import Compose
from docker_compose.domain.compose.service import DN, FQDN
from docker_compose.domain.compose.service.env import Environment
from docker_compose.domain.compose.service.health_check import HealthCheck
from docker_compose.domain.compose.service.networks import Network
from docker_compose.domain.compose.service.port import Port
from docker_compose.domain.compose.service.volumes import Volumes
@@ -29,119 +21,142 @@ 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']
class HealthCheck(TypedDict):
test: tuple[str, ...]
interval: str | None
timeout: str | None
retries: int | None
start_period: str | None
class ServiceReadKeys(StrEnum):
image='image'
restart='restart'
class ServiceReadDict(TypedDict):
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,...]]
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,...]]
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,...]
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
@final
@dataclass(slots=True)
class Service:
_traefik_labels = (
class Service(Slots):
_traefik_labels:tuple[str,...] = (
"traefik.enable=true",
f"traefik.http.routers.{DN.repl}.rule=Host(`{fmt_replace_str('url')}`)",
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 = ("no-new-privileges:true",)
_sec_opts:tuple[str,...] = ("no-new-privileges:true",)
compose: Compose
path: InitVar[Path]
def __init__(self, compose:Compose, path:Path) -> None:
self.compose :Final[Compose]= compose
self.service_name:Final[str] = path.stem
image: str
service_name: str = Field(init=False, exclude=True)
user: str | None = Field(default=None)
shm_size: str | None = Field(default=None)
restart: str = Field(default="unless-stopped")
depends_on: dict[str, dict[str, str]] = Field(
default_factory=dict,
exclude_if=lambda v: not v, # pyright: ignore[reportAny]
)
data: Final[ServiceReadDict] = self.load(path)
command: tuple[str, ...] = Field(
default=(),
exclude_if=lambda v: not v, # pyright: ignore[reportAny]
)
entrypoint: tuple[str, ...] = Field(
default=(),
exclude_if=lambda v: not v, # pyright: ignore[reportAny]
)
environment: tuple[Environment, ...] = Field(
default=(),
exclude_if=lambda v: not v, # pyright: ignore[reportAny]
)
labels_raw: tuple[str, ...] = Field(
default=(),
exclude=True,
)
logging: tuple[str, ...] = Field(
default=(),
exclude_if=lambda v: not v, # pyright: ignore[reportAny]
)
networks: tuple[Network, ...] = Field(
default=(),
exclude_if=lambda v: not v, # pyright: ignore[reportAny]
)
security_opt: tuple[str, ...] = Field(
default=(),
exclude_if=lambda v: not v, # pyright: ignore[reportAny]
)
volumes: tuple[Volumes, ...] = Field(
default=(),
exclude_if=lambda v: not v, # pyright: ignore[reportAny]
)
ports: tuple[Port, ...] = Field(
default=(),
exclude_if=lambda v: not v, # pyright: ignore[reportAny]
)
healthcheck: HealthCheck | None = Field(
default=None,
exclude_if=lambda v: not v, # pyright: ignore[reportAny]
)
self.container_name:Final[str] = f"{DN.repl}_{self.service_name}"
def __post_init__(self, path: Path) -> None:
self.service_name = path.stem
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')
@computed_field
@property
def container_name(self):
return f"{DN.repl}_{self.service_name}"
def as_dict(self) -> dict[str, ServiceWriteDict]:
return {
self.container_name: ServiceWriteDict(
container_name=self.container_name,
image=self.image,
restart=self.restart,
user = self.user,
shm_size=self.shm_size,
depends_on=self.depends_on,
command = self.command,
entrypoint=self.entrypoint,
environment=self.environment,
labels=self.labels,
logging=self.logging,
networks=tuple(network.val for network in self.networks),
security_opts=self.security_opt,
volumes=tuple(str(vol) for vol in self.volumes),
ports=tuple(str(port) for port in self.ports),
healthcheck=self.healthcheck,
)
}
def __iter__(self) -> Generator[ReplaceStr]:
yield FQDN
yield DN
yield ReplaceStr("application", self.service_name)
yield ReplaceStr(ReplaceStr.fmt("service"), self.service_name)
def __call__(self, data: str) -> str:
return reduce(lambda s, f: f(s), self, data)
@computed_field
@property
def labels(self):
def labels(self) -> tuple[str, ...]:
if "traefik.enable=true" not in self.labels_raw:
return self.labels_raw
return self.labels_raw + self._traefik_labels
return tuple(chain(self.labels_raw,self._traefik_labels))
@field_validator(
"command",
"entrypoint",
"labels_raw",
"logging",
"security_opt",
mode="after",
)
@classmethod
def string_lists(cls, data: Iterable[str]) -> tuple[str, ...]:
return tuple(s.strip() for s in data)
def string_lists(self, data:ServiceReadDict, key:str) -> tuple[str, ...]:
return tuple(self.string_lists_sub(data.get(key,())))
@classmethod
def from_path(cls, compose: Compose, path: Path) -> Self:
@staticmethod
def string_lists_sub(data: Iterable[str]) -> Iterator[str]:
for s in data:
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()}
@staticmethod
def load(path: Path) -> ServiceReadDict:
with path.open("rt") as f:
data = cast(dict[str, Any], yaml.safe_load(f))
return cls(compose, path, **data) # pyright: ignore[reportAny]
@model_serializer(mode="wrap")
def dump(self, handler: SerializerFunctionWrapHandler) -> dict[str, Any]:
data = cast(dict[str, Any], handler(self))
return {data.pop("container_name"): data}
@property
def as_dict(self) -> dict[str, Any]:
return RootModel[Service](self).model_dump(exclude_none=True) # pyright: ignore[reportAny]
data =yaml.safe_load(f) # pyright: ignore[reportAny]
return TypeAdapter(ServiceReadDict).validate_python(data)

View File

@@ -1,32 +1,28 @@
from __future__ import annotations
from typing import TYPE_CHECKING, Self, final
from typing import TYPE_CHECKING, Final, final, override
from pydantic import ConfigDict, model_serializer
from pydantic.dataclasses import dataclass
from autoslot import Slots
from docker_compose.domain.compose.service.service import Service
if TYPE_CHECKING:
from docker_compose.domain.compose.service.service import Service
@final
@dataclass(slots=True, config=ConfigDict(str_strip_whitespace=True))
class Volumes:
class Volumes(Slots):
sep = ":"
service: Service
_src: str
dest: str
@classmethod
def from_str(cls, service: Service, src: str) -> Self:
return cls(service, *src.split(cls.sep, 2))
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
@property
def src(self) -> str:
return self.service(self._src)
@model_serializer(mode="plain")
def serialize_model(self) -> str:
@override
def __str__(self) -> str:
return f"{self.src}{self.sep}{self.dest}"

View File

@@ -1,31 +1,21 @@
from dataclasses import InitVar
from pathlib import Path
from typing import Any, Self, cast, final
from typing import Any, Final, cast, final
import yaml
from pydantic import ConfigDict, Field
from pydantic.dataclasses import dataclass
from autoslot import Slots
@final
@dataclass(slots=True, config=ConfigDict(str_strip_whitespace=True))
class VolumeFile:
path: InitVar[Path]
name: str = Field(init=False)
class VolumeFile(Slots):
def __init__(self, path:Path) -> None:
self.name:Final = path.stem
self.data: Final = self.load(path)
data: dict[str, Any]
def __post_init__(self, path: Path) -> None:
self.name = path.stem
@classmethod
def from_path(cls, path: Path) -> Self:
@staticmethod
def load(path: Path) -> dict[str, Any]:
with path.open("rt") as f:
data = cast(dict[str, Any], yaml.safe_load(f))
return cls(path, data)
return cast(dict[str, Any], yaml.safe_load(f))
#
# @field_validator("name", mode="after")
# @classmethod
# def name_validate(cls, s: str) -> str:
# return s.strip()
@property
def as_dict(self) -> dict[str, dict[str, Any]]:
return {self.name:self.data}

View File

@@ -1,26 +1,22 @@
from __future__ import annotations
from collections.abc import Generator
from typing import TYPE_CHECKING, Any, cast, final
from typing import TYPE_CHECKING, Final, final
from pydantic import Field, model_serializer
from pydantic.dataclasses import dataclass
from pydantic_core.core_schema import SerializerFunctionWrapHandler
from autoslot import Slots
from docker_compose.domain.env.env_row import EnvRow
if TYPE_CHECKING:
from docker_compose.domain.paths.src import SrcPaths
@final
@dataclass(slots=True)
class EnvData:
src_paths: SrcPaths
data: tuple[EnvRow, ...] = Field(init=False)
class EnvData(Slots):
def __init__(self, src_paths:SrcPaths) -> None:
self.src_paths:Final = src_paths
self.data:Final = tuple(self.lines)
def __post_init__(self):
self.data = tuple(self.lines)
@property
def lines(self) -> Generator[EnvRow]:
@@ -28,8 +24,10 @@ class EnvData:
for line in f:
if line.startswith("#"):
continue
yield EnvRow.from_str(self, line)
yield EnvRow(self, line)
@property
def as_list(self):
return tuple(str(row) for row in self.data)
@model_serializer(mode="wrap")
def serialize_model(self, handler: SerializerFunctionWrapHandler) -> list[str]:
return cast(dict[str, Any], handler(self))["data"] # pyright: ignore[reportAny]

View File

@@ -1,40 +1,32 @@
from __future__ import annotations
import secrets
from typing import Self, final
from typing import TYPE_CHECKING, Final, final, override
from pydantic import ConfigDict, computed_field, field_validator, model_serializer
from pydantic.dataclasses import dataclass
from autoslot import Slots
if TYPE_CHECKING:
from docker_compose.domain.env.env_data import EnvData
# if TYPE_CHECKING:
# from docker_compose.env.env_data import EnvData
@final
@dataclass(slots=True, frozen=True, config=ConfigDict(str_strip_whitespace=True))
class EnvRow:
parent: EnvData
key: str
_val: str
class EnvRow(Slots):
def __init__(self, parent:EnvData, raw:str) -> None:
key, val = (s.strip() for s in raw.split("="))
@classmethod
def from_str(cls, parent: EnvData, raw: str) -> Self:
return cls(parent, *raw.split("="))
@field_validator("key", mode="after")
@classmethod
def strip_string(cls, s: str) -> str:
if s.startswith("#"):
self.parent:Final = parent
self.key:Final= key.strip()
self._val:Final = val.strip()
if self.key.startswith("#"):
raise ValueError
return s
@model_serializer(mode="plain")
def model_serializer(self) -> str:
return f"{self.key}={self.val}"
@computed_field
@property
def val(self) -> str:
return self._val.replace("{_PSWD}", secrets.token_urlsafe(12))
@override
def __str__(self) -> str:
return f"{self.key}={self.val}"

View File

@@ -0,0 +1,20 @@
from enum import StrEnum
class ReStrings(StrEnum):
APP='name'
ORG='org'
SERVICE='service'
URL='url'
class YAML_EXTS(StrEnum):
YML= '.yml'
YAML= '.yaml'
class FILES(StrEnum):
COMPOSE= f"docker-compose{YAML_EXTS.YML}"
BIND_VOLS= f"bind_volumes{YAML_EXTS.YML}"
SERVICES= 'services'
VOLUMES= 'volumes'
CFG= f"cfg{YAML_EXTS.YML}"
ENV= '.env'

View File

@@ -1,33 +1,23 @@
from __future__ import annotations
from pathlib import Path
from typing import TYPE_CHECKING, final
from typing import TYPE_CHECKING, Final, final
from pydantic.dataclasses import dataclass
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
@final
@dataclass
class DestPath:
org_data: OrgData
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)
# @classmethod
# def from_path(cls, path: Path) -> Iterator[Self]:
# return map(cls, OrgData.from_path(path))
@property
def base_path(self):
return APP_ROOT.joinpath(*self.org_data)
@property
def compose_path(self) -> Path:
return self.base_path.joinpath("docker-compose.yml")
@property
def env_path(self) -> Path:
return self.base_path.joinpath(".env")

View File

@@ -3,13 +3,13 @@ from __future__ import annotations
from collections.abc import Generator, Iterator
from enum import StrEnum
from functools import reduce
from typing import TYPE_CHECKING, Self, cast, final
from typing import TYPE_CHECKING, Final, NotRequired, Self, TypedDict, final
import yaml
from pydantic import ConfigDict, Field, field_validator
from pydantic.dataclasses import dataclass
from autoslot import Slots
from pydantic import ConfigDict, TypeAdapter
from docker_compose import fmt_replace_str
from docker_compose.domain.paths import ReStrings
from docker_compose.domain.paths.dest import DestPath
from docker_compose.domain.render.render import Render
from docker_compose.util import ReplaceStr
@@ -17,30 +17,35 @@ from docker_compose.util import ReplaceStr
if TYPE_CHECKING:
from docker_compose.domain.paths.src import SrcPaths
class Orgs(StrEnum):
PERSONAL = "personal"
STRYTEN = "stryten"
C4 = "c4"
@final
class OrgSub(TypedDict):
__pydantic_config__ = ConfigDict(extra='forbid') # pyright: ignore[reportGeneralTypeIssues]
url: NotRequired[str]
type OrgDict = dict[Orgs, OrgSub]
# type OrgDict = dict[Orgs, OrgSub]
@final
@dataclass(
slots=True,
order=True,
config=ConfigDict(use_enum_values=True, str_strip_whitespace=True),
)
class OrgData:
src_paths: SrcPaths
app: str
org: Orgs
url: str | None
dest: DestPath = Field(init=False)
render: Render = Field(init=False)
class OrgData(Slots):
def __init__(self, src_paths:SrcPaths, app:str, org:str, url: str|None) -> None:
if url:
url = url.strip()
self.src_paths: Final = src_paths
self.app: Final= app.strip()
self.org: Final=org.strip()
self.url: Final = url if url else None
self.dest: Final = DestPath(self)
self.render: Final=Render(self)
def __post_init__(self):
self.dest = DestPath(self)
self.render = Render(self)
def __call__(self, string: str) -> str:
return reduce(lambda s, f: f(s), self.replace_funcs, string)
@@ -51,32 +56,21 @@ class OrgData:
@property
def _replace_args(self) -> Generator[tuple[str, str]]:
yield "app", self.app
yield ReStrings.APP, self.app
yield "org", self.org
yield "url", ".".join((self.url, "ccamper7", "net")) if self.url else ""
@property
def replace_funcs(self) -> Generator[ReplaceStr]:
for s, r in self._replace_args:
yield ReplaceStr(fmt_replace_str(s), r)
@field_validator("app", "org", mode="after")
@classmethod
def strip(cls, v: str) -> str:
return v.strip()
@field_validator("url", mode="before")
@classmethod
def strip_url(cls, v: str) -> str:
if not v:
return v
return v.strip()
yield ReplaceStr(ReplaceStr.fmt(s), r)
@classmethod
def from_src_path(cls, path: SrcPaths) -> Iterator[Self]:
# log_cls(cls, path=str(path))
validate = TypeAdapter[OrgDict](OrgDict).validate_python
with path.cfg_file.open("rt") as f:
data = cast(dict[str, dict[str, str]], yaml.safe_load(f))
data = yaml.safe_load(f) # pyright: ignore[reportAny]
app = path.cfg_file.stem
for org, _dict in data.items():
yield cls(app=app, org=org, **_dict) # pyright: ignore[reportArgumentType]
for org, _dict in validate(data).items():
yield cls(path, app, org, _dict.get('url'))

View File

@@ -1,58 +1,34 @@
from collections.abc import Iterator
from pathlib import Path
from typing import Self, final
from typing import Final, Self, cast, final
from pydantic import Field
from pydantic.dataclasses import dataclass
from autoslot import Slots
from docker_compose import TEMPLATE_ROOT
from docker_compose.domain.compose.compose import Compose
from docker_compose.domain.env.env_data import EnvData
from docker_compose.domain.paths import FILES, YAML_EXTS
from docker_compose.domain.paths.org import OrgData, Orgs
@final
@dataclass(slots=True)
class SrcPaths:
YAML_EXTS = frozenset((".yml", ".yaml"))
class SrcPaths(Slots):
def __init__(self, path:Path) -> None:
self.path:Final=path
path: Path
compose: Compose = Field(init=False)
cfg: dict[Orgs, OrgData] = Field(init=False)
env: EnvData = Field(init=False)
def __post_init__(self):
self.compose = Compose(self)
self.cfg = {obj.org: obj for obj in OrgData.from_src_path(self)}
self.env = EnvData(self)
@property
def compose_file(self):
return self.path.joinpath("docker-compose.yml")
@property
def bind_vol_path(self):
return self.path.joinpath("bind_volumes.yml")
@property
def service_files(self) -> Iterator[Path]:
yield from self.get_yaml_files("services")
@property
def volume_files(self) -> Iterator[Path]:
yield from self.get_yaml_files("volumes")
@property
def cfg_file(self):
return self.path.joinpath("cfg.yml")
@property
def env_file(self) -> Path:
return self.path.joinpath(".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():
if service.suffix not in self.YAML_EXTS:
if service.suffix not in YAML_EXTS:
continue
yield service

View File

@@ -2,9 +2,9 @@ from __future__ import annotations
from collections.abc import Iterator
from pathlib import Path
from typing import TYPE_CHECKING, final
from typing import TYPE_CHECKING, Final, final
from pydantic.dataclasses import dataclass
from autoslot import Slots
from docker_compose import ROOT
@@ -13,13 +13,11 @@ if TYPE_CHECKING:
@final
@dataclass(frozen=True, slots=True)
class BindVols:
# data_rep = Replace("data", str(DATA_ROOT))
render: Render
class BindVols(Slots):
def __init__(self, render:Render) -> None:
self.render:Final = render
def __call__(self):
# def mk_bind_vols(self) -> None:
for path in self:
path.mkdir(parents=True, exist_ok=True)

View File

@@ -1,11 +1,11 @@
from __future__ import annotations
from typing import TYPE_CHECKING, final, override
from typing import TYPE_CHECKING, Final, final, override
from pydantic import Field
from pydantic.dataclasses import dataclass
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:
@@ -13,24 +13,11 @@ if TYPE_CHECKING:
@final
@dataclass(slots=True)
class Render:
org_data: OrgData
bind_vols: BindVols = Field(init=False)
def __post_init__(self):
class Render(Slots):
def __init__(self, org_data:OrgData) -> None:
self.org_data:Final[OrgData] =org_data
self.bind_vols = BindVols(self)
# template: Compose = Field(init=False)
# org_data: dict[str, OrgData] = Field(init=False)
# def __post_init__(self, path: Path) -> None:
# self.src_paths = SrcPaths(path)
# self.template = Compose.from_path(self.src_paths.compose_file)
# self.org_data = {
# obj.org: obj for obj in OrgData.from_src_path(self.src_paths.cfg_file)
# }
@property
def template(self) -> Compose:
return self.org_data.src_paths.compose
@@ -43,105 +30,3 @@ class Render:
with self.org_data.dest.compose_path.open("wt") as f:
_ = f.write(str(self))
# @property
# def proxy_nets(self) -> Iterator[str]:
# for net in self.template.compose_data.networks:
# if not net.external:
# continue
# yield self.render(net.full_name)
#
# @final
# @dataclass(frozen=True, slots=True)
# class RenderByOrg:
# template: Template
# renders: dict[str, Render]
#
# def __iter__(self) -> Iterator[Render]:
# yield from self.renders.values()
# # yield render
#
# def __call__(self) -> None:
# self.template()
# for render in self:
# render()
# self.write_bind_vol_data()
#
# def write_bind_vol_data(self):
# write_yaml(self.vols, self.template.dest_path.bind_vol_path)
#
# def __getitem__(self, key: str) -> Render:
# return self.renders[key]
#
# def __bool__(self) -> bool:
# return bool(self.renders)
#
# @staticmethod
# def from_path_sub(template: Template, path: Path) -> Iterator[tuple[str, Render]]:
# for org in OrgData.from_src_path(path):
# yield org.org.dest, Render(template, org)
#
# @classmethod
# def from_path(cls, path: Path) -> Self:
# template = Template.from_src_path(path)
# return cls(
# template,
# dict(cls.from_path_sub(template, path)),
# )
#
# @property
# def app(self):
# return self.template.compose_data.name
#
# @property
# def vols(self) -> Iterator[str]:
# for render in self:
# for path in render.bind_vols:
# yield str(path)
#
# @property
# def proxy_nets(self) -> Iterator[str]:
# for render in self:
# yield from render.proxy_nets
#
#
# @final
# @dataclass(frozen=True, slots=True)
# class RenderByApp:
# renders: dict[str, RenderByOrg]
#
# def __iter__(self) -> Iterator[RenderByOrg]:
# yield from self.renders.values()
#
# def __call__(self) -> None:
# for obj in self:
# obj()
#
# @staticmethod
# def _get_folders(path_: Path) -> Iterator[Path]:
# for path in path_.iterdir():
# if not path.is_dir():
# continue
# if path.stem == "traefik":
# continue
# yield path
#
# @classmethod
# def _from_path_sub(cls, path_: Path) -> Iterator[tuple[str, RenderByOrg]]:
# for path in cls._get_folders(path_):
# by_org = RenderByOrg.from_path(path)
# yield by_org.app, by_org
#
# @classmethod
# def from_path(cls, path: Path) -> Self:
# return cls(dict(cls._from_path_sub(path)))
#
# @classmethod
# def load_all(cls) -> Self:
# return cls.from_path(TEMPLATE_ROOT)
#
# @property
# def proxy_nets(self) -> Iterator[str]:
# for render in self:
# yield from render.proxy_nets

View File

@@ -1,13 +1,15 @@
import re
from functools import partial, reduce
from typing import Any, final, override
from collections.abc import Iterator, Sequence, Set
from dataclasses import dataclass
from re import Pattern
from typing import Any, Final, final, override
import yaml
from pydantic.dataclasses import dataclass
from autoslot import Slots
@final
@dataclass
@dataclass(frozen=True, slots=True)
class ReplaceStr:
src: str
repl: str
@@ -15,24 +17,79 @@ class ReplaceStr:
def __call__(self, s: str) -> str:
return s.replace(self.src, self.repl)
@staticmethod
def fmt(src: str) -> str:
return f"${{_{src.upper()}}}"
class DictCleanup(Slots):
def __init__(self, data: Any) -> None: # pyright: ignore[reportAny]
self.data: Final[dict[Any, Any] | Any] = (
data.copy() if isinstance(data, dict) else data
)
def __call__(self) -> Any: # pyright: ignore[reportAny]
return self.base(self.data) # pyright: ignore[reportAny]
def base(self, data: Any) -> Any: # pyright: ignore[reportAny]
if isinstance(data, tuple | list):
return tuple(self.list_prep(data)) # pyright: ignore[reportUnknownArgumentType]
if isinstance(data, Set | Iterator):
return tuple(sorted(self.list_prep(data)))
if isinstance(data, dict):
return self.dict_prep(data) # pyright: ignore[reportUnknownArgumentType]
return data # pyright: ignore[reportAny]
# raise TypeError
def list_prep(
self, data: Set[Any] | Sequence[Any] | Iterator[Any]
) -> Iterator[Any]:
for val in data: # pyright: ignore[reportAny]
if not val and not isinstance(val, bool):
continue
yield self.base(data)
# if isinstance(data, (Set,Sequence,Iterator,dict)):
# if isinstance(val, tuple|list):
# yield tuple(self.list_prep(val))
# if isinstance(val, Set|Iterator):
# yield tuple(sorted(self.list_prep(val)))
# if isinstance(val, dict):
# yield dict(self.dict_prep(val))
def dict_prep(self, data: dict[Any, Any]) -> dict[Any, Any]:
for k in tuple(data.keys()): # pyright: ignore[reportAny]
data[k] = self.base(data[k])
if not data[k] and not isinstance(data[k], bool):
del data[k]
return data
@final
class YamlUtil:
indent = partial(re.compile(r"(^\s?-)", re.MULTILINE).sub, r" \g<1>")
port = partial(re.compile(r"(\W*?)(\d+:\d+)", re.MULTILINE).sub, r'\g<1>"\g<2>"')
class YamlUtil(Slots):
indent: Final[tuple[Pattern[str], str]] = (
re.compile(r"(^\s?-)", re.MULTILINE),
r" \g<1>",
)
port: Final[tuple[Pattern[str], str]] = (
re.compile(r"(\W*?)(\d+:\d+)", re.MULTILINE),
r'\g<1>"\g<2>"',
)
def __init__(self, data: dict[Any, Any]) -> None:
self.data: Final = DictCleanup(data)() # pyright: ignore[reportAny]
class VerboseSafeDumper(yaml.SafeDumper):
@override
def ignore_aliases(self, data: object) -> bool:
return True
def __call__(self, data: dict[Any, Any]) -> str:
return reduce(
lambda s, f: f(s),
self,
yaml.dump(data, Dumper=self.VerboseSafeDumper),
)
def __call__(self) -> str:
data = yaml.dump(self.data, Dumper=self.VerboseSafeDumper) # pyright: ignore[reportAny]
for regex, repl in self:
data = regex.sub(repl, data)
return data
def __iter__(self):
def __iter__(self) -> Iterator[tuple[Pattern[str], str]]:
yield self.indent
yield self.port

11
uv.lock generated
View File

@@ -11,6 +11,15 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/78/b6/6307fbef88d9b5ee7421e68d78a9f162e0da4900bc5f5793f6d3d0e34fb8/annotated_types-0.7.0-py3-none-any.whl", hash = "sha256:1f02e8b43a8fbbc3f3e0d4f0f4bfc8131bcb4eebe8849b8e5c773f3a1c582a53", size = 13643, upload-time = "2024-05-20T21:33:24.1Z" },
]
[[package]]
name = "autoslot"
version = "2025.11.1"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/3c/c8/818ff7ed59421f965cc2284ad20686da17e917d0c6d687c2d1b045555527/autoslot-2025.11.1.tar.gz", hash = "sha256:6ccaf4aa6db21e8a4b04a97338fb9efd03c4544cf1c9ac5d0fb12d70a920cb08", size = 11493, upload-time = "2025-11-27T13:32:19.088Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/bd/f3/376b4977beeafca00bd9b0cd84e1277c093fe3aac65507054fe025be55c3/autoslot-2025.11.1-py2.py3-none-any.whl", hash = "sha256:fa642b58b082d0021177fcef11dd0634a3fa3e8f078828cade52bc9a44c337f5", size = 8141, upload-time = "2025-11-27T13:32:17.01Z" },
]
[[package]]
name = "basedpyright"
version = "1.37.1"
@@ -37,6 +46,7 @@ name = "docker-compose"
version = "0.1.0"
source = { editable = "." }
dependencies = [
{ name = "autoslot" },
{ name = "basedpyright" },
{ name = "loguru" },
{ name = "pydantic" },
@@ -47,6 +57,7 @@ dependencies = [
[package.metadata]
requires-dist = [
{ name = "autoslot", specifier = ">=2025.11.1" },
{ name = "basedpyright", specifier = ">=1.37.1" },
{ name = "loguru", specifier = ">=0.7.3" },
{ name = "pydantic", specifier = ">=2.12.5" },