Main api

awspub/__init__.py

@@ -0,0 +1,3 @@
+from awspub.api import cleanup, create, list, publish, verify
+
+__all__ = ["create", "list", "publish", "cleanup", "verify"]

awspub/api.py

@@ -0,0 +1,165 @@
+import logging
+import pathlib
+from typing import Dict, Iterator, List, Optional, Tuple
+
+from awspub.context import Context
+from awspub.image import Image, _ImageInfo
+from awspub.s3 import S3
+
+logger = logging.getLogger(__name__)
+
+
+def _images_grouped(
+    images: List[Tuple[str, Image, Dict[str, _ImageInfo]]], group: Optional[str]
+) -> Tuple[Dict[str, Dict[str, str]], Dict[str, Dict[str, Dict[str, str]]]]:
+    """
+    Group the given images by name and by group
+
+    :param images: the images
+    :type images: List[Tuple[str, Image, Dict[str, _ImageInfo]]]
+    :param group: a optional group name
+    :type group: Optional[str]
+    :return: the images grouped by name and by group
+    :rtype: Tuple[Dict[str, Dict[str, str]], Dict[str, Dict[str, Dict[str, str]]]
+    """
+    images_by_name: Dict[str, Dict[str, str]] = dict()
+    images_by_group: Dict[str, Dict[str, Dict[str, str]]] = dict()
+    for image_name, image, image_result in images:
+        images_region_id: Dict[str, str] = {key: val.image_id for (key, val) in image_result.items()}
+        images_by_name[image_name] = images_region_id
+        for image_group in image.conf.get("groups", []):
+            if group and image_group != group:
+                continue
+            if not images_by_group.get(image_group):
+                images_by_group[image_group] = {}
+            images_by_group[image_group][image_name] = images_region_id
+    return images_by_name, images_by_group
+
+
+def _images_filtered(context: Context, group: Optional[str]) -> Iterator[Tuple[str, Image]]:
+    """
+    Filter the images from ctx based on the given args
+
+    :param context: the context
+    :type context: a awspub.context.Context instance
+    :param group: a optional group name
+    :type group: Optional[str]
+    """
+    for image_name in context.conf["images"].keys():
+        image = Image(context, image_name)
+        if group:
+            # limit the images to process to the group given on the command line
+            if group not in image.conf.get("groups", []):
+                logger.info(f"skipping image {image_name} because not part of group {group}")
+                continue
+
+        logger.info(f"processing image {image_name} (group: {group})")
+        yield image_name, image
+
+
+def create(
+    config: pathlib.Path, config_mapping: pathlib.Path, group: Optional[str]
+) -> Tuple[Dict[str, Dict[str, str]], Dict[str, Dict[str, Dict[str, str]]]]:
+    """
+    Create images in the partition of the used account based on
+    the given configuration file and the config mapping
+
+    :param config: the configuration file path
+    :type config: pathlib.Path
+    :param config_mapping: the config template mapping file path
+    :type config_mapping: pathlib.Path
+    :param group: only handles images from given group
+    :type group: Optional[str]
+    :return: the images grouped by name and by group
+    :rtype: Tuple[Dict[str, Dict[str, str]], Dict[str, Dict[str, Dict[str, str]]]
+    """
+
+    ctx = Context(config, config_mapping)
+    s3 = S3(ctx)
+    s3.upload_file(ctx.conf["source"]["path"])
+    images: List[Tuple[str, Image, Dict[str, _ImageInfo]]] = []
+    for image_name, image in _images_filtered(ctx, group):
+        image_result: Dict[str, _ImageInfo] = image.create()
+        images.append((image_name, image, image_result))
+    images_by_name, images_by_group = _images_grouped(images, group)
+    return images_by_name, images_by_group
+
+
+def list(
+    config: pathlib.Path, config_mapping: pathlib.Path, group: Optional[str]
+) -> Tuple[Dict[str, Dict[str, str]], Dict[str, Dict[str, Dict[str, str]]]]:
+    """
+    List images in the partition of the used account based on
+    the given configuration file and the config mapping
+
+    :param config: the configuration file path
+    :type config: pathlib.Path
+    :param config_mapping: the config template mapping file path
+    :type config_mapping: pathlib.Path
+    :param group: only handles images from given group
+    :type group: Optional[str]
+    :return: the images grouped by name and by group
+    :rtype: Tuple[Dict[str, Dict[str, str]], Dict[str, Dict[str, Dict[str, str]]]
+    """
+    ctx = Context(config, config_mapping)
+    images: List[Tuple[str, Image, Dict[str, _ImageInfo]]] = []
+    for image_name, image in _images_filtered(ctx, group):
+        image_result: Dict[str, _ImageInfo] = image.list()
+        images.append((image_name, image, image_result))
+
+    images_by_name, images_by_group = _images_grouped(images, group)
+    return images_by_name, images_by_group
+
+
+def publish(config: pathlib.Path, config_mapping: pathlib.Path, group: Optional[str]):
+    """
+    Make available images in the partition of the used account based on
+    the given configuration file public
+
+    :param config: the configuration file path
+    :type config: pathlib.Path
+    :param config_mapping: the config template mapping file path
+    :type config_mapping: pathlib.Path
+    :param group: only handles images from given group
+    :type group: Optional[str]
+    """
+    ctx = Context(config, config_mapping)
+    for image_name, image in _images_filtered(ctx, group):
+        image.publish()
+
+
+def cleanup(config: pathlib.Path, config_mapping: pathlib.Path, group: Optional[str]):
+    """
+    Cleanup available images in the partition of the used account based on
+    the given configuration file
+
+    :param config: the configuration file path
+    :type config: pathlib.Path
+    :param config_mapping: the config template mapping file path
+    :type config_mapping: pathlib.Path
+    :param group: only handles images from given group
+    :type group: Optional[str]
+    """
+    ctx = Context(config, config_mapping)
+    for image_name, image in _images_filtered(ctx, group):
+        image.cleanup()
+
+
+def verify(config: pathlib.Path, config_mapping: pathlib.Path, group: Optional[str]) -> Dict[str, Dict]:
+    """
+    Verify available images in the partition of the used account based on
+    the given configuration file.
+    This is EXPERIMENTAL and doesn't work reliable yet!
+
+    :param config: the configuration file path
+    :type config: pathlib.Path
+    :param config_mapping: the config template mapping file path
+    :type config_mapping: pathlib.Path
+    :param group: only handles images from given group
+    :type group: Optional[str]
+    """
+    problems: Dict[str, Dict] = dict()
+    ctx = Context(config, config_mapping)
+    for image_name, image in _images_filtered(ctx, group):
+        problems[image_name] = image.verify()
+    return problems

awspub/cli/__init__.py

@@ -5,111 +5,52 @@
 import logging
 import pathlib
 import sys
-from typing import Dict, Iterator, List, Optional, Tuple
 
-from awspub.context import Context
-from awspub.image import Image, _ImageInfo
-from awspub.s3 import S3
+import awspub
 
 logger = logging.getLogger(__name__)
 
 
-def _images_filtered(context: Context, group: Optional[str]) -> Iterator[Tuple[str, Image]]:
-    """
-    Filter the images from ctx based on the given args
-    :param context: the context
-    :type context: a awspub.context.Context instance
-    :param group: a optional group name
-    :type group: Optional[str]
-    """
-    for image_name in context.conf["images"].keys():
-        image = Image(context, image_name)
-        if group:
-            # limit the images to process to the group given on the command line
-            if group not in image.conf.get("groups", []):
-                logger.info(f"skipping image {image_name} because not part of group {group}")
-                continue
-
-        logger.info(f"processing image {image_name} (group: {group})")
-        yield image_name, image
-
-
-def _images_json(images: List[Tuple[str, Image, Dict[str, _ImageInfo]]], group: Optional[str]):
-    """
-    Return json data which is the output for eg. the create and list commands
-    That data has images listed by name but also images grouped by the group
-    """
-    images_by_name: Dict[str, Dict[str, str]] = dict()
-    images_by_group: Dict[str, Dict[str, Dict[str, str]]] = dict()
-    for image_name, image, image_result in images:
-        images_region_id: Dict[str, str] = {key: val.image_id for (key, val) in image_result.items()}
-        images_by_name[image_name] = images_region_id
-        for image_group in image.conf.get("groups", []):
-            if group and image_group != group:
-                continue
-            if not images_by_group.get(image_group):
-                images_by_group[image_group] = {}
-            images_by_group[image_group][image_name] = images_region_id
-    return json.dumps({"images": images_by_name, "images-by-group": images_by_group}, indent=4)
-
-
 def _create(args) -> None:
     """
     Create images based on the given configuration and write json
     data to the given output
     """
-    ctx = Context(args.config, args.config_mapping)
-    s3 = S3(ctx)
-    s3.upload_file(ctx.conf["source"]["path"])
-    images: List[Tuple[str, Image, Dict[str, _ImageInfo]]] = []
-    for image_name, image in _images_filtered(ctx, args.group):
-        image_result: Dict[str, _ImageInfo] = image.create()
-        images.append((image_name, image, image_result))
-
-    args.output.write((_images_json(images, args.group)))
+    images_by_name, images_by_group = awspub.create(args.config, args.config_mapping, args.group)
+    images_json = json.dumps({"images": images_by_name, "images-by-group": images_by_group}, indent=4)
+    args.output.write(images_json)
 
 
 def _list(args) -> None:
     """
     List images based on the given configuration and write json
     data to the given output
     """
-    ctx = Context(args.config, args.config_mapping)
-    images: List[Tuple[str, Image, Dict[str, _ImageInfo]]] = []
-    for image_name, image in _images_filtered(ctx, args.group):
-        image_result: Dict[str, _ImageInfo] = image.list()
-        images.append((image_name, image, image_result))
-
-    args.output.write((_images_json(images, args.group)))
+    images_by_name, images_by_group = awspub.list(args.config, args.config_mapping, args.group)
+    images_json = json.dumps({"images": images_by_name, "images-by-group": images_by_group}, indent=4)
+    args.output.write(images_json)
 
 
 def _verify(args) -> None:
     """
     Verify available images against configuration
     """
-    problems: Dict[str, Dict] = dict()
-    ctx = Context(args.config, args.config_mapping)
-    for image_name, image in _images_filtered(ctx, args.group):
-        problems[image_name] = image.verify()
+    problems = awspub.verify(args.config, args.config_mapping, args.group)
     args.output.write((json.dumps({"problems": problems}, indent=4)))
 
 
 def _cleanup(args) -> None:
     """
     Cleanup available images
     """
-    ctx = Context(args.config, args.config_mapping)
-    for image_name, image in _images_filtered(ctx, args.group):
-        image.cleanup()
+    awspub.cleanup(args.config, args.config_mapping, args.group)
 
 
-def _public(args) -> None:
+def _publish(args) -> None:
     """
     Make available images public
     """
-    ctx = Context(args.config, args.config_mapping)
-    for image_name, image in _images_filtered(ctx, args.group):
-        image.public()
+    awspub.publish(args.config, args.config_mapping, args.group)
 
 
 def _parser():
@@ -161,16 +102,16 @@ def _parser():
 
     p_cleanup.set_defaults(func=_cleanup)
 
-    # public
-    p_public = p_sub.add_parser("public", help="Publish images")
-    p_public.add_argument(
+    # publish
+    p_publish = p_sub.add_parser("publish", help="Publish images")
+    p_publish.add_argument(
         "--output", type=argparse.FileType("w+"), help="output file path. defaults to stdout", default=sys.stdout
     )
-    p_public.add_argument("--config-mapping", type=pathlib.Path, help="the image config template mapping file path")
-    p_public.add_argument("--group", type=str, help="only handles images from given group")
-    p_public.add_argument("config", type=pathlib.Path, help="the image configuration file path")
+    p_publish.add_argument("--config-mapping", type=pathlib.Path, help="the image config template mapping file path")
+    p_publish.add_argument("--group", type=str, help="only handles images from given group")
+    p_publish.add_argument("config", type=pathlib.Path, help="the image configuration file path")
 
-    p_public.set_defaults(func=_public)
+    p_publish.set_defaults(func=_publish)
 
     return parser
 

awspub/image.py

@@ -474,7 +474,7 @@ def create(self) -> Dict[str, _ImageInfo]:
 
         return images
 
-    def public(self) -> None:
+    def publish(self) -> None:
         """
         Handle all publication steps
         - make image and underlying root device snapshot public if the public flag is set