sync

src/docker_compose/domain/compose/compose.py

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

src/docker_compose/domain/compose/service/networks.py

@@ -5,25 +5,26 @@ 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:
+    def __init__(self, service: Service, val: str) -> None:
         self.service: Final[Service] = service
         self.val: Final[str] = val.strip()
-        self.name:Final[str] = f"{DN.repl}_{self.val}"
-        self.external :Final[bool]= "proxy" in self.val
-
+        self.name: Final[str] = f"{DN.repl}_{self.val}"
+        self.external: Final[bool] = "proxy" in self.val
 
     @property
     def as_dict(self) -> NetworkDict:

src/docker_compose/domain/compose/service/service.py

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

src/docker_compose/domain/compose/service/volumes.py

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

src/docker_compose/domain/compose/volume_files.py

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

src/docker_compose/domain/env/env_data.py

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

src/docker_compose/domain/paths/dest.py

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

src/docker_compose/domain/paths/src.py

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

src/docker_compose/domain/render/bind_vols.py

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

src/docker_compose/domain/render/render.py

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