Compare commits
2 Commits
7f749380ff
...
b04a997561
| Author | SHA256 | Date | |
|---|---|---|---|
| b04a997561 | |||
| 13793f2d70 |
4
.gitignore
vendored
4
.gitignore
vendored
@@ -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
1
.idea/.gitignore
generated
vendored
Normal file
@@ -0,0 +1 @@
|
|||||||
|
workspace.xml
|
||||||
4
.vscode/settings.json
vendored
4
.vscode/settings.json
vendored
@@ -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"
|
||||||
]
|
]
|
||||||
|
|||||||
@@ -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",
|
||||||
|
|
||||||
|
]
|
||||||
|
|||||||
@@ -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()}}}"
|
|
||||||
|
|||||||
@@ -1,70 +1,60 @@
|
|||||||
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(
|
||||||
return {self.name: ChainMap(*(s.as_dict for s in self.services))}
|
*(self.networks)
|
||||||
|
)),
|
||||||
|
"volumes": dict(ChainMap(*(vol.as_dict for vol in self.volumes))),
|
||||||
|
}
|
||||||
|
|
||||||
@override
|
@override
|
||||||
def __str__(self) -> str:
|
def __str__(self) -> str:
|
||||||
|
|||||||
@@ -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)))),
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -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']}"
|
|
||||||
@@ -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()
|
|
||||||
@@ -1,52 +1,30 @@
|
|||||||
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
|
||||||
|
from docker_compose.domain.compose.service.service import Service
|
||||||
|
|
||||||
if TYPE_CHECKING:
|
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]
|
|
||||||
|
|||||||
@@ -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}"
|
||||||
|
|||||||
@@ -1,26 +1,18 @@
|
|||||||
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.compose import Compose
|
||||||
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
|
||||||
@@ -29,119 +21,142 @@ from docker_compose.util import ReplaceStr
|
|||||||
if TYPE_CHECKING:
|
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]
|
|
||||||
|
|||||||
@@ -1,32 +1,28 @@
|
|||||||
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
|
|
||||||
|
from docker_compose.domain.compose.service.service import Service
|
||||||
|
|
||||||
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 = ":"
|
||||||
|
def __init__(self, service:Service, raw:str) -> None:
|
||||||
service: Service
|
src, dest = (s.strip() for s in raw.split(self.sep))
|
||||||
|
self.service: Final[Service] = service
|
||||||
_src: str
|
self._src: Final[str]=src
|
||||||
dest: str
|
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}"
|
||||||
|
|||||||
@@ -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()
|
|
||||||
|
|||||||
28
src/docker_compose/domain/env/env_data.py
vendored
28
src/docker_compose/domain/env/env_data.py
vendored
@@ -1,26 +1,22 @@
|
|||||||
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
|
||||||
|
|
||||||
if TYPE_CHECKING:
|
if TYPE_CHECKING:
|
||||||
from docker_compose.domain.paths.src import SrcPaths
|
from docker_compose.domain.paths.src import SrcPaths
|
||||||
|
|
||||||
|
|
||||||
@final
|
@final
|
||||||
@dataclass(slots=True)
|
class EnvData(Slots):
|
||||||
class EnvData:
|
|
||||||
src_paths: SrcPaths
|
def __init__(self, src_paths:SrcPaths) -> None:
|
||||||
data: tuple[EnvRow, ...] = Field(init=False)
|
self.src_paths:Final = src_paths
|
||||||
|
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 +24,10 @@ 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)
|
||||||
|
|
||||||
|
@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]
|
|
||||||
|
|||||||
40
src/docker_compose/domain/env/env_row.py
vendored
40
src/docker_compose/domain/env/env_row.py
vendored
@@ -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}"
|
||||||
|
|
||||||
|
|||||||
@@ -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'
|
||||||
@@ -1,33 +1,23 @@
|
|||||||
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
|
||||||
|
from docker_compose.domain.paths.org import OrgData
|
||||||
|
|
||||||
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)
|
||||||
|
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")
|
|
||||||
|
|||||||
@@ -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'))
|
||||||
@@ -1,58 +1,34 @@
|
|||||||
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]]= {cast(Orgs,obj.org): obj for obj in OrgData.from_src_path(self)}
|
||||||
cfg: dict[Orgs, OrgData] = Field(init=False)
|
self.env:Final=EnvData(self)
|
||||||
env: EnvData = Field(init=False)
|
self.compose_file :Final= self.path.joinpath(FILES.COMPOSE)
|
||||||
|
self.bind_vol_path:Final= self.path.joinpath(FILES.BIND_VOLS)
|
||||||
def __post_init__(self):
|
self.service_files:Final=tuple(self.get_yaml_files(FILES.SERVICES))
|
||||||
self.compose = Compose(self)
|
self.volume_files:Final= tuple(self.get_yaml_files(FILES.VOLUMES))
|
||||||
self.cfg = {obj.org: obj for obj in OrgData.from_src_path(self)}
|
self.cfg_file:Final= self.path.joinpath(FILES.CFG)
|
||||||
self.env = EnvData(self)
|
self.env_file:Final= self.path.joinpath(FILES.ENV)
|
||||||
|
|
||||||
@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")
|
|
||||||
|
|
||||||
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
|
||||||
|
|
||||||
|
|||||||
@@ -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)
|
||||||
|
|
||||||
|
|||||||
@@ -1,11 +1,11 @@
|
|||||||
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.paths.org import OrgData
|
||||||
from docker_compose.domain.render.bind_vols import BindVols
|
from docker_compose.domain.render.bind_vols import BindVols
|
||||||
|
|
||||||
if TYPE_CHECKING:
|
if TYPE_CHECKING:
|
||||||
@@ -13,24 +13,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
|
||||||
@@ -43,105 +30,3 @@ class Render:
|
|||||||
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
|
|
||||||
|
|||||||
@@ -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
11
uv.lock
generated
@@ -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" },
|
||||||
|
|||||||
Reference in New Issue
Block a user