ccamper7
This commit is contained in:
@@ -11,8 +11,8 @@ class OrgDataYaml(TypedDict):
|
|||||||
|
|
||||||
|
|
||||||
class CfgDataYaml(TypedDict):
|
class CfgDataYaml(TypedDict):
|
||||||
name: str
|
services: list[str]
|
||||||
files: list[str]
|
volumes: NotRequired[list[str]]
|
||||||
orgs: list[OrgDataYaml]
|
orgs: list[OrgDataYaml]
|
||||||
|
|
||||||
|
|
||||||
@@ -28,5 +28,6 @@ class OrgData:
|
|||||||
class CfgData:
|
class CfgData:
|
||||||
src_paths: SrcPaths
|
src_paths: SrcPaths
|
||||||
name: str
|
name: str
|
||||||
files: frozenset[Path]
|
services: frozenset[Path]
|
||||||
|
volumes: frozenset[Path] | None
|
||||||
orgs: frozenset[OrgData]
|
orgs: frozenset[OrgData]
|
||||||
|
|||||||
@@ -3,16 +3,17 @@ from typing import cast
|
|||||||
from compose.cfg.entity import CfgData, CfgDataYaml
|
from compose.cfg.entity import CfgData, CfgDataYaml
|
||||||
from compose.cfg.get import cfg_get_orgs
|
from compose.cfg.get import cfg_get_orgs
|
||||||
from compose.src_path.entity import SrcPaths
|
from compose.src_path.entity import SrcPaths
|
||||||
from compose.src_path.get import src_path_get_files
|
from compose.src_path.get import src_path_get_services, src_path_get_volumes
|
||||||
from compose.util import read_yml
|
from compose.util import read_yml
|
||||||
|
|
||||||
|
|
||||||
def cfg_data_factory(cfg_dir: SrcPaths) -> CfgData:
|
def cfg_data_factory(src_paths: SrcPaths) -> CfgData:
|
||||||
data = cast(CfgDataYaml, read_yml(cfg_dir.cfg_file))
|
data = cast(CfgDataYaml, read_yml(src_paths.cfg_file))
|
||||||
|
vols = frozenset(src_path_get_volumes(src_paths, data))
|
||||||
return CfgData(
|
return CfgData(
|
||||||
cfg_dir,
|
src_paths,
|
||||||
data["name"],
|
src_paths.cfg_dir.name,
|
||||||
frozenset(src_path_get_files(cfg_dir, data)),
|
frozenset(src_path_get_services(src_paths, data)),
|
||||||
|
vols if vols else None,
|
||||||
frozenset(cfg_get_orgs(data)),
|
frozenset(cfg_get_orgs(data)),
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -2,6 +2,7 @@ from collections.abc import Iterator
|
|||||||
from typing import cast
|
from typing import cast
|
||||||
|
|
||||||
from compose.cfg.entity import CfgData, CfgDataYaml, OrgData
|
from compose.cfg.entity import CfgData, CfgDataYaml, OrgData
|
||||||
|
from compose.compose.entity import VolYaml
|
||||||
from compose.service.entity import Service, ServiceYaml
|
from compose.service.entity import Service, ServiceYaml
|
||||||
from compose.util import read_yml
|
from compose.util import read_yml
|
||||||
|
|
||||||
@@ -21,6 +22,14 @@ def cfg_get_orgs(data: CfgDataYaml) -> Iterator[OrgData]:
|
|||||||
|
|
||||||
|
|
||||||
def cfg_get_services(cfg_data: CfgData) -> Iterator[tuple[str, Service]]:
|
def cfg_get_services(cfg_data: CfgData) -> Iterator[tuple[str, Service]]:
|
||||||
for path in cfg_data.files:
|
for path in cfg_data.services:
|
||||||
_dict = cast(ServiceYaml, read_yml(path))
|
_dict = cast(ServiceYaml, read_yml(path))
|
||||||
yield path.stem, Service.from_dict(_dict)
|
yield path.stem, Service.from_dict(_dict)
|
||||||
|
|
||||||
|
|
||||||
|
def cfg_get_volumes(cfg_data: CfgData) -> Iterator[tuple[str, VolYaml]]:
|
||||||
|
vols = cfg_data.volumes
|
||||||
|
if vols is None:
|
||||||
|
return
|
||||||
|
for path in vols:
|
||||||
|
yield path.stem, cast(VolYaml, read_yml(path))
|
||||||
|
|||||||
@@ -1,52 +1,67 @@
|
|||||||
from dataclasses import asdict, dataclass
|
from dataclasses import asdict, dataclass
|
||||||
from typing import Literal, NotRequired, Self, TypedDict, final
|
from typing import Literal, NotRequired, Self, TypedDict, final
|
||||||
|
|
||||||
|
from compose.cfg import T_YamlDict
|
||||||
from compose.cfg.entity import CfgData
|
from compose.cfg.entity import CfgData
|
||||||
from compose.net.entities import Net, NetTraefik, NetYaml
|
from compose.net.entities import Net, NetTraefik, NetYaml
|
||||||
from compose.service.entity import Service, ServiceYaml, TraefikService
|
from compose.service.entity import Service, ServiceYaml, TraefikService
|
||||||
from compose.service.get import services_get_networks
|
from compose.service.get import services_get_networks
|
||||||
from compose.util import to_yaml
|
from compose.util import to_yaml
|
||||||
|
|
||||||
|
type VolYaml = dict[str, T_YamlDict]
|
||||||
|
|
||||||
|
|
||||||
class ComposeYaml(TypedDict):
|
class ComposeYaml(TypedDict):
|
||||||
name: str
|
name: str
|
||||||
networks: NotRequired[NetYaml]
|
|
||||||
services: dict[str, ServiceYaml]
|
services: dict[str, ServiceYaml]
|
||||||
|
networks: NotRequired[NetYaml]
|
||||||
|
volumes: NotRequired[dict[str, T_YamlDict]]
|
||||||
|
|
||||||
|
|
||||||
@final
|
@final
|
||||||
@dataclass(frozen=True, slots=True)
|
@dataclass(frozen=True, slots=True)
|
||||||
class Compose:
|
class Compose:
|
||||||
cfg: CfgData
|
cfg: CfgData
|
||||||
networks: Net | None
|
|
||||||
services: dict[str, Service]
|
services: dict[str, Service]
|
||||||
|
networks: Net | None
|
||||||
|
volumes: VolYaml | None
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def from_dict(cls, cfg: CfgData, data: ComposeYaml) -> Self:
|
def from_dict(cls, cfg: CfgData, data: ComposeYaml) -> Self:
|
||||||
# services = dict[str, ComposeService]()
|
# services = dict[str, ComposeService]()
|
||||||
services = dict(get_services_dict(data))
|
services = dict(_get_services_dict(data))
|
||||||
|
# vols = frozenset(_get_volumes_dict(data))
|
||||||
return cls(
|
return cls(
|
||||||
cfg,
|
cfg,
|
||||||
services_get_networks(services.values()),
|
|
||||||
services,
|
services,
|
||||||
|
services_get_networks(services.values()),
|
||||||
|
data.get("volumes"),
|
||||||
)
|
)
|
||||||
|
|
||||||
def as_yaml(self) -> str:
|
def as_yaml(self) -> str:
|
||||||
return to_yaml(asdict(self))
|
return to_yaml(asdict(self))
|
||||||
|
|
||||||
|
|
||||||
def get_services_dict(data: ComposeYaml):
|
def _get_services_dict(data: ComposeYaml):
|
||||||
for k, v in data["services"].items():
|
for k, v in data["services"].items():
|
||||||
yield k, Service.from_dict(v)
|
yield k, Service.from_dict(v)
|
||||||
|
|
||||||
|
|
||||||
|
# def _get_volumes_dict(data: ComposeYaml) -> Iterator[VolYaml]:
|
||||||
|
# vols = data.get("volumes")
|
||||||
|
# if vols is None:
|
||||||
|
# return
|
||||||
|
# for k, v in vols.items():
|
||||||
|
# yield {k: v}
|
||||||
|
|
||||||
|
|
||||||
@final
|
@final
|
||||||
@dataclass(frozen=True, slots=True)
|
@dataclass(frozen=True, slots=True)
|
||||||
class TraefikCompose:
|
class TraefikCompose:
|
||||||
cfg: CfgData
|
cfg: CfgData
|
||||||
networks: NetTraefik
|
|
||||||
services: dict[Literal["traefik"], TraefikService]
|
services: dict[Literal["traefik"], TraefikService]
|
||||||
|
networks: NetTraefik
|
||||||
|
volumes: None
|
||||||
|
|
||||||
def as_yaml(self) -> str:
|
def as_yaml(self) -> str:
|
||||||
return to_yaml(asdict(self))
|
return to_yaml(asdict(self))
|
||||||
|
|||||||
@@ -1,7 +1,6 @@
|
|||||||
|
|
||||||
from compose.cfg.entity import CfgData
|
from compose.cfg.entity import CfgData
|
||||||
from compose.cfg.get import cfg_get_services
|
from compose.cfg.get import cfg_get_services, cfg_get_volumes
|
||||||
from compose.compose.entity import Compose
|
from compose.compose.entity import Compose, VolYaml
|
||||||
|
|
||||||
# from compose.service.factory import get_traefik_service
|
# from compose.service.factory import get_traefik_service
|
||||||
from compose.service.get import services_get_networks
|
from compose.service.get import services_get_networks
|
||||||
@@ -9,10 +8,13 @@ from compose.service.get import services_get_networks
|
|||||||
|
|
||||||
def compose_factory(cfg_data: CfgData) -> Compose:
|
def compose_factory(cfg_data: CfgData) -> Compose:
|
||||||
services = dict(cfg_get_services(cfg_data))
|
services = dict(cfg_get_services(cfg_data))
|
||||||
|
vols: VolYaml | None = dict(cfg_get_volumes(cfg_data))
|
||||||
|
|
||||||
return Compose(
|
return Compose(
|
||||||
cfg_data,
|
cfg_data,
|
||||||
services_get_networks(services.values()),
|
|
||||||
services,
|
services,
|
||||||
|
services_get_networks(services.values()),
|
||||||
|
vols if vols else None,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -2,6 +2,14 @@ from compose.cfg.entity import CfgDataYaml
|
|||||||
from compose.src_path.entity import SrcPaths
|
from compose.src_path.entity import SrcPaths
|
||||||
|
|
||||||
|
|
||||||
def src_path_get_files(src_paths: SrcPaths, data: CfgDataYaml):
|
def src_path_get_services(src_paths: SrcPaths, data: CfgDataYaml):
|
||||||
for path in data["files"]:
|
for path in data["services"]:
|
||||||
|
yield src_paths.cfg_dir.joinpath(path)
|
||||||
|
|
||||||
|
|
||||||
|
def src_path_get_volumes(src_paths: SrcPaths, data: CfgDataYaml):
|
||||||
|
vols = data.get("volumes")
|
||||||
|
if vols is None:
|
||||||
|
return
|
||||||
|
for path in vols:
|
||||||
yield src_paths.cfg_dir.joinpath(path)
|
yield src_paths.cfg_dir.joinpath(path)
|
||||||
|
|||||||
@@ -31,12 +31,7 @@ def template_factory(compose: Compose | TraefikCompose) -> Iterator[Template]:
|
|||||||
if not vols:
|
if not vols:
|
||||||
vols = None
|
vols = None
|
||||||
|
|
||||||
orgs = cfg_data.orgs
|
for org_data in cfg_data.orgs:
|
||||||
if orgs is None:
|
|
||||||
yield Template(compose, None, vols)
|
|
||||||
return
|
|
||||||
|
|
||||||
for org_data in orgs:
|
|
||||||
args = replace_args_factory(
|
args = replace_args_factory(
|
||||||
cfg_data,
|
cfg_data,
|
||||||
org_data,
|
org_data,
|
||||||
|
|||||||
1
src/compose/volumes/entity.py
Normal file
1
src/compose/volumes/entity.py
Normal file
@@ -0,0 +1 @@
|
|||||||
|
|
||||||
Reference in New Issue
Block a user