Compare commits
2 Commits
b04a997561
...
main
| Author | SHA256 | Date | |
|---|---|---|---|
| df9d786b70 | |||
| 0c5686b3a7 |
@@ -14,12 +14,14 @@ from docker_compose.domain.compose.volume_files import VolumeFile
|
||||
if TYPE_CHECKING:
|
||||
from docker_compose.domain.paths.src import SrcPaths
|
||||
|
||||
|
||||
class ComposeDict(TypedDict):
|
||||
name: str
|
||||
services: dict[str, ServiceWriteDict]
|
||||
networks: dict[str, NetworkDictSub]
|
||||
volumes: dict[str, Any]
|
||||
|
||||
|
||||
@final
|
||||
class Compose(Slots):
|
||||
def __init__(self, src_paths: SrcPaths) -> None:
|
||||
@@ -44,15 +46,12 @@ class Compose(Slots):
|
||||
for network in service.networks:
|
||||
yield network.as_dict
|
||||
|
||||
|
||||
@property
|
||||
def as_dict(self) -> ComposeDict:
|
||||
return {
|
||||
'name':self.name,
|
||||
"name": self.name,
|
||||
"services": dict(ChainMap(*(s.as_dict for s in self.services))),
|
||||
"networks": dict(ChainMap(
|
||||
*(self.networks)
|
||||
)),
|
||||
"networks": dict(ChainMap(*(self.networks))),
|
||||
"volumes": dict(ChainMap(*(vol.as_dict for vol in self.volumes))),
|
||||
}
|
||||
|
||||
|
||||
@@ -5,17 +5,19 @@ from typing import TYPE_CHECKING, Final, TypedDict, final
|
||||
from autoslot import Slots
|
||||
|
||||
from docker_compose.domain.compose.service import DN
|
||||
from docker_compose.domain.compose.service.service import Service
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from docker_compose.domain.compose.service.service import Service
|
||||
|
||||
|
||||
class NetworkDictSub(TypedDict):
|
||||
name: str
|
||||
external: bool
|
||||
|
||||
|
||||
type NetworkDict = dict[str, NetworkDictSub]
|
||||
|
||||
|
||||
@final
|
||||
class Network(Slots):
|
||||
def __init__(self, service: Service, val: str) -> None:
|
||||
@@ -24,7 +26,6 @@ class Network(Slots):
|
||||
self.name: Final[str] = f"{DN.repl}_{self.val}"
|
||||
self.external: Final[bool] = "proxy" in self.val
|
||||
|
||||
|
||||
@property
|
||||
def as_dict(self) -> NetworkDict:
|
||||
return {self.name: NetworkDictSub(name=self.name, external=self.external)}
|
||||
|
||||
@@ -11,7 +11,6 @@ import yaml
|
||||
from autoslot import Slots
|
||||
from pydantic import TypeAdapter
|
||||
|
||||
from docker_compose.domain.compose.compose import Compose
|
||||
from docker_compose.domain.compose.service import DN, FQDN
|
||||
from docker_compose.domain.compose.service.networks import Network
|
||||
from docker_compose.domain.compose.service.port import Port
|
||||
@@ -21,8 +20,12 @@ from docker_compose.util import ReplaceStr
|
||||
if TYPE_CHECKING:
|
||||
from docker_compose.domain.compose.compose import Compose
|
||||
|
||||
|
||||
class DependsOnDict(TypedDict):
|
||||
condition: Literal['service_started', 'service_healthy', 'service_completed_successfully']
|
||||
condition: Literal[
|
||||
"service_started", "service_healthy", "service_completed_successfully"
|
||||
]
|
||||
|
||||
|
||||
class HealthCheck(TypedDict):
|
||||
test: tuple[str, ...]
|
||||
@@ -31,9 +34,12 @@ class HealthCheck(TypedDict):
|
||||
retries: int | None
|
||||
start_period: str | None
|
||||
|
||||
|
||||
class ServiceReadKeys(StrEnum):
|
||||
image='image'
|
||||
restart='restart'
|
||||
image = "image"
|
||||
restart = "restart"
|
||||
|
||||
|
||||
class ServiceReadDict(TypedDict):
|
||||
image: str
|
||||
restart: NotRequired[str]
|
||||
@@ -70,6 +76,7 @@ class ServiceWriteDict(TypedDict):
|
||||
ports: tuple[str, ...]
|
||||
healthcheck: HealthCheck | None
|
||||
|
||||
|
||||
@final
|
||||
class Service(Slots):
|
||||
_traefik_labels: tuple[str, ...] = (
|
||||
@@ -89,21 +96,33 @@ class Service(Slots):
|
||||
|
||||
self.container_name: Final[str] = f"{DN.repl}_{self.service_name}"
|
||||
|
||||
self.image:Final[str] = data['image']
|
||||
self.user:Final[str | None] = data.get('user')
|
||||
self.shm_size:Final[str | None] = data.get('shm_size')
|
||||
self.restart:Final[str] = data.get('restart',"unless-stopped")
|
||||
self.depends_on:Final[str | dict[str, DependsOnDict] | None] = data.get('depends_on')
|
||||
self.command:Final[tuple[str, ...]] = self.string_lists(data, 'command')
|
||||
self.entrypoint:Final[tuple[str, ...]] = self.string_lists(data,'entrypoint')
|
||||
self.environment:Final[dict[str, str]] = self.string_dict(data.get('environment',{}) )
|
||||
self.labels_raw:Final[tuple[str, ...]] = self.string_lists(data,'labels')
|
||||
self.logging:Final[tuple[str, ...]] =self.string_lists( data,'logging')
|
||||
self.networks:Final[tuple[Network, ...]] = tuple(Network(self, s) for s in data.get('networks', ()) )
|
||||
self.security_opt:Final[tuple[str, ...]] = self.string_lists(data,'security_opts')
|
||||
self.volumes:Final[tuple[Volumes, ...]] = tuple(Volumes(self, s) for s in data.get('volumes', ()) )
|
||||
self.ports:Final[tuple[Port, ...]] = tuple(Port(s) for s in data.get('ports', ()))
|
||||
self.healthcheck:Final[HealthCheck | None] = data.get('healthcheck')
|
||||
self.image: Final[str] = data["image"]
|
||||
self.user: Final[str | None] = data.get("user")
|
||||
self.shm_size: Final[str | None] = data.get("shm_size")
|
||||
self.restart: Final[str] = data.get("restart", "unless-stopped")
|
||||
self.depends_on: Final[str | dict[str, DependsOnDict] | None] = data.get(
|
||||
"depends_on"
|
||||
)
|
||||
self.command: Final[tuple[str, ...]] = self.string_lists(data, "command")
|
||||
self.entrypoint: Final[tuple[str, ...]] = self.string_lists(data, "entrypoint")
|
||||
self.environment: Final[dict[str, str]] = self.string_dict(
|
||||
data.get("environment", {})
|
||||
)
|
||||
self.labels_raw: Final[tuple[str, ...]] = self.string_lists(data, "labels")
|
||||
self.logging: Final[tuple[str, ...]] = self.string_lists(data, "logging")
|
||||
self.networks: Final[tuple[Network, ...]] = tuple(
|
||||
Network(self, s) for s in data.get("networks", ())
|
||||
)
|
||||
self.security_opt: Final[tuple[str, ...]] = self.string_lists(
|
||||
data, "security_opts"
|
||||
)
|
||||
self.volumes: Final[tuple[Volumes, ...]] = tuple(
|
||||
Volumes(self, s) for s in data.get("volumes", ())
|
||||
)
|
||||
self.ports: Final[tuple[Port, ...]] = tuple(
|
||||
Port(s) for s in data.get("ports", ())
|
||||
)
|
||||
self.healthcheck: Final[HealthCheck | None] = data.get("healthcheck")
|
||||
|
||||
@property
|
||||
def as_dict(self) -> dict[str, ServiceWriteDict]:
|
||||
@@ -128,7 +147,6 @@ class Service(Slots):
|
||||
)
|
||||
}
|
||||
|
||||
|
||||
def __iter__(self) -> Generator[ReplaceStr]:
|
||||
yield FQDN
|
||||
yield DN
|
||||
|
||||
@@ -4,8 +4,6 @@ from typing import TYPE_CHECKING, Final, final, override
|
||||
|
||||
from autoslot import Slots
|
||||
|
||||
from docker_compose.domain.compose.service.service import Service
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from docker_compose.domain.compose.service.service import Service
|
||||
|
||||
@@ -13,6 +11,7 @@ if TYPE_CHECKING:
|
||||
@final
|
||||
class Volumes(Slots):
|
||||
sep = ":"
|
||||
|
||||
def __init__(self, service: Service, raw: str) -> None:
|
||||
src, dest = (s.strip() for s in raw.split(self.sep))
|
||||
self.service: Final[Service] = service
|
||||
|
||||
5
src/docker_compose/domain/env/env_data.py
vendored
5
src/docker_compose/domain/env/env_data.py
vendored
@@ -10,14 +10,13 @@ from docker_compose.domain.env.env_row import EnvRow
|
||||
if TYPE_CHECKING:
|
||||
from docker_compose.domain.paths.src import SrcPaths
|
||||
|
||||
|
||||
@final
|
||||
class EnvData(Slots):
|
||||
|
||||
def __init__(self, src_paths: SrcPaths) -> None:
|
||||
self.src_paths: Final = src_paths
|
||||
self.data: Final = tuple(self.lines)
|
||||
|
||||
|
||||
@property
|
||||
def lines(self) -> Generator[EnvRow]:
|
||||
with self.src_paths.env_file.open(mode="rt") as f:
|
||||
@@ -29,5 +28,3 @@ class EnvData(Slots):
|
||||
@property
|
||||
def as_list(self):
|
||||
return tuple(str(row) for row in self.data)
|
||||
|
||||
|
||||
@@ -7,7 +7,6 @@ from autoslot import Slots
|
||||
|
||||
from docker_compose import APP_ROOT
|
||||
from docker_compose.domain.paths import FILES
|
||||
from docker_compose.domain.paths.org import OrgData
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from docker_compose.domain.paths.org import OrgData
|
||||
@@ -20,4 +19,3 @@ class DestPath(Slots):
|
||||
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)
|
||||
|
||||
@@ -17,7 +17,9 @@ class SrcPaths(Slots):
|
||||
self.path: Final = path
|
||||
|
||||
self.compose: Final = Compose(self)
|
||||
self.cfg:Final[dict[Orgs,OrgData]]= {cast(Orgs,obj.org): obj for obj in OrgData.from_src_path(self)}
|
||||
self.cfg: Final[dict[Orgs, OrgData]] = {
|
||||
cast(Orgs, obj.org): obj for obj in OrgData.from_src_path(self)
|
||||
}
|
||||
self.env: Final = EnvData(self)
|
||||
self.compose_file: Final = self.path.joinpath(FILES.COMPOSE)
|
||||
self.bind_vol_path: Final = self.path.joinpath(FILES.BIND_VOLS)
|
||||
|
||||
@@ -5,7 +5,6 @@ from typing import TYPE_CHECKING, Final, final, override
|
||||
from autoslot import Slots
|
||||
|
||||
from docker_compose.domain.compose.compose import Compose
|
||||
from docker_compose.domain.paths.org import OrgData
|
||||
from docker_compose.domain.render.bind_vols import BindVols
|
||||
|
||||
if TYPE_CHECKING:
|
||||
@@ -29,4 +28,3 @@ class Render(Slots):
|
||||
def __call__(self):
|
||||
with self.org_data.dest.compose_path.open("wt") as f:
|
||||
_ = f.write(str(self))
|
||||
|
||||
|
||||
Reference in New Issue
Block a user