def upload_assets()

in infra/upload_assets.py [0:0]

• 18 lines of code
• 4 McCabe index (conditional complexity)

def upload_assets(cdk_dir: str = "cdk.out") -> None:
    """Parses the asset files in cdk directory and uploads resources to S3

    Args:
        cdk_dir (str): The cdk directory
    """
    for asset_path in glob.glob(f"{cdk_dir}/*.assets.json"):
        logger.debug(f"Processing asset: {asset_path}")
        with open(asset_path, "r") as f:
            asset = json.load(f)
        for key in asset["files"]:
            meta = asset["files"][key]
            # Get source info
            src = meta["source"]
            file_path = os.path.join(cdk_dir, src["path"])
            content_type = "application/json"
            if src["packaging"] == "zip":
                logger.info(f"Packaging zip: {file_path}")
                file_path = make_zipfile(file_path)
                content_type = "application/zip"
            # Get the destination
            dest = meta["destinations"]["current_account-current_region"]
            bucket_name = dest["bucketName"]
            object_key = dest["objectKey"]
            # Upload file to s3
            upload_file(file_path, bucket_name, object_key, content_type)