Source code for runway.cfngin.hooks.ecs

"""AWS ECS hook."""
# pylint: disable=no-self-argument
from __future__ import annotations

import logging
from typing import TYPE_CHECKING, Any, Dict, List, Union

from pydantic import validator
from typing_extensions import TypedDict

from ...utils import BaseModel

    from mypy_boto3_ecs.type_defs import CreateClusterResponseTypeDef

    from ...context import CfnginContext

LOGGER = logging.getLogger(__name__)

[docs]class CreateClustersHookArgs(BaseModel): """Hook arguments for ``create_clusters``.""" clusters: List[str] """List of cluster names to create.""" @validator("clusters", allow_reuse=True, pre=True) def _convert_clusters(cls, v: Union[List[str], str]) -> List[str]: """Convert value of ``clusters`` from str to list.""" if isinstance(v, str): return [v] return v
[docs]class CreateClustersResponseTypeDef(TypedDict): """Response from create_clusters.""" clusters: Dict[str, CreateClusterResponseTypeDef]
[docs]def create_clusters( context: CfnginContext, *__args: Any, **kwargs: Any ) -> CreateClustersResponseTypeDef: """Create ECS clusters. Args: context: CFNgin context object. """ args = CreateClustersHookArgs.parse_obj(kwargs) ecs_client = context.get_session().client("ecs") cluster_info: Dict[str, Any] = {} for cluster in args.clusters: LOGGER.debug("creating ECS cluster: %s", cluster) response = ecs_client.create_cluster(clusterName=cluster) cluster_info[response.get("cluster", {}).get("clusterName", "")] = response return {"clusters": cluster_info}