def __init__()

in cdk/stacks/sample_kinesis_stream_producer/sample_kinesis_stream_producer_stack.py [0:0]

• 14 lines of code
• 1 McCabe index (conditional complexity)

    def __init__(self, scope: core.Construct, construct_id: str, stream: aws_kinesis.IStream, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        sample_device_producer = aws_lambda_python.PythonFunction(self, 'SampleDeviceProducer',
                                                                  entry='stacks/sample_kinesis_stream_producer/producer_lambda',
                                                                  index='app.py',
                                                                  runtime=aws_lambda.Runtime.PYTHON_3_8,
                                                                  timeout=core.Duration.seconds(30))

        stream.grant_write(sample_device_producer)

        lambda_input = {"Stream": stream.stream_name}
        Rule(self, 'ProducerTriggerEventRule',
             enabled=True,
             schedule=Schedule.rate(Duration.minutes(1)),
             targets=[aws_events_targets.LambdaFunction(handler=sample_device_producer,
                                                        event=RuleTargetInput.from_object(lambda_input))])