Source code for

#  -*- coding: utf-8 -*-
# SPDX-License-Identifier: MPL-2.0
# Copyright 2020-2021 John Mille <>

Module to define the entry point for AWS Event Rules
import warnings

from ecs_composex.common import LOG, NONALPHANUM, build_template
from ecs_composex.common.compose_resources import (
from ecs_composex.common.stacks import ComposeXStack
from ecs_composex.ecs.ecs_params import CLUSTER_NAME, FARGATE_VERSION
from import MOD_KEY, RES_KEY
from import create_events_template

[docs]def validate_service_definition(service): required_keys = ["name", "TaskCount"] if not set(required_keys).issubset(service): raise KeyError( "Services definition must contain at least", required_keys, "Got", service.keys(), )
[docs]class Rule(XResource): """ Class to define an Event Rule """
[docs] def handle_families_targets_expansion(self, service, settings): the_service = [s for s in if == service["name"]][0] for family_name in the_service.families: family_name = NONALPHANUM.sub("", family_name) if family_name not in [f[0].name for f in self.families_targets]: self.families_targets.append( ( settings.families[family_name], False, [the_service], service["TaskCount"], service, ) )
[docs] def set_services_targets(self, settings): """ Override method to map services and families targets of the services defined specifically for events TargetStructure: (family, family_wide, services[], access) :param ecs_composex.common.settings.ComposeXSettings settings: :return: """ if not"No services defined for {}") return for service in validate_service_definition(service) service_name = service["name"] if service_name in settings.families and service_name not in [ f[0].name for f in self.families_targets ]: self.families_targets.append( ( settings.families[service_name], True, settings.families[service_name].services, service["TaskCount"], service, ) ) elif service_name in settings.families and service_name in [ f[0].name for f in self.families_targets ]: LOG.warning( f"The family {service_name} has already been added. Skipping" ) elif service_name in [ for s in]: self.handle_families_targets_expansion(service, settings)
[docs]class XStack(ComposeXStack): """ Class to handle events stack """ def __init__(self, title, settings, **kwargs): """ Method to initialize the XStack for Events :param str title: title for the stack :param ecs_composex.common.settings.ComposeXSettings settings: Execution settings :param dict kwargs: """ set_resources(settings, Rule, RES_KEY, MOD_KEY) x_resources = settings.compose_content[RES_KEY].values() new_resources = set_new_resources(x_resources, RES_KEY, False) lookup_resources = set_lookup_resources(x_resources, RES_KEY) use_resources = set_use_resources(x_resources, RES_KEY, False) if new_resources or use_resources: stack_template = build_template( "Events rules for ComposeX", [CLUSTER_NAME, FARGATE_VERSION], ) super().__init__(title, stack_template, **kwargs) create_events_template(self, settings, new_resources) else: self.is_void = True if lookup_resources or use_resources: warnings.warn( f"{RES_KEY} does not support Lookup. You can only create new resources" )