104 lines
2.7 KiB
Python
104 lines
2.7 KiB
Python
from typing import TYPE_CHECKING, Any, BinaryIO, Dict, List
|
|
|
|
from .container import Container
|
|
from .image import Image, ImageReference
|
|
from .method import Method
|
|
from .service import Service
|
|
|
|
|
|
if TYPE_CHECKING:
|
|
import tomli as tomllib
|
|
else:
|
|
try:
|
|
import tomllib
|
|
except ImportError:
|
|
import tomli as tomllib
|
|
|
|
|
|
class State:
|
|
__slots__ = ('images', 'containers', 'services', )
|
|
|
|
images: List[Image]
|
|
containers: List[Container]
|
|
services: List[Service]
|
|
|
|
def __init__(
|
|
self,
|
|
images: List[Image],
|
|
containers: List[Container],
|
|
services: List[Service],
|
|
) -> None:
|
|
self.images = images
|
|
self.containers = containers
|
|
self.services = services
|
|
|
|
def __repr__(self) -> str:
|
|
return f'State({repr(self.images)}, {repr(self.containers)}, {repr(self.services)})'
|
|
|
|
|
|
def image_from_toml(spec: Dict[str, Any]) -> Image:
|
|
"""
|
|
Loads an Image spec from toml
|
|
"""
|
|
return Image(str(spec['path']), str(spec['hash']))
|
|
|
|
|
|
def container_from_toml(spec: Dict[str, Any]) -> Container:
|
|
"""
|
|
Loads a Container spec from toml
|
|
"""
|
|
return Container(ImageReference(str(spec['image'])), str(spec['runtime']))
|
|
|
|
|
|
def service_from_toml(spec: Dict[str, Any]) -> Service:
|
|
"""
|
|
Loads a Service spec from toml
|
|
"""
|
|
methods: List[Method] = []
|
|
|
|
return Service(str(spec['name']), methods)
|
|
|
|
|
|
def from_toml(toml: BinaryIO) -> State:
|
|
"""
|
|
Loads the given toml and builds a state instance out of it
|
|
"""
|
|
toml_dict = tomllib.load(toml)
|
|
|
|
images: List[Image] = []
|
|
images_by_name: Dict[str, Image] = {}
|
|
containers: List[Container] = []
|
|
services: List[Service] = []
|
|
|
|
for name, spec in toml_dict.items():
|
|
if spec['apiVersion'] != 'v0':
|
|
raise NotImplementedError('apiVersion', spec['apiVersion'])
|
|
|
|
if spec['kind'] == 'Image':
|
|
image = image_from_toml(spec)
|
|
images.append(image)
|
|
|
|
assert name not in images_by_name, f'Duplicate image name: {name}'
|
|
images_by_name[name] = image
|
|
continue
|
|
|
|
if spec['kind'] == 'Container':
|
|
containers.append(container_from_toml(spec))
|
|
continue
|
|
|
|
if spec['kind'] == 'Service':
|
|
services.append(service_from_toml(spec))
|
|
continue
|
|
|
|
raise NotImplementedError(spec)
|
|
|
|
for container in containers:
|
|
if not isinstance(container.image, ImageReference):
|
|
continue
|
|
|
|
image_opt = images_by_name.get(container.image.name, None)
|
|
assert image_opt is not None, f'Image reference not resolved: {container.image.name}'
|
|
container.image = image_opt
|
|
|
|
return State(images, containers, services)
|