168 lines
4.4 KiB
Python
168 lines
4.4 KiB
Python
from typing import TYPE_CHECKING, Any, BinaryIO, Dict, List, Tuple
|
|
|
|
from . import valuetype
|
|
from .container import Container
|
|
from .image import Image, ImageReference
|
|
from .method import Method
|
|
from .service import ContainerMatch, 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_import_from_toml(spec: Dict[str, Any]) -> Tuple[str, Method]:
|
|
"""
|
|
Loads an Method spec from toml, when it comes in the form of Image.imports
|
|
"""
|
|
service = spec.pop('service')
|
|
method = spec.pop('method')
|
|
arg_types = spec.pop('arg_types', [])
|
|
return_type = spec.pop('return_type', 'none')
|
|
|
|
assert not spec, ('Unrecognized values in image.imports', spec)
|
|
|
|
return (str(service), Method(
|
|
str(method),
|
|
[
|
|
valuetype.LOOKUP_TABLE[x]
|
|
for x in arg_types
|
|
],
|
|
valuetype.LOOKUP_TABLE[return_type],
|
|
), )
|
|
|
|
|
|
def image_from_toml(spec: Dict[str, Any]) -> Image:
|
|
"""
|
|
Loads an Image spec from toml
|
|
"""
|
|
imports = [
|
|
image_import_from_toml(imp_spec)
|
|
for imp_spec in spec.get('imports', [])
|
|
]
|
|
|
|
return Image(str(spec['path']), str(spec['hash']), imports)
|
|
|
|
|
|
def container_from_toml(name: str, spec: Dict[str, Any]) -> Container:
|
|
"""
|
|
Loads a Container spec from toml
|
|
"""
|
|
return Container(name, ImageReference(str(spec['image'])), str(spec['runtime']))
|
|
|
|
|
|
def service_container_match_from_toml(spec: Dict[str, Any]) -> ContainerMatch:
|
|
"""
|
|
Loads a service.ContainerMatch spec from toml
|
|
"""
|
|
return ContainerMatch(str(spec['byName']))
|
|
|
|
|
|
def method_from_toml(spec: Dict[str, Any]) -> Method:
|
|
"""
|
|
Loads an Method spec from toml
|
|
"""
|
|
name = spec.pop('name')
|
|
arg_types = spec.pop('arg_types', [])
|
|
return_type = spec.pop('return_type', 'none')
|
|
|
|
assert not spec, ('Unrecognized values in image.imports', spec)
|
|
|
|
return Method(
|
|
str(name),
|
|
[
|
|
valuetype.LOOKUP_TABLE[x]
|
|
for x in arg_types
|
|
],
|
|
valuetype.LOOKUP_TABLE[return_type],
|
|
)
|
|
|
|
|
|
def service_from_toml(spec: Dict[str, Any]) -> Service:
|
|
"""
|
|
Loads a Service spec from toml
|
|
"""
|
|
spec.pop('apiVersion')
|
|
spec.pop('kind')
|
|
|
|
name = spec.pop('name')
|
|
container_match = spec.pop('containerMatch')
|
|
methods = spec.pop('methods', [])
|
|
|
|
assert not spec, ('Unrecognized values in service', spec)
|
|
|
|
return Service(str(name), service_container_match_from_toml(container_match), [
|
|
method_from_toml(x)
|
|
for x in 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(name, 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)
|