109 lines
3.3 KiB
Python
Executable file
109 lines
3.3 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
import argparse
|
|
import json
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from typing import Self
|
|
|
|
from python_on_whales import docker
|
|
from python_on_whales import DockerClient
|
|
|
|
PROJECT_ROOT_DIRECTORY = Path(__file__).parent.parent.resolve()
|
|
SERVICES_DIRECTORY = PROJECT_ROOT_DIRECTORY / "services"
|
|
AVAILIBLE_SERVICES = list(
|
|
service.name for service in SERVICES_DIRECTORY.iterdir()
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class Server:
|
|
server_name: str
|
|
services: list[str]
|
|
context_name: str | None = None
|
|
|
|
@property
|
|
def context(self) -> str:
|
|
return self.context_name if self.context_name else self.server_name
|
|
|
|
@classmethod
|
|
def parse_dict(cls, data: dict[str, Any]) -> Self:
|
|
if not (server_name := data.get("server_name")):
|
|
raise ValueError("The service is missing server_name property.")
|
|
|
|
if not (services := data.get("services")):
|
|
raise ValueError("The config is missing services property.")
|
|
|
|
if not isinstance(services, list):
|
|
raise ValueError("The services property should be an array.")
|
|
|
|
return cls(server_name=server_name, services=services)
|
|
|
|
|
|
@dataclass
|
|
class ServicesConfig:
|
|
servers: list[Server]
|
|
|
|
def search_service(self, service_name: str) -> Server:
|
|
servers_matched = []
|
|
for server in self.servers:
|
|
if service_name in server.services:
|
|
servers_matched.append(server)
|
|
if len(servers_matched) != 1:
|
|
raise ValueError("Two services with the same name found.")
|
|
|
|
return servers_matched.pop()
|
|
|
|
@classmethod
|
|
def parse_dict(cls, data: dict[str, Any]) -> Self:
|
|
if not (all_servers := data.get("servers")):
|
|
raise ValueError("The config is missing servers property.")
|
|
if not isinstance(all_servers, list):
|
|
raise ValueError("The servers property should be an array.")
|
|
servers = []
|
|
for server in all_servers:
|
|
if not isinstance(server, dict):
|
|
raise ValueError("The server should be an object.")
|
|
servers.append(Server.parse_dict(server))
|
|
|
|
return cls(servers=servers)
|
|
|
|
|
|
def _get_service_context(service_name: str, config: ServicesConfig) -> str:
|
|
context_names = [ctx.name for ctx in docker.context.list()]
|
|
server_config = config.search_service(service_name)
|
|
|
|
if server_config.context not in context_names:
|
|
raise ValueError("Context not found on docker host.")
|
|
|
|
return server_config.context
|
|
|
|
|
|
def deploy_service(service_name: str, config: ServicesConfig) -> str:
|
|
context = _get_service_context(service_name, config)
|
|
docker = DockerClient(context=context)
|
|
return str(docker.docker_cmd)
|
|
|
|
|
|
def load_config(config_file: Path | str) -> ServicesConfig:
|
|
config_file = Path(config_file)
|
|
unknown_data = json.loads(config_file.read_text())
|
|
return ServicesConfig.parse_dict(unknown_data)
|
|
|
|
|
|
def _add_args(parser: argparse.ArgumentParser):
|
|
parser.add_argument("services_file")
|
|
parser.add_argument("service_name", choices=AVAILIBLE_SERVICES)
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser("deployservice")
|
|
_add_args(parser)
|
|
parsed_args = parser.parse_args()
|
|
config = load_config(parsed_args.services_file)
|
|
print(deploy_service(parsed_args.service_name, config))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|