PK!B;cnab/__init__.pyfrom cnab.types import ( Action, Credential, ImagePlatform, Ref, Image, InvocationImage, Maintainer, Destination, Metadata, Parameter, Bundle, ) from cnab.cnab import CNAB from cnab.invocation_image import CNABDirectory PK!պff cnab/cnab.pyimport json import tempfile from typing import Union from cnab.types import Bundle, Action from cnab.util import extract_docker_images class CNAB: bundle: Bundle name: str def __init__(self, bundle: Union[Bundle, dict, str], name: str = None): if isinstance(bundle, Bundle): self.bundle = bundle elif isinstance(bundle, dict): self.bundle = Bundle.from_dict(bundle) elif isinstance(bundle, str): with open(bundle) as f: data = json.load(f) self.bundle = Bundle.from_dict(data) else: raise TypeError self.name = name or self.bundle.name def run(self, action: str, credentials: dict = {}, parameters: dict = {}): import docker # type: ignore # check if action is supported assert action in self.actions client = docker.from_env() docker_images = extract_docker_images(self.bundle.invocation_images) assert len(docker_images) == 1 # check if parameters passed in are in bundle parameters errors = [] for key in parameters: if key not in self.bundle.parameters: errors.append(f"Invalid parameter provided: {key}") assert len(errors) == 0 # check if required parameters have been passed in required = [] for param in self.bundle.parameters: parameter = self.bundle.parameters[param] if parameter.required: required.append(param) for param in required: assert param in parameters # validate passed in params for param in parameters: parameter = self.bundle.parameters[param] if parameter.allowed_values: assert param in parameter.allowed_values if isinstance(param, int): if parameter.max_value: assert param <= parameter.max_value if parameter.min_value: assert param >= parameter.min_value elif isinstance(param, str): if parameter.max_length: assert len(param) <= parameter.max_length if parameter.min_length: assert len(param) >= parameter.min_length env = { "CNAB_INSTALLATION_NAME": self.name, "CNAB_BUNDLE_NAME": self.bundle.name, "CNAB_ACTION": action, } # build environment hash for param in self.bundle.parameters: parameter = self.bundle.parameters[param] if parameter.destination: if parameter.destination.env: key = parameter.destination.env value = ( parameters[param] if param in parameters else parameter.default_value ) env[key] = value if parameter.destination.path: # not yet supported pass mounts = [] if self.bundle.credentials: for name in self.bundle.credentials: # check credential has been provided assert name in credentials credential = self.bundle.credentials[name] if credential.env: # discussing behavour in https://github.com/deislabs/cnab-spec/issues/69 assert credential.env[:5] != "CNAB_" env[credential.env] = credentials[name] if credential.path: tmp = tempfile.NamedTemporaryFile(mode="w+", delete=True) tmp.write(credentials[name]) tmp.flush() mounts.append( docker.types.Mount( target=credential.path, source=tmp.name, read_only=True, type="bind", ) ) # Mount image maps for runtime usage tmp = tempfile.NamedTemporaryFile(mode="w+", delete=True) tmp.write(json.dumps(self.bundle.images)) tmp.flush() mounts.append( docker.types.Mount( target="/cnab/app/image-map.json", source=tmp.name, read_only=True, type="bind", ) ) return client.containers.run( docker_images[0].image, "/cnab/app/run", auto_remove=False, remove=True, environment=env, mounts=mounts, ) @property def actions(self) -> dict: actions = { "install": Action(modifies=True), "uninstall": Action(modifies=True), "upgrade": Action(modifies=True), } if self.bundle.actions: actions.update(self.bundle.actions) return actions @property def parameters(self) -> dict: return self.bundle.parameters @property def credentials(self) -> dict: return self.bundle.credentials PK!}V~g g cnab/invocation_image.pyimport os from typing import Union, List class InvalidCNABDirectoryError(Exception): pass class CNABDirectory(object): path: str def __init__(self, path: str): self.path = path def has_cnab_directory(self) -> bool: cnab = os.path.join(self.path, "cnab") return os.path.isdir(cnab) def has_app_directory(self) -> bool: app = os.path.join(self.path, "cnab", "app") return os.path.isdir(app) def has_no_misc_files_in_cnab_dir(self) -> bool: cnab = os.path.join(self.path, "cnab") disallowed_dirs: List[str] = [] disallowed_files: List[str] = [] for root, dirs, files in os.walk(cnab): disallowed_dirs = [x for x in dirs if x not in ["app", "build"]] disallowed_files = [ x for x in files if x not in ["LICENSE", "README.md", "README.txt"] ] break if disallowed_dirs or disallowed_files: return False else: return True def has_run(self) -> bool: run = os.path.join(self.path, "cnab", "app", "run") return os.path.isfile(run) def has_executable_run(self) -> bool: run = os.path.join(self.path, "cnab", "app", "run") return os.access(run, os.X_OK) def readme(self) -> Union[bool, str]: readme = os.path.join(self.path, "cnab", "README") txt = readme + ".txt" md = readme + ".md" if os.path.isfile(txt): with open(txt, "r") as content: return content.read() elif os.path.isfile(md): with open(md, "r") as content: return content.read() else: return False def license(self) -> Union[bool, str]: license = os.path.join(self.path, "cnab", "LICENSE") if os.path.isfile(license): with open(license, "r") as content: return content.read() else: return False def valid(self) -> bool: errors = [] if not self.has_executable_run(): errors.append("Run entrypoint is not executable") if not self.has_run(): errors.append("Missing a run entrypoint") if not self.has_app_directory(): errors.append("Missing the app directory") if not self.has_cnab_directory(): errors.append("Missing the cnab directory") if not self.has_no_misc_files_in_cnab_dir(): errors.append("Has additional files in the cnab directory") if len(errors) == 0: return True else: raise InvalidCNABDirectoryError(errors) PK!a  cnab/test_cnab.pyimport pytest # type: ignore from cnab import CNAB, Bundle, InvocationImage class HelloWorld(object): @pytest.fixture def app(self): return CNAB("fixtures/helloworld/bundle.json") class TestHelloWorld(HelloWorld): @pytest.mark.parametrize("action", ["install", "upgrade", "uninstall"]) def test_actions_present(self, app, action): assert action in app.actions def test_credentials_empty(self, app): assert app.credentials == None def test_port_parameter_present(self, app): assert "port" in app.parameters def test_port_details(self, app): assert app.parameters["port"].type == "int" assert app.parameters["port"].default_value == 8080 def test_app_name(self, app): assert app.name == "helloworld" def test_version(self, app): assert app.bundle.version == "0.1.1" def test_app_bundle(self, app): assert isinstance(app.bundle, Bundle) def test_invocation_images(self, app): assert len(app.bundle.invocation_images) == 1 @pytest.mark.docker class TestIntegrationHelloWorld(HelloWorld): @pytest.fixture def install(self, app): return str(app.run("install", parameters={"port": 9090})) def test_run(self, install): assert "install" in install class TestHelloHelm(object): @pytest.fixture def app(self): return CNAB("fixtures/hellohelm/bundle.json") @pytest.mark.parametrize("action", ["install", "upgrade", "uninstall", "status"]) def test_actions_present(self, app, action): assert action in app.actions def test_app_from_dict(): bundle = { "name": "helloworld", "version": "0.1.1", "invocationImages": [ {"imageType": "docker", "image": "cnab/helloworld:latest"} ], "images": {}, "parameters": { "port": { "defaultValue": 8080, "type": "int", "destination": {"env": "PORT"}, "metadata": {"descriptiob": "the public port"}, } }, "maintainers": [ {"email": "test@example.com", "name": "test", "url": "example.com"} ], } assert CNAB(bundle) def test_app_from_bundle(): bundle = Bundle( name="sample", version="0.1.0", invocation_images=[ InvocationImage(image_type="docker", image="garethr/helloworld:0.1.0") ], ) assert CNAB(bundle) def test_app_from_invalid_input(): with pytest.raises(TypeError): CNAB(1) PK!ACcnab/test_invocation_image.pyimport pytest # type: ignore from cnab import CNABDirectory from cnab.invocation_image import InvalidCNABDirectoryError class SampleCNAB(object): @pytest.fixture def directory(self): return CNABDirectory("fixtures/invocationimage") class TestCNABDirectory(SampleCNAB): def test_has_app_dir(self, directory): assert directory.has_app_directory() def test_has_cnab_dir(self, directory): assert directory.has_cnab_directory() def test_has_readme(self, directory): assert isinstance(directory.readme(), str) def test_has_license(self, directory): assert isinstance(directory.license(), str) def test_has_no_misc_files(self, directory): assert directory.has_no_misc_files_in_cnab_dir() def test_has_run(self, directory): assert directory.has_run() def test_has_executable(self, directory): assert directory.has_executable_run() def test_is_valid(self, directory): assert directory.valid() class InvalidCNAB(object): @pytest.fixture def directory(self): return CNABDirectory("fixtures/invalidinvocationimage") class TestInvalidCNABDirectory(InvalidCNAB): def test_has_no_app_dir(self, directory): assert not directory.has_app_directory() def test_has_cnab_dir(self, directory): assert directory.has_cnab_directory() def test_has_readme(self, directory): assert isinstance(directory.readme(), str) def test_has_no_license(self, directory): assert not directory.license() def test_has_invalid_misc_files(self, directory): assert not directory.has_no_misc_files_in_cnab_dir() def test_has_no_run(self, directory): assert not directory.has_run() def test_has_no_executable(self, directory): assert not directory.has_executable_run() def test_is_invalid(self, directory): with pytest.raises(InvalidCNABDirectoryError): directory.valid() PK!( %%cnab/test_types.pyimport json from cnab import ( Bundle, Credential, InvocationImage, Action, Parameter, Metadata, Maintainer, Destination, ) import pytest # type: ignore class TestMinimalParameters(object): @pytest.fixture def bundle(self): return Bundle( name="sample", version="0.1.0", invocation_images=[ InvocationImage(image_type="docker", image="garethr/helloworld:0.1.0") ], ) def test_bundle_images_empty(self, bundle): assert bundle.images == {} def test_bundle_parameters_empty(self, bundle): assert bundle.parameters == {} def test_bundle_credentials_empty(self, bundle): assert bundle.credentials == {} def test_bundle_default_schema_version(self, bundle): assert bundle.schema_version == "v1" def test_bundle_keywords_empty(self, bundle): assert bundle.keywords == [] def test_bundle_actions_empty(self, bundle): assert bundle.actions == {} def test_bundle_maintainers_empty(self, bundle): assert bundle.maintainers == [] def test_convert_bundle_to_dict(self, bundle): assert isinstance(bundle.to_dict(), dict) def test_bundle_description_blank(self, bundle): assert not bundle.description def test_convert_bundle_to_json(self, bundle): assert isinstance(bundle.to_json(), str) def test_convert_bundle_to_pretty_json(self, bundle): assert isinstance(bundle.to_json(pretty=True), str) def test_read_bundle(): with open("fixtures/helloworld/bundle.json") as f: data = json.load(f) assert isinstance(Bundle.from_dict(data), Bundle) class TestAllParameters(object): @pytest.fixture def bundle(self): return Bundle( name="sample", version="0.1.0", invocation_images=[ InvocationImage(image_type="docker", image="garethr/helloworld:0.1.0") ], actions={ "status": Action(modifies=False), "explode": Action(modifies=True), }, parameters={ "port": Parameter( type="int", default_value=8080, destination=Destination(env="PORT"), metadata=Metadata(description="the public port"), ) }, credentials={"kubeconfig": Credential(path="/root/.kube/config")}, description="test", keywords=["test1", "test2"], maintainers=[ Maintainer(email="test@example.com", name="test", url="example.com") ], images={}, schema_version="v2", ) def test_bundle_set_schema_version(self, bundle): assert bundle.schema_version == "v2" def test_bundle_set_description(self, bundle): assert bundle.description == "test" @pytest.mark.parametrize("keyword", ["test1", "test2"]) def test_bundle_set_keywords(self, bundle, keyword): assert keyword in bundle.keywords @pytest.mark.parametrize("action", ["status", "explode"]) def test_bundle_set_actions(self, bundle, action): assert action in bundle.actions def test_bundle_set_maintainer(self, bundle): assert len(bundle.maintainers) == 1 def test_bundle_set_credentials(self, bundle): assert len(bundle.credentials) == 1 def test_bundle_kubeconfig_credential(self, bundle): assert "kubeconfig" in bundle.credentials def test_bundle_set_parameters(self, bundle): assert len(bundle.parameters) == 1 def test_bundle_port_parameter(self, bundle): assert "port" in bundle.parameters def test_convert_bundle_to_dict(self, bundle): assert isinstance(bundle.to_dict(), dict) PK!cnab/test_util.pyimport pytest # type: ignore import cnab.util from cnab import InvocationImage class TestExtractImages(object): @pytest.fixture def filtered(self): images = [ InvocationImage(image="oci"), InvocationImage(image="docker", image_type="docker"), ] return cnab.util.extract_docker_images(images) def test_filtered_images(self, filtered): assert len(filtered) == 1 def test_extract_docker_images(self, filtered): assert filtered[0].image == "docker" PK!DB>> cnab/types.pyimport canonicaljson # type: ignore from dataclasses import dataclass, field from typing import Optional, Any, List, Union, Dict, TypeVar, Callable, Type, cast T = TypeVar("T") def from_bool(x: Any) -> bool: if not isinstance(x, bool): raise Exception(f"{x} not a boolean") return x def from_none(x: Any) -> Any: if not x is None: raise Exception(f"{x} not None") return x def from_union(fs, x): for f in fs: try: return f(x) except: pass assert False def from_str(x: Any) -> str: if not isinstance(x, str): raise Exception(f"{x} not a string") return x def from_list(f: Callable[[Any], T], x: Any) -> List[T]: if not isinstance(x, list): raise Exception(f"{x} not a list") return [f(y) for y in x] def from_int(x: Any) -> int: if not (isinstance(x, int) and not isinstance(x, bool)): raise Exception(f"{x} not an integer") return x def to_class(c: Type[T], x: Any) -> dict: if not isinstance(x, c): raise Exception(f"{x} not a {c}") return cast(Any, x).to_dict() def from_dict(f: Callable[[Any], T], x: Any) -> Dict[str, T]: if not isinstance(x, dict): raise Exception(f"{x} not a dictionary") return {k: f(v) for (k, v) in x.items()} def clean(result: Dict) -> dict: return {k: v for k, v in result.items() if v} @dataclass class Action: modifies: Optional[bool] = None stateless: Optional[bool] = None description: Optional[str] = None @staticmethod def from_dict(obj: Any) -> "Action": assert isinstance(obj, dict) modifies = from_union([from_bool, from_none], obj.get("modifies")) stateless = from_union([from_bool, from_none], obj.get("stateless")) description = from_union([from_str, from_none], obj.get("description")) return Action(modifies, stateless, description) def to_dict(self) -> dict: result: dict = {} result["modifies"] = from_union([from_bool, from_none], self.modifies) result["stateless"] = from_union([from_bool, from_none], self.stateless) result["description"] = from_union([from_str, from_none], self.description) return clean(result) @dataclass class Credential: description: Optional[str] = None env: Optional[str] = None path: Optional[str] = None @staticmethod def from_dict(obj: Any) -> "Credential": assert isinstance(obj, dict) description = from_union([from_str, from_none], obj.get("description")) env = from_union([from_str, from_none], obj.get("env")) path = from_union([from_str, from_none], obj.get("path")) return Credential(description, env, path) def to_dict(self) -> dict: result: dict = {} result["description"] = from_union([from_str, from_none], self.description) result["env"] = from_union([from_str, from_none], self.env) result["path"] = from_union([from_str, from_none], self.path) return clean(result) @dataclass class ImagePlatform: architecture: Optional[str] = None os: Optional[str] = None @staticmethod def from_dict(obj: Any) -> "ImagePlatform": assert isinstance(obj, dict) architecture = from_union([from_str, from_none], obj.get("architecture")) os = from_union([from_str, from_none], obj.get("os")) return ImagePlatform(architecture, os) def to_dict(self) -> dict: result: dict = {} result["architecture"] = from_union([from_str, from_none], self.architecture) result["os"] = from_union([from_str, from_none], self.os) return clean(result) @dataclass class Ref: field: Optional[str] = None media_type: Optional[str] = None path: Optional[str] = None @staticmethod def from_dict(obj: Any) -> "Ref": assert isinstance(obj, dict) field = from_union([from_str, from_none], obj.get("field")) media_type = from_union([from_str, from_none], obj.get("mediaType")) path = from_union([from_str, from_none], obj.get("path")) return Ref(field, media_type, path) def to_dict(self) -> dict: result: dict = {} result["field"] = from_union([from_str, from_none], self.field) result["mediaType"] = from_union([from_str, from_none], self.media_type) result["path"] = from_union([from_str, from_none], self.path) return clean(result) @dataclass class Image: image: str description: Optional[str] = None digest: Optional[str] = None image_type: Optional[str] = None media_type: Optional[str] = None platform: Optional[ImagePlatform] = None refs: List[Ref] = field(default_factory=list) size: Optional[int] = None @staticmethod def from_dict(obj: Any) -> "Image": assert isinstance(obj, dict) description = from_union([from_str, from_none], obj.get("description")) digest = from_union([from_str, from_none], obj.get("digest")) image = from_str(obj.get("image")) image_type = from_union([from_str, from_none], obj.get("imageType")) media_type = from_union([from_str, from_none], obj.get("mediaType")) platform = from_union([ImagePlatform.from_dict, from_none], obj.get("platform")) refs = from_union( [lambda x: from_list(Ref.from_dict, x), from_none], obj.get("refs") ) size = from_union([from_int, from_none], obj.get("size")) return Image( description, digest, image, image_type, media_type, platform, refs, size ) def to_dict(self) -> dict: result: dict = {} result["description"] = from_union([from_str, from_none], self.description) result["digest"] = from_union([from_str, from_none], self.digest) result["image"] = from_str(self.image) result["imageType"] = from_union([from_str, from_none], self.image_type) result["mediaType"] = from_union([from_str, from_none], self.media_type) result["platform"] = from_union( [lambda x: to_class(ImagePlatform, x), from_none], self.platform ) result["refs"] = from_list(lambda x: to_class(Ref, x), self.refs) result["size"] = from_union([from_int, from_none], self.size) return clean(result) @dataclass class InvocationImage: image: str digest: Optional[str] = None image_type: Optional[str] = "oci" media_type: Optional[str] = None platform: Optional[ImagePlatform] = None size: Optional[str] = None @staticmethod def from_dict(obj: Any) -> "InvocationImage": assert isinstance(obj, dict) digest = from_union([from_str, from_none], obj.get("digest")) image = from_str(obj.get("image")) image_type = from_union([from_str, from_none], obj.get("imageType")) media_type = from_union([from_str, from_none], obj.get("mediaType")) platform = from_union([ImagePlatform.from_dict, from_none], obj.get("platform")) size = from_union([from_str, from_none], obj.get("size")) return InvocationImage(image, digest, image_type, media_type, platform, size) def to_dict(self) -> dict: result: dict = {} result["digest"] = from_union([from_str, from_none], self.digest) result["image"] = from_str(self.image) result["imageType"] = from_union([from_str, from_none], self.image_type) result["mediaType"] = from_union([from_str, from_none], self.media_type) result["platform"] = from_union( [lambda x: to_class(ImagePlatform, x), from_none], self.platform ) result["size"] = from_union([from_str, from_none], self.size) return clean(result) @dataclass class Maintainer: name: str email: Optional[str] = None url: Optional[str] = None @staticmethod def from_dict(obj: Any) -> "Maintainer": assert isinstance(obj, dict) name = from_union([from_str, from_none], obj.get("name")) email = from_union([from_str, from_none], obj.get("email")) url = from_union([from_str, from_none], obj.get("url")) return Maintainer(name, email, url) def to_dict(self) -> dict: result: dict = {} result["name"] = from_union([from_str, from_none], self.name) result["email"] = from_union([from_str, from_none], self.email) result["url"] = from_union([from_str, from_none], self.url) return clean(result) @dataclass class Destination: description: Optional[str] = None env: Optional[str] = None path: Optional[str] = None @staticmethod def from_dict(obj: Any) -> "Destination": assert isinstance(obj, dict) description = from_union([from_str, from_none], obj.get("description")) env = from_union([from_str, from_none], obj.get("env")) path = from_union([from_str, from_none], obj.get("path")) return Destination(description, env, path) def to_dict(self) -> dict: result: dict = {} result["description"] = from_union([from_str, from_none], self.description) result["env"] = from_union([from_str, from_none], self.env) result["path"] = from_union([from_str, from_none], self.path) return clean(result) @dataclass class Metadata: description: Optional[str] = None @staticmethod def from_dict(obj: Any) -> "Metadata": assert isinstance(obj, dict) description = from_union([from_str, from_none], obj.get("description")) return Metadata(description) def to_dict(self) -> dict: result: dict = {} result["description"] = from_union([from_str, from_none], self.description) return clean(result) @dataclass class Parameter: type: str destination: Destination default_value: Union[bool, int, None, str] = None allowed_values: Optional[List[Any]] = field(default_factory=list) max_length: Optional[int] = None max_value: Optional[int] = None metadata: Optional[Metadata] = None min_length: Optional[int] = None min_value: Optional[int] = None required: Optional[bool] = None @staticmethod def from_dict(obj: Any) -> "Parameter": assert isinstance(obj, dict) allowed_values = from_union( [lambda x: from_list(lambda x: x, x), from_none], obj.get("allowedValues") ) default_value = from_union( [from_int, from_bool, from_none, from_str], obj.get("defaultValue") ) destination = from_union([Destination.from_dict], obj.get("destination")) max_length = from_union([from_int, from_none], obj.get("maxLength")) max_value = from_union([from_int, from_none], obj.get("maxValue")) metadata = from_union([Metadata.from_dict, from_none], obj.get("metadata")) min_length = from_union([from_int, from_none], obj.get("minLength")) min_value = from_union([from_int, from_none], obj.get("minValue")) required = from_union([from_bool, from_none], obj.get("required")) type = from_str(obj.get("type")) return Parameter( type, destination, default_value, allowed_values, max_length, max_value, metadata, min_length, min_value, required, ) def to_dict(self) -> dict: result: dict = {} result["allowedValues"] = from_list(lambda x: x, self.allowed_values) result["destination"] = from_union( [lambda x: to_class(Destination, x)], self.destination ) result["maxLength"] = from_union([from_int, from_none], self.max_length) result["maxValue"] = from_union([from_int, from_none], self.max_value) result["metadata"] = from_union( [lambda x: to_class(Metadata, x), from_none], self.metadata ) result["minLength"] = from_union([from_int, from_none], self.min_length) result["minValue"] = from_union([from_int, from_none], self.min_value) result["required"] = from_union([from_bool, from_none], self.required) result["type"] = from_str(self.type) return clean(result) @dataclass class Bundle: name: str version: str invocation_images: List[InvocationImage] schema_version: Optional[str] = "v1" actions: Dict[str, Action] = field(default_factory=dict) credentials: Dict[str, Credential] = field(default_factory=dict) description: Optional[str] = None license: Optional[str] = None images: Dict[str, Image] = field(default_factory=dict) keywords: List[str] = field(default_factory=list) maintainers: List[Maintainer] = field(default_factory=list) parameters: Dict[str, Parameter] = field(default_factory=dict) @staticmethod def from_dict(obj: Any) -> "Bundle": assert isinstance(obj, dict) actions = from_union( [lambda x: from_dict(Action.from_dict, x), from_none], obj.get("actions") ) credentials = from_union( [lambda x: from_dict(Credential.from_dict, x), from_none], obj.get("credentials"), ) description = from_union([from_str, from_none], obj.get("description")) license = from_union([from_str, from_none], obj.get("license")) images = from_union( [lambda x: from_dict(Image.from_dict, x), from_none], obj.get("images") ) invocation_images = from_list( InvocationImage.from_dict, obj.get("invocationImages") ) keywords = from_union( [lambda x: from_list(from_str, x), from_none], obj.get("keywords") ) maintainers = from_union( [lambda x: from_list(Maintainer.from_dict, x), from_none], obj.get("maintainers"), ) name = from_str(obj.get("name")) parameters = from_union( [lambda x: from_dict(Parameter.from_dict, x), from_none], obj.get("parameters"), ) schema_version = from_union([from_str, from_none], obj.get("schemaVersion")) version = from_str(obj.get("version")) return Bundle( name, version, invocation_images, schema_version, actions, credentials, description, license, images, keywords, maintainers, parameters, ) def to_dict(self) -> dict: result: dict = {} result["actions"] = from_dict(lambda x: to_class(Action, x), self.actions) result["credentials"] = from_dict( lambda x: to_class(Credential, x), self.credentials ) result["description"] = from_union([from_str, from_none], self.description) result["license"] = from_union([from_str, from_none], self.license) result["images"] = from_dict(lambda x: to_class(Image, x), self.images) result["invocationImages"] = from_list( lambda x: to_class(InvocationImage, x), self.invocation_images ) result["keywords"] = from_list(from_str, self.keywords) result["maintainers"] = from_list( lambda x: to_class(Maintainer, x), self.maintainers ) result["name"] = from_str(self.name) result["parameters"] = from_dict( lambda x: to_class(Parameter, x), self.parameters ) result["schemaVersion"] = from_str(self.schema_version) result["version"] = from_str(self.version) return clean(result) def to_json(self, pretty: bool = False) -> str: if pretty: func = canonicaljson.encode_pretty_printed_json else: func = canonicaljson.encode_canonical_json return func(self.to_dict()).decode() PK!:J cnab/util.pyfrom typing import List from cnab.types import InvocationImage def extract_docker_images(images: List[InvocationImage]) -> list: return list(filter(lambda x: x.image_type == "docker", images)) PK!<%[[cnab-0.1.7.dist-info/LICENSE pycnab Copyright (C) 2018 Gareth Rushgrove Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at https://www.apache.org/licenses/LICENSE-2.0 Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License. PK!HnHTUcnab-0.1.7.dist-info/WHEEL A н#Z;/"d&F[xzw@Zpy3Fv]\fi4WZ^EgM_-]#0(q7PK!H鿼cnab-0.1.7.dist-info/METADATAWko6_qbǶEui2ؙLMfXELIF"UJ({.)A,}}^#SFgt'WZ(=NJnv5jI+c|EJ^=J:kZx6tYKGJWF'L#gXCY}| ]v,Tg(*9;I t7X'tZ[(l3(uOnϝ͢}}8M~:sj$PIRhݚښM!z!KԀQ0'oEF)tG6r`8N 甑IN/C &%ɲiQڟdgF)of2z!]aU)3Jwin$)=k9@ҹ{KT1(֟*3И&J"e6HE^oJIsȬA|"QE`~$):k@2O9KhΚRZ.X8s,ymrAl(pӦLU*hzS p₮?d7 5_:k!JOAi^%jGtQ7e|DMy2d]ȐBHD7o=lu< ښrvy\]oΡvӪMlJPQ4W4le2JZ$y"t-rr*Z ]l%N#bo]hD*hɊKU%{UA ϔTJr :ar(FXOlIa?|I̴RG{G,!|x"Ik# 8ҭc&|tuN~M &Ĺs2=Zmgш.6J|V CGPl%Mr.jgWmRlaEnz{ @L!K-:PkO\;}3B|(t%0XB 0;8[Ƌ4vŨ0v4 'qXx8w >]83EE`{qf%beQiL[7{eo) _f"~4ݣ ZDwћ{x#(Lw~\rLg}3觀vDTZ,]ibfm,lFT߹ @00&7]7Fk $mh>1#5 nWVȏBLTI;cN;^8&>RzEG4zPr+GaJ;{VfWx+خCyd;%Bs/5sRi2mAߝy2v i׸EQlQBe3%g(" $CVW( rsJ_V ~r2L1XԵ(|C[Ӽa&IR9❯ϞnZ|'h瞤i3@Ҧq!D \mfaŖ) 99;guMFUhQv9}1,x\nӸKp U> 6Dcnab/types.pyPK!:J fcnab/util.pyPK!<%[[Xcnab-0.1.7.dist-info/LICENSEPK!HnHTUcnab-0.1.7.dist-info/WHEELPK!H鿼ycnab-0.1.7.dist-info/METADATAPK!Hv{eYcnab-0.1.7.dist-info/RECORDPK b