Compare commits

...

4 Commits

Author SHA256 Message Date
df9d786b70 sync 2026-01-23 15:17:32 -06:00
0c5686b3a7 sync 2026-01-23 15:15:49 -06:00
b04a997561 sync 2026-01-23 15:00:14 -06:00
13793f2d70 sync 2026-01-23 14:59:58 -06:00
24 changed files with 439 additions and 600 deletions

4
.gitignore vendored
View File

@@ -6,8 +6,6 @@ dist/
wheels/ wheels/
*.egg-info *.egg-info
# Virtual environments
.venv
# Logs # Logs
logs/ logs/
typings/

1
.idea/.gitignore generated vendored Normal file
View File

@@ -0,0 +1 @@
workspace.xml

View File

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

View File

@@ -6,6 +6,7 @@ readme = "README.md"
authors = [{ name = "Christian Camper", email = "ccamper7@gmail.com" }] authors = [{ name = "Christian Camper", email = "ccamper7@gmail.com" }]
requires-python = ">=3.13" requires-python = ">=3.13"
dependencies = [ dependencies = [
"autoslot>=2025.11.1",
"basedpyright>=1.37.1", "basedpyright>=1.37.1",
"loguru>=0.7.3", "loguru>=0.7.3",
"pydantic>=2.12.5", "pydantic>=2.12.5",
@@ -24,3 +25,11 @@ build-backend = "uv_build"
[tool.basedpyright] [tool.basedpyright]
reportExplicitAny = "none" reportExplicitAny = "none"
reportImportCycles = "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) # SQLModel.metadata.create_all(ENGINE)
# #
def fmt_replace_str(src: str) -> str:
return f"${{_{src.upper()}}}"

View File

@@ -1,70 +1,59 @@
from __future__ import annotations from __future__ import annotations
from collections import ChainMap from collections import ChainMap
from collections.abc import Generator, MutableMapping from collections.abc import Iterator
from typing import TYPE_CHECKING, Any, final, override from typing import TYPE_CHECKING, Any, Final, TypedDict, final, override
import yaml import yaml
from pydantic import Field, RootModel, computed_field, model_serializer from autoslot import Slots
from pydantic.dataclasses import dataclass
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 from docker_compose.domain.compose.volume_files import VolumeFile
if TYPE_CHECKING: if TYPE_CHECKING:
from docker_compose.domain.paths.src import SrcPaths 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 @final
@dataclass(slots=True) class Compose(Slots):
class Compose: def __init__(self, src_paths: SrcPaths) -> None:
src_paths: SrcPaths self.src_paths: Final[SrcPaths] = src_paths
name: str = Field(init=False) self.name: Final[str] = self.src_paths.path.stem
services: tuple[Service, ...] = Field(init=False) self.services: Final[tuple[Service, ...]] = tuple(self.service_files)
volumes: tuple[VolumeFile, ...] = Field(init=False) self.volumes: Final[tuple[VolumeFile, ...]] = tuple(self.volume_files)
def __post_init__(self):
self.name = self.src_paths.path.stem
self.services = tuple(self.service_files)
self.volumes = tuple(self.volume_files)
@property @property
def service_files(self): def service_files(self) -> Iterator[Service]:
for path in self.src_paths.service_files: for path in self.src_paths.service_files:
yield Service.from_path(self, path) yield Service(self, path)
@property @property
def volume_files(self): def volume_files(self) -> Iterator[VolumeFile]:
for path in self.src_paths.volume_files: for path in self.src_paths.volume_files:
yield VolumeFile.from_path(path) yield VolumeFile(path)
@property @property
def networks_sub(self) -> Generator[dict[str, Any]]: def networks(self) -> Iterator[NetworkDict]:
for service in self.services: for service in self.services:
for network in service.networks: 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 @property
def as_dict(self) -> dict[Any, Any]: def as_dict(self) -> ComposeDict:
return RootModel[Compose](self).model_dump(exclude_none=True) # pyright: ignore[reportAny] return {
"name": self.name,
@model_serializer(mode="plain") "services": dict(ChainMap(*(s.as_dict for s in self.services))),
def dump(self) -> dict[str, Any]: "networks": dict(ChainMap(*(self.networks))),
return {self.name: ChainMap(*(s.as_dict for s in self.services))} "volumes": dict(ChainMap(*(vol.as_dict for vol in self.volumes))),
}
@override @override
def __str__(self) -> str: 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 from docker_compose.util import ReplaceStr
DN = ReplaceStr( DN = ReplaceStr(
fmt_replace_str("dn"), ReplaceStr.fmt("dn"),
"_".join(fmt_replace_str(s) for s in ("org", "name")), "_".join(ReplaceStr.fmt(s) for s in (ReStrings.ORG, ReStrings.APP)),
) )
FQDN = ReplaceStr( FQDN = ReplaceStr(
fmt_replace_str("fqdn"), ReplaceStr.fmt("fqdn"),
"_".join(fmt_replace_str(s) for s in ("org", "name", "services")), "_".join(ReplaceStr.fmt(s) for s in (ReStrings.ORG, ReStrings.APP, ReStrings.SERVICE)),
) )
DATA_DN = ReplaceStr( DATA_DN = ReplaceStr(
fmt_replace_str("data"), ReplaceStr.fmt("data"),
str(APP_ROOT.joinpath(*(fmt_replace_str(s) for s in ("org", "name")))), 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,16 +1,8 @@
from __future__ import annotations from __future__ import annotations
from typing import TYPE_CHECKING, Any, cast, final from typing import TYPE_CHECKING, Final, TypedDict, final
from pydantic import ( from autoslot import Slots
ConfigDict,
Field,
RootModel,
SerializationInfo,
SerializerFunctionWrapHandler,
model_serializer,
)
from pydantic.dataclasses import dataclass
from docker_compose.domain.compose.service import DN from docker_compose.domain.compose.service import DN
@@ -18,35 +10,22 @@ if TYPE_CHECKING:
from docker_compose.domain.compose.service.service import Service from docker_compose.domain.compose.service.service import Service
class NetworkDictSub(TypedDict):
name: str
external: bool
type NetworkDict = dict[str, NetworkDictSub]
@final @final
@dataclass(slots=True, config=ConfigDict(str_strip_whitespace=True)) class Network(Slots):
class Network: def __init__(self, service: Service, val: str) -> None:
val: str self.service: Final[Service] = service
service: Service = Field(init=False) self.val: Final[str] = val.strip()
self.name: Final[str] = f"{DN.repl}_{self.val}"
self.external: Final[bool] = "proxy" in self.val
@property @property
def name(self): def as_dict(self) -> NetworkDict:
return f"{DN.repl}_{self.val}" return {self.name: NetworkDictSub(name=self.name, external=self.external)}
@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]

View File

@@ -1,21 +1,16 @@
from typing import Self, final from typing import Final, final, override
from pydantic import model_serializer from autoslot import Slots
from pydantic.dataclasses import dataclass
@final @final
@dataclass(slots=True) class Port(Slots):
class Port:
sep = ":" 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 @override
dest: int def __str__(self) -> str:
@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:
return f"{self.src}{self.sep}{self.dest}" return f"{self.src}{self.sep}{self.dest}"

View File

@@ -1,26 +1,17 @@
from __future__ import annotations from __future__ import annotations
from collections.abc import Generator, Iterable from collections.abc import Generator, Iterable, Iterator
from dataclasses import InitVar from enum import StrEnum
from functools import reduce from functools import reduce
from itertools import chain
from pathlib import Path 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 import yaml
from pydantic import ( from autoslot import Slots
Field, from pydantic import TypeAdapter
RootModel,
SerializerFunctionWrapHandler,
computed_field,
field_validator,
model_serializer,
)
from pydantic.dataclasses import dataclass
from docker_compose import fmt_replace_str
from docker_compose.domain.compose.service import DN, FQDN 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.networks import Network
from docker_compose.domain.compose.service.port import Port from docker_compose.domain.compose.service.port import Port
from docker_compose.domain.compose.service.volumes import Volumes from docker_compose.domain.compose.service.volumes import Volumes
@@ -30,118 +21,160 @@ if TYPE_CHECKING:
from docker_compose.domain.compose.compose import Compose 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 @final
@dataclass(slots=True) class Service(Slots):
class Service: _traefik_labels: tuple[str, ...] = (
_traefik_labels = (
"traefik.enable=true", "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.http.routers.{DN.repl}.entrypoints=websecure",
f"traefik.docker.network={DN.repl}_proxy", f"traefik.docker.network={DN.repl}_proxy",
f"traefik.http.routers.{DN.repl}.tls.certresolver=le", 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 def __init__(self, compose: Compose, path: Path) -> None:
path: InitVar[Path] self.compose: Final[Compose] = compose
self.service_name: Final[str] = path.stem
image: str data: Final[ServiceReadDict] = self.load(path)
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]
)
command: tuple[str, ...] = Field( self.container_name: Final[str] = f"{DN.repl}_{self.service_name}"
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]
)
def __post_init__(self, path: Path) -> None: self.image: Final[str] = data["image"]
self.service_name = path.stem 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 @property
def container_name(self): def as_dict(self) -> dict[str, ServiceWriteDict]:
return f"{DN.repl}_{self.service_name}" 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]: def __iter__(self) -> Generator[ReplaceStr]:
yield FQDN yield FQDN
yield DN yield DN
yield ReplaceStr("application", self.service_name) yield ReplaceStr(ReplaceStr.fmt("service"), self.service_name)
def __call__(self, data: str) -> str: def __call__(self, data: str) -> str:
return reduce(lambda s, f: f(s), self, data) return reduce(lambda s, f: f(s), self, data)
@computed_field
@property @property
def labels(self): def labels(self) -> tuple[str, ...]:
if "traefik.enable=true" not in self.labels_raw: if "traefik.enable=true" not in self.labels_raw:
return 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( def string_lists(self, data: ServiceReadDict, key: str) -> tuple[str, ...]:
"command", return tuple(self.string_lists_sub(data.get(key, ())))
"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)
@classmethod @staticmethod
def from_path(cls, compose: Compose, path: Path) -> Self: 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: with path.open("rt") as f:
data = cast(dict[str, Any], yaml.safe_load(f)) data = yaml.safe_load(f) # pyright: ignore[reportAny]
return cls(compose, path, **data) # pyright: ignore[reportAny] return TypeAdapter(ServiceReadDict).validate_python(data)
@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]

View File

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

View File

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

View File

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

View File

@@ -1,40 +1,32 @@
from __future__ import annotations from __future__ import annotations
import secrets 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 autoslot import Slots
from pydantic.dataclasses import dataclass
if TYPE_CHECKING:
from docker_compose.domain.env.env_data import EnvData from docker_compose.domain.env.env_data import EnvData
# if TYPE_CHECKING:
# from docker_compose.env.env_data import EnvData
@final @final
@dataclass(slots=True, frozen=True, config=ConfigDict(str_strip_whitespace=True)) class EnvRow(Slots):
class EnvRow: def __init__(self, parent:EnvData, raw:str) -> None:
parent: EnvData key, val = (s.strip() for s in raw.split("="))
key: str
_val: str
@classmethod self.parent:Final = parent
def from_str(cls, parent: EnvData, raw: str) -> Self: self.key:Final= key.strip()
return cls(parent, *raw.split("=")) self._val:Final = val.strip()
if self.key.startswith("#"):
@field_validator("key", mode="after")
@classmethod
def strip_string(cls, s: str) -> str:
if s.startswith("#"):
raise ValueError raise ValueError
return s
@model_serializer(mode="plain")
def model_serializer(self) -> str:
return f"{self.key}={self.val}"
@computed_field
@property @property
def val(self) -> str: def val(self) -> str:
return self._val.replace("{_PSWD}", secrets.token_urlsafe(12)) 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,21 @@
from __future__ import annotations from __future__ import annotations
from pathlib import Path 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 import APP_ROOT
from docker_compose.domain.paths import FILES
if TYPE_CHECKING: if TYPE_CHECKING:
from docker_compose.domain.paths.org import OrgData from docker_compose.domain.paths.org import OrgData
@final @final
@dataclass class DestPath(Slots):
class DestPath: def __init__(self, org_data: OrgData) -> None:
org_data: OrgData self.org_data: Final[OrgData] = org_data
self.base_path: Final[Path] = APP_ROOT.joinpath(*self.org_data)
# @classmethod self.compose_path: Final[Path] = self.base_path.joinpath(FILES.COMPOSE)
# def from_path(cls, path: Path) -> Iterator[Self]: self.env_path: Final[Path] = self.base_path.joinpath(FILES.ENV)
# 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 collections.abc import Generator, Iterator
from enum import StrEnum from enum import StrEnum
from functools import reduce from functools import reduce
from typing import TYPE_CHECKING, Self, cast, final from typing import TYPE_CHECKING, Final, NotRequired, Self, TypedDict, final
import yaml import yaml
from pydantic import ConfigDict, Field, field_validator from autoslot import Slots
from pydantic.dataclasses import dataclass 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.paths.dest import DestPath
from docker_compose.domain.render.render import Render from docker_compose.domain.render.render import Render
from docker_compose.util import ReplaceStr from docker_compose.util import ReplaceStr
@@ -17,30 +17,35 @@ from docker_compose.util import ReplaceStr
if TYPE_CHECKING: if TYPE_CHECKING:
from docker_compose.domain.paths.src import SrcPaths from docker_compose.domain.paths.src import SrcPaths
class Orgs(StrEnum): class Orgs(StrEnum):
PERSONAL = "personal" PERSONAL = "personal"
STRYTEN = "stryten" STRYTEN = "stryten"
C4 = "c4" 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 @final
@dataclass( class OrgData(Slots):
slots=True, def __init__(self, src_paths:SrcPaths, app:str, org:str, url: str|None) -> None:
order=True, if url:
config=ConfigDict(use_enum_values=True, str_strip_whitespace=True), url = url.strip()
)
class OrgData: self.src_paths: Final = src_paths
src_paths: SrcPaths
app: str self.app: Final= app.strip()
org: Orgs self.org: Final=org.strip()
url: str | None self.url: Final = url if url else None
dest: DestPath = Field(init=False)
render: Render = Field(init=False) 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: def __call__(self, string: str) -> str:
return reduce(lambda s, f: f(s), self.replace_funcs, string) return reduce(lambda s, f: f(s), self.replace_funcs, string)
@@ -51,32 +56,21 @@ class OrgData:
@property @property
def _replace_args(self) -> Generator[tuple[str, str]]: def _replace_args(self) -> Generator[tuple[str, str]]:
yield "app", self.app yield ReStrings.APP, self.app
yield "org", self.org yield "org", self.org
yield "url", ".".join((self.url, "ccamper7", "net")) if self.url else "" yield "url", ".".join((self.url, "ccamper7", "net")) if self.url else ""
@property @property
def replace_funcs(self) -> Generator[ReplaceStr]: def replace_funcs(self) -> Generator[ReplaceStr]:
for s, r in self._replace_args: for s, r in self._replace_args:
yield ReplaceStr(fmt_replace_str(s), r) yield ReplaceStr(ReplaceStr.fmt(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()
@classmethod @classmethod
def from_src_path(cls, path: SrcPaths) -> Iterator[Self]: def from_src_path(cls, path: SrcPaths) -> Iterator[Self]:
# log_cls(cls, path=str(path)) # log_cls(cls, path=str(path))
validate = TypeAdapter[OrgDict](OrgDict).validate_python
with path.cfg_file.open("rt") as f: 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 app = path.cfg_file.stem
for org, _dict in data.items(): for org, _dict in validate(data).items():
yield cls(app=app, org=org, **_dict) # pyright: ignore[reportArgumentType] yield cls(path, app, org, _dict.get('url'))

View File

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

View File

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

View File

@@ -1,9 +1,8 @@
from __future__ import annotations from __future__ import annotations
from typing import TYPE_CHECKING, final, override from typing import TYPE_CHECKING, Final, final, override
from pydantic import Field from autoslot import Slots
from pydantic.dataclasses import dataclass
from docker_compose.domain.compose.compose import Compose from docker_compose.domain.compose.compose import Compose
from docker_compose.domain.render.bind_vols import BindVols from docker_compose.domain.render.bind_vols import BindVols
@@ -13,24 +12,11 @@ if TYPE_CHECKING:
@final @final
@dataclass(slots=True) class Render(Slots):
class Render: def __init__(self, org_data: OrgData) -> None:
org_data: OrgData self.org_data: Final[OrgData] = org_data
bind_vols: BindVols = Field(init=False)
def __post_init__(self):
self.bind_vols = BindVols(self) 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 @property
def template(self) -> Compose: def template(self) -> Compose:
return self.org_data.src_paths.compose return self.org_data.src_paths.compose
@@ -42,106 +28,3 @@ class Render:
def __call__(self): def __call__(self):
with self.org_data.dest.compose_path.open("wt") as f: with self.org_data.dest.compose_path.open("wt") as f:
_ = f.write(str(self)) _ = 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 import re
from functools import partial, reduce from collections.abc import Iterator, Sequence, Set
from typing import Any, final, override from dataclasses import dataclass
from re import Pattern
from typing import Any, Final, final, override
import yaml import yaml
from pydantic.dataclasses import dataclass from autoslot import Slots
@final @final
@dataclass @dataclass(frozen=True, slots=True)
class ReplaceStr: class ReplaceStr:
src: str src: str
repl: str repl: str
@@ -15,24 +17,79 @@ class ReplaceStr:
def __call__(self, s: str) -> str: def __call__(self, s: str) -> str:
return s.replace(self.src, self.repl) 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 @final
class YamlUtil: class YamlUtil(Slots):
indent = partial(re.compile(r"(^\s?-)", re.MULTILINE).sub, r" \g<1>") indent: Final[tuple[Pattern[str], str]] = (
port = partial(re.compile(r"(\W*?)(\d+:\d+)", re.MULTILINE).sub, r'\g<1>"\g<2>"') 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): class VerboseSafeDumper(yaml.SafeDumper):
@override @override
def ignore_aliases(self, data: object) -> bool: def ignore_aliases(self, data: object) -> bool:
return True return True
def __call__(self, data: dict[Any, Any]) -> str: def __call__(self) -> str:
return reduce( data = yaml.dump(self.data, Dumper=self.VerboseSafeDumper) # pyright: ignore[reportAny]
lambda s, f: f(s), for regex, repl in self:
self, data = regex.sub(repl, data)
yaml.dump(data, Dumper=self.VerboseSafeDumper), return data
)
def __iter__(self): def __iter__(self) -> Iterator[tuple[Pattern[str], str]]:
yield self.indent yield self.indent
yield self.port 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" }, { 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]] [[package]]
name = "basedpyright" name = "basedpyright"
version = "1.37.1" version = "1.37.1"
@@ -37,6 +46,7 @@ name = "docker-compose"
version = "0.1.0" version = "0.1.0"
source = { editable = "." } source = { editable = "." }
dependencies = [ dependencies = [
{ name = "autoslot" },
{ name = "basedpyright" }, { name = "basedpyright" },
{ name = "loguru" }, { name = "loguru" },
{ name = "pydantic" }, { name = "pydantic" },
@@ -47,6 +57,7 @@ dependencies = [
[package.metadata] [package.metadata]
requires-dist = [ requires-dist = [
{ name = "autoslot", specifier = ">=2025.11.1" },
{ name = "basedpyright", specifier = ">=1.37.1" }, { name = "basedpyright", specifier = ">=1.37.1" },
{ name = "loguru", specifier = ">=0.7.3" }, { name = "loguru", specifier = ">=0.7.3" },
{ name = "pydantic", specifier = ">=2.12.5" }, { name = "pydantic", specifier = ">=2.12.5" },