Repository: awslabs/aws-ddk (language: python)

Issue:
implement `ScheduledEventRule` construct
native scheduled event rule that replaces and event based route



Discussion / hints:

Fix patch:
diff --git a/API.md b/API.md
index b33499f0..55ef4992 100644
--- a/API.md
+++ b/API.md
@@ -3637,6 +3637,7 @@ const addRuleProps: AddRuleProps = { ... }
 | <code><a href="#aws-ddk-core.AddRuleProps.property.id">id</a></code> | <code>string</code> | *No description.* |
 | <code><a href="#aws-ddk-core.AddRuleProps.property.overrideRule">overrideRule</a></code> | <code>aws-cdk-lib.aws_events.IRule</code> | *No description.* |
 | <code><a href="#aws-ddk-core.AddRuleProps.property.ruleName">ruleName</a></code> | <code>string</code> | *No description.* |
+| <code><a href="#aws-ddk-core.AddRuleProps.property.schedule">schedule</a></code> | <code>aws-cdk-lib.aws_events.Schedule</code> | *No description.* |
 
 ---
 
@@ -3690,6 +3691,16 @@ public readonly ruleName: string;
 
 ---
 
+##### `schedule`<sup>Optional</sup> <a name="schedule" id="aws-ddk-core.AddRuleProps.property.schedule"></a>
+
+```typescript
+public readonly schedule: Schedule;
+```
+
+- *Type:* aws-cdk-lib.aws_events.Schedule
+
+---
+
 ### AddSecurityLintStageProps <a name="AddSecurityLintStageProps" id="aws-ddk-core.AddSecurityLintStageProps"></a>
 
 #### Initializer <a name="Initializer" id="aws-ddk-core.AddSecurityLintStageProps.Initializer"></a>
@@ -3746,6 +3757,7 @@ const addStageProps: AddStageProps = { ... }
 | <code><a href="#aws-ddk-core.AddStageProps.property.stage">stage</a></code> | <code><a href="#aws-ddk-core.Stage">Stage</a></code> | *No description.* |
 | <code><a href="#aws-ddk-core.AddStageProps.property.overrideRule">overrideRule</a></code> | <code>aws-cdk-lib.aws_events.IRule</code> | *No description.* |
 | <code><a href="#aws-ddk-core.AddStageProps.property.ruleName">ruleName</a></code> | <code>string</code> | *No description.* |
+| <code><a href="#aws-ddk-core.AddStageProps.property.schedule">schedule</a></code> | <code>aws-cdk-lib.aws_events.Schedule</code> | *No description.* |
 | <code><a href="#aws-ddk-core.AddStageProps.property.skipRule">skipRule</a></code> | <code>boolean</code> | *No description.* |
 
 ---
@@ -3780,6 +3792,16 @@ public readonly ruleName: string;
 
 ---
 
+##### `schedule`<sup>Optional</sup> <a name="schedule" id="aws-ddk-core.AddStageProps.property.schedule"></a>
+
+```typescript
+public readonly schedule: Schedule;
+```
+
+- *Type:* aws-cdk-lib.aws_events.Schedule
+
+---
+
 ##### `skipRule`<sup>Optional</sup> <a name="skipRule" id="aws-ddk-core.AddStageProps.property.skipRule"></a>
 
 ```typescript
diff --git a/src/pipelines/pipelines.ts b/src/pipelines/pipelines.ts
index c430f6ea..1eb4e21f 100644
--- a/src/pipelines/pipelines.ts
+++ b/src/pipelines/pipelines.ts
@@ -14,6 +14,7 @@ export interface AddStageProps {
   readonly skipRule?: boolean;
   readonly overrideRule?: events.IRule;
   readonly ruleName?: string;
+  readonly schedule?: events.Schedule;
 }
 
 export interface AddRuleProps {
@@ -22,6 +23,7 @@ export interface AddRuleProps {
   readonly eventTargets?: events.IRuleTarget[];
   readonly overrideRule?: events.IRule;
   readonly ruleName?: string;
+  readonly schedule?: events.Schedule;
 }
 
 export class DataPipeline extends Construct {
@@ -46,7 +48,7 @@ export class DataPipeline extends Construct {
     const skipRule = props.skipRule ?? false;
 
     if (props.overrideRule) {
-      this.addRule({ overrideRule: props.overrideRule });
+      this.addRule({ overrideRule: props.overrideRule, schedule: props.schedule });
     } else if (this.previousStage && skipRule === false) {
       if (stage.targets === undefined) {
         throw new Error(
@@ -58,6 +60,7 @@ export class DataPipeline extends Construct {
         id: `${stage.node.id} Rule`,
         eventPattern: this.previousStage?.eventPattern,
         eventTargets: stage.targets,
+        schedule: props.schedule,
       });
     }
 
@@ -76,6 +79,12 @@ export class DataPipeline extends Construct {
     this.rules.push(
       props.overrideRule
         ? props.overrideRule
+        : props.schedule
+        ? new events.Rule(this, props.id!, {
+            schedule: props.schedule,
+            targets: props.eventTargets,
+            ruleName: props.ruleName,
+          })
         : new events.Rule(this, props.id!, {
             eventPattern: props.eventPattern,
             targets: props.eventTargets,


Test patch:
diff --git a/test/data-pipelines.test.ts b/test/data-pipelines.test.ts
index 8bf2ca92..e43c512f 100644
--- a/test/data-pipelines.test.ts
+++ b/test/data-pipelines.test.ts
@@ -1,10 +1,12 @@
 import path from "path";
+import * as glue_alpha from "@aws-cdk/aws-glue-alpha";
 import * as cdk from "aws-cdk-lib";
 import { Match, Template } from "aws-cdk-lib/assertions";
+import * as events from "aws-cdk-lib/aws-events";
 import * as lambda from "aws-cdk-lib/aws-lambda";
 import * as s3 from "aws-cdk-lib/aws-s3";
 
-import { DataPipeline, FirehoseToS3Stage, S3EventStage, SqsToLambdaStage } from "../src";
+import { DataPipeline, FirehoseToS3Stage, GlueTransformStage, S3EventStage, SqsToLambdaStage } from "../src";
 
 test("Basic DataPipeline", () => {
   const stack = new cdk.Stack();
@@ -74,3 +76,46 @@ test("DataPipeline cannot have a Stage without targets in the middle", () => {
     pipeline.addStage({ stage: s3EventStage });
   }).toThrowError();
 });
+
+test("DataPipeline with Scheduled stage", () => {
+  const stack = new cdk.Stack();
+  const bucket = new s3.Bucket(stack, "Bucket");
+
+  const firehoseToS3Stage = new FirehoseToS3Stage(stack, "Firehose To S3 Stage", { s3Bucket: bucket });
+
+  const sqsToLambdaStage = new SqsToLambdaStage(stack, "SQS To Lambda Stage 2", {
+    lambdaFunctionProps: {
+      code: lambda.Code.fromAsset(path.join(__dirname, "/../src/")),
+      handler: "commons.handlers.lambda_handler",
+      memorySize: cdk.Size.mebibytes(512),
+      layers: [
+        lambda.LayerVersion.fromLayerVersionArn(stack, "Layer", "arn:aws:lambda:us-east-1:222222222222:layer:dummy:1"),
+      ],
+    },
+  });
+
+  const glueTranformStage = new GlueTransformStage(stack, "glue-transform", {
+    jobProps: {
+      executable: glue_alpha.JobExecutable.pythonEtl({
+        glueVersion: glue_alpha.GlueVersion.V3_0,
+        script: glue_alpha.Code.fromBucket(s3.Bucket.fromBucketName(stack, "bucket", "my-bucket"), "my-script"),
+        pythonVersion: glue_alpha.PythonVersion.THREE,
+      }),
+      jobName: "myJob",
+    },
+    crawlerName: "myCrawler",
+  });
+  const pipeline = new DataPipeline(stack, "Pipeline", {});
+  pipeline
+    .addStage({ stage: firehoseToS3Stage })
+    .addStage({ stage: sqsToLambdaStage })
+    .addStage({ stage: glueTranformStage, schedule: events.Schedule.rate(cdk.Duration.minutes(5)) });
+
+  const template = Template.fromStack(stack);
+
+  template.resourceCountIs("AWS::Events::Rule", 2);
+  template.hasResourceProperties("AWS::Events::Rule", {
+    State: "ENABLED",
+    ScheduleExpression: "rate(5 minutes)",
+  });
+});