Merge branch 'morosi-feat' into 'master'

Add AssetLayout enum

See merge request it/e3-aws!90

Author: adanaja

src/e3/aws/troposphere/asset.py

@@ -3,6 +3,7 @@
 from functools import cached_property
 from hashlib import sha256
 from pathlib import Path
+from enum import Enum, auto
 from typing import TYPE_CHECKING
 
 from e3.aws.troposphere import Asset
@@ -12,6 +13,15 @@
     from collections.abc import Sequence
 
 
+class AssetLayout(Enum):
+    """Available layouts when uploading assets to S3."""
+
+    TREE = auto()
+    """Tree structure where S3 objects are put under assets/AssetName/."""
+    FLAT = auto()
+    """Flat structure where S3 objects are put under assets/."""
+
+
 class DirectoryAsset(Asset):
     """General purpose to create directory asset."""
 
@@ -22,6 +32,7 @@ def __init__(
         data_dir: str,
         ignore: str | Sequence[str] | None = None,
         versioning: bool = True,
+        layout: AssetLayout = AssetLayout.TREE,
     ) -> None:
         """Initialize DirectoryAsset.
 
@@ -30,11 +41,13 @@ def __init__(
         :param ignore: glob pattern or list of files or directories to ignore
         :param versioning: if True, elements are not uploaded if no change has been
           detected
+        :param layout: the layout for this asset
         """
         super().__init__(name)
         self.data_dir = data_dir
         self.ignore = ignore
         self.versioning = versioning
+        self.layout = layout
 
     @cached_property
     def _cached_ignored(self) -> list[Path]:
@@ -73,7 +86,11 @@ def asset_name(self) -> str:
     @property
     def s3_key(self) -> str:
         """Return a unique S3 key with the checksum of the folder."""
-        return f"{self.name}/{self.asset_name}"
+        return (
+            f"{self.name}/{self.asset_name}"
+            if self.layout == AssetLayout.TREE
+            else self.asset_name
+        )
 
     def upload(
         self,
@@ -105,17 +122,26 @@ def upload(
 class FileAsset(Asset):
     """General purpose to create File asset."""
 
-    def __init__(self, name: str, *, file_path: str, versioning: bool = True) -> None:
+    def __init__(
+        self,
+        name: str,
+        *,
+        file_path: str,
+        versioning: bool = True,
+        layout: AssetLayout = AssetLayout.TREE,
+    ) -> None:
         """Initialize FileAsset.
 
         :param name: name of the asset
         :param file_path: file path to add to the asset
         :param versioning: if True, the file is not uploaded if no change has been
           detected
+        :param layout: the layout for this asset
         """
         super().__init__(name)
         self.file_path = file_path
         self.versioning = versioning
+        self.layout = layout
 
     @cached_property
     def checksum(self) -> str:
@@ -135,7 +161,11 @@ def asset_name(self) -> str:
     @property
     def s3_key(self) -> str:
         """Return a unique S3 key with the checksum of the file."""
-        return f"{self.name}/{self.asset_name}"
+        return (
+            f"{self.name}/{self.asset_name}"
+            if self.layout == AssetLayout.TREE
+            else self.asset_name
+        )
 
     def upload(
         self,

src/e3/aws/troposphere/awslambda/__init__.py

@@ -22,6 +22,7 @@
 from e3.aws import name_to_id
 from e3.aws.cfn import client
 from e3.aws.troposphere import Construct, Asset
+from e3.aws.troposphere.asset import AssetLayout
 from e3.aws.troposphere.iam.policy_document import PolicyDocument
 from e3.aws.troposphere.iam.policy_statement import PolicyStatement
 from e3.aws.troposphere.iam.role import Role
@@ -104,18 +105,21 @@ def __init__(
         code_dir: str,
         runtime: str,
         requirement_file: str | None = None,
+        layout: AssetLayout = AssetLayout.TREE,
     ) -> None:
         """Initialize PyFunctionAsset.
 
         :param name: name of the archive
         :param code_dir: directory that contains the Python code
         :param runtime: the Python runtime
         :param requirement_file: the list of Python dependencies
+        :param layout: the layout for this asset
         """
         super().__init__(name)
         self.code_dir = code_dir
         self.runtime = runtime
         self.requirement_file = requirement_file
+        self.layout = layout
 
         # Temporary directory where the archive is created
         self._archive_tmpd: TemporaryDirectory | None = None
@@ -208,7 +212,11 @@ def archive_name(self) -> str:
     @cached_property
     def s3_key(self) -> str:
         """Return a unique S3 key with the checksum of the package."""
-        return f"{self.name}/{self.archive_name}"
+        return (
+            f"{self.name}/{self.archive_name}"
+            if self.layout == AssetLayout.TREE
+            else self.archive_name
+        )
 
     def populate_package_dir(self, package_dir: str) -> None:
         """Copy user code into package directory.