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

Issue:
(core) Airflow (MWWA) data pipeline stage
**Is your feature request related to a problem? Please describe.**
DDK Airflow integration

**Describe the solution you'd like**
I'd like to be able to kickoff MWWA DAG's in DDK DataPipelines. A stage should kickoff the DAG and wait for it to complete.

**Describe alternatives you've considered**
N/A
**Additional context**
N/A

*P.S. Please Don't attach files. Add code snippets directly in the message body instead.*


Discussion / hints:
Marking this issue as stale due to inactivity. This helps our maintainers find and focus on the active issues. If this issue receives no comments in the next 7 days it will automatically be closed.
I would like to bump this up as we have a feature request for a customer to add this as a data pipeline stage

Fix patch:
diff --git a/API.md b/API.md
index 658b3f44..633568ac 100644
--- a/API.md
+++ b/API.md
@@ -100,6 +100,240 @@ This project is licensed under the Apache-2.0 License.
 
 ## Constructs <a name="Constructs" id="Constructs"></a>
 
+### AirflowDataPipeline <a name="AirflowDataPipeline" id="aws-ddk-core.AirflowDataPipeline"></a>
+
+#### Initializers <a name="Initializers" id="aws-ddk-core.AirflowDataPipeline.Initializer"></a>
+
+```typescript
+import { AirflowDataPipeline } from 'aws-ddk-core'
+
+new AirflowDataPipeline(scope: Construct, id: string, props: AirflowPipelineProps)
+```
+
+| **Name** | **Type** | **Description** |
+| --- | --- | --- |
+| <code><a href="#aws-ddk-core.AirflowDataPipeline.Initializer.parameter.scope">scope</a></code> | <code>constructs.Construct</code> | *No description.* |
+| <code><a href="#aws-ddk-core.AirflowDataPipeline.Initializer.parameter.id">id</a></code> | <code>string</code> | *No description.* |
+| <code><a href="#aws-ddk-core.AirflowDataPipeline.Initializer.parameter.props">props</a></code> | <code><a href="#aws-ddk-core.AirflowPipelineProps">AirflowPipelineProps</a></code> | *No description.* |
+
+---
+
+##### `scope`<sup>Required</sup> <a name="scope" id="aws-ddk-core.AirflowDataPipeline.Initializer.parameter.scope"></a>
+
+- *Type:* constructs.Construct
+
+---
+
+##### `id`<sup>Required</sup> <a name="id" id="aws-ddk-core.AirflowDataPipeline.Initializer.parameter.id"></a>
+
+- *Type:* string
+
+---
+
+##### `props`<sup>Required</sup> <a name="props" id="aws-ddk-core.AirflowDataPipeline.Initializer.parameter.props"></a>
+
+- *Type:* <a href="#aws-ddk-core.AirflowPipelineProps">AirflowPipelineProps</a>
+
+---
+
+#### Methods <a name="Methods" id="Methods"></a>
+
+| **Name** | **Description** |
+| --- | --- |
+| <code><a href="#aws-ddk-core.AirflowDataPipeline.toString">toString</a></code> | Returns a string representation of this construct. |
+| <code><a href="#aws-ddk-core.AirflowDataPipeline.createVpc">createVpc</a></code> | *No description.* |
+
+---
+
+##### `toString` <a name="toString" id="aws-ddk-core.AirflowDataPipeline.toString"></a>
+
+```typescript
+public toString(): string
+```
+
+Returns a string representation of this construct.
+
+##### `createVpc` <a name="createVpc" id="aws-ddk-core.AirflowDataPipeline.createVpc"></a>
+
+```typescript
+public createVpc(scope: Construct, environmentName: string, vpcCidr: string): IVpc
+```
+
+###### `scope`<sup>Required</sup> <a name="scope" id="aws-ddk-core.AirflowDataPipeline.createVpc.parameter.scope"></a>
+
+- *Type:* constructs.Construct
+
+---
+
+###### `environmentName`<sup>Required</sup> <a name="environmentName" id="aws-ddk-core.AirflowDataPipeline.createVpc.parameter.environmentName"></a>
+
+- *Type:* string
+
+---
+
+###### `vpcCidr`<sup>Required</sup> <a name="vpcCidr" id="aws-ddk-core.AirflowDataPipeline.createVpc.parameter.vpcCidr"></a>
+
+- *Type:* string
+
+---
+
+#### Static Functions <a name="Static Functions" id="Static Functions"></a>
+
+| **Name** | **Description** |
+| --- | --- |
+| <code><a href="#aws-ddk-core.AirflowDataPipeline.isConstruct">isConstruct</a></code> | Checks if `x` is a construct. |
+
+---
+
+##### ~~`isConstruct`~~ <a name="isConstruct" id="aws-ddk-core.AirflowDataPipeline.isConstruct"></a>
+
+```typescript
+import { AirflowDataPipeline } from 'aws-ddk-core'
+
+AirflowDataPipeline.isConstruct(x: any)
+```
+
+Checks if `x` is a construct.
+
+###### `x`<sup>Required</sup> <a name="x" id="aws-ddk-core.AirflowDataPipeline.isConstruct.parameter.x"></a>
+
+- *Type:* any
+
+Any object.
+
+---
+
+#### Properties <a name="Properties" id="Properties"></a>
+
+| **Name** | **Type** | **Description** |
+| --- | --- | --- |
+| <code><a href="#aws-ddk-core.AirflowDataPipeline.property.node">node</a></code> | <code>constructs.Node</code> | The tree node. |
+| <code><a href="#aws-ddk-core.AirflowDataPipeline.property.dagProcessingLogs">dagProcessingLogs</a></code> | <code>string</code> | *No description.* |
+| <code><a href="#aws-ddk-core.AirflowDataPipeline.property.dagS3Path">dagS3Path</a></code> | <code>string</code> | *No description.* |
+| <code><a href="#aws-ddk-core.AirflowDataPipeline.property.mwaaEnvironment">mwaaEnvironment</a></code> | <code>aws-cdk-lib.aws_mwaa.CfnEnvironment</code> | *No description.* |
+| <code><a href="#aws-ddk-core.AirflowDataPipeline.property.s3Bucket">s3Bucket</a></code> | <code>aws-cdk-lib.aws_s3.IBucket</code> | *No description.* |
+| <code><a href="#aws-ddk-core.AirflowDataPipeline.property.schedulerLogsLevel">schedulerLogsLevel</a></code> | <code>string</code> | *No description.* |
+| <code><a href="#aws-ddk-core.AirflowDataPipeline.property.taskLogsLevel">taskLogsLevel</a></code> | <code>string</code> | *No description.* |
+| <code><a href="#aws-ddk-core.AirflowDataPipeline.property.vpc">vpc</a></code> | <code>aws-cdk-lib.aws_ec2.IVpc</code> | *No description.* |
+| <code><a href="#aws-ddk-core.AirflowDataPipeline.property.webserverLogsLevel">webserverLogsLevel</a></code> | <code>string</code> | *No description.* |
+| <code><a href="#aws-ddk-core.AirflowDataPipeline.property.workerLogsLevel">workerLogsLevel</a></code> | <code>string</code> | *No description.* |
+| <code><a href="#aws-ddk-core.AirflowDataPipeline.property.pluginFile">pluginFile</a></code> | <code>aws-cdk-lib.aws_s3_deployment.BucketDeployment</code> | *No description.* |
+
+---
+
+##### `node`<sup>Required</sup> <a name="node" id="aws-ddk-core.AirflowDataPipeline.property.node"></a>
+
+```typescript
+public readonly node: Node;
+```
+
+- *Type:* constructs.Node
+
+The tree node.
+
+---
+
+##### `dagProcessingLogs`<sup>Required</sup> <a name="dagProcessingLogs" id="aws-ddk-core.AirflowDataPipeline.property.dagProcessingLogs"></a>
+
+```typescript
+public readonly dagProcessingLogs: string;
+```
+
+- *Type:* string
+
+---
+
+##### `dagS3Path`<sup>Required</sup> <a name="dagS3Path" id="aws-ddk-core.AirflowDataPipeline.property.dagS3Path"></a>
+
+```typescript
+public readonly dagS3Path: string;
+```
+
+- *Type:* string
+
+---
+
+##### `mwaaEnvironment`<sup>Required</sup> <a name="mwaaEnvironment" id="aws-ddk-core.AirflowDataPipeline.property.mwaaEnvironment"></a>
+
+```typescript
+public readonly mwaaEnvironment: CfnEnvironment;
+```
+
+- *Type:* aws-cdk-lib.aws_mwaa.CfnEnvironment
+
+---
+
+##### `s3Bucket`<sup>Required</sup> <a name="s3Bucket" id="aws-ddk-core.AirflowDataPipeline.property.s3Bucket"></a>
+
+```typescript
+public readonly s3Bucket: IBucket;
+```
+
+- *Type:* aws-cdk-lib.aws_s3.IBucket
+
+---
+
+##### `schedulerLogsLevel`<sup>Required</sup> <a name="schedulerLogsLevel" id="aws-ddk-core.AirflowDataPipeline.property.schedulerLogsLevel"></a>
+
+```typescript
+public readonly schedulerLogsLevel: string;
+```
+
+- *Type:* string
+
+---
+
+##### `taskLogsLevel`<sup>Required</sup> <a name="taskLogsLevel" id="aws-ddk-core.AirflowDataPipeline.property.taskLogsLevel"></a>
+
+```typescript
+public readonly taskLogsLevel: string;
+```
+
+- *Type:* string
+
+---
+
+##### `vpc`<sup>Required</sup> <a name="vpc" id="aws-ddk-core.AirflowDataPipeline.property.vpc"></a>
+
+```typescript
+public readonly vpc: IVpc;
+```
+
+- *Type:* aws-cdk-lib.aws_ec2.IVpc
+
+---
+
+##### `webserverLogsLevel`<sup>Required</sup> <a name="webserverLogsLevel" id="aws-ddk-core.AirflowDataPipeline.property.webserverLogsLevel"></a>
+
+```typescript
+public readonly webserverLogsLevel: string;
+```
+
+- *Type:* string
+
+---
+
+##### `workerLogsLevel`<sup>Required</sup> <a name="workerLogsLevel" id="aws-ddk-core.AirflowDataPipeline.property.workerLogsLevel"></a>
+
+```typescript
+public readonly workerLogsLevel: string;
+```
+
+- *Type:* string
+
+---
+
+##### `pluginFile`<sup>Optional</sup> <a name="pluginFile" id="aws-ddk-core.AirflowDataPipeline.property.pluginFile"></a>
+
+```typescript
+public readonly pluginFile: BucketDeployment;
+```
+
+- *Type:* aws-cdk-lib.aws_s3_deployment.BucketDeployment
+
+---
+
+
 ### AppFlowIngestionStage <a name="AppFlowIngestionStage" id="aws-ddk-core.AppFlowIngestionStage"></a>
 
 Stage that contains a step function that runs an AppFlow flow ingestion.
@@ -5992,6 +6226,514 @@ Name of the stage.
 
 ---
 
+### AirflowPipelineProps <a name="AirflowPipelineProps" id="aws-ddk-core.AirflowPipelineProps"></a>
+
+#### Initializer <a name="Initializer" id="aws-ddk-core.AirflowPipelineProps.Initializer"></a>
+
+```typescript
+import { AirflowPipelineProps } from 'aws-ddk-core'
+
+const airflowPipelineProps: AirflowPipelineProps = { ... }
+```
+
+#### Properties <a name="Properties" id="Properties"></a>
+
+| **Name** | **Type** | **Description** |
+| --- | --- | --- |
+| <code><a href="#aws-ddk-core.AirflowPipelineProps.property.name">name</a></code> | <code>string</code> | The name of your Amazon MWAA environment. |
+| <code><a href="#aws-ddk-core.AirflowPipelineProps.property.airflowConfigurationOptions">airflowConfigurationOptions</a></code> | <code>any</code> | A list of key-value pairs containing the Airflow configuration options for your environment. |
+| <code><a href="#aws-ddk-core.AirflowPipelineProps.property.airflowVersion">airflowVersion</a></code> | <code>string</code> | The version of Apache Airflow to use for the environment. |
+| <code><a href="#aws-ddk-core.AirflowPipelineProps.property.dagS3Path">dagS3Path</a></code> | <code>string</code> | The relative path to the DAGs folder on your Amazon S3 bucket. |
+| <code><a href="#aws-ddk-core.AirflowPipelineProps.property.environmentClass">environmentClass</a></code> | <code>string</code> | The environment class type. |
+| <code><a href="#aws-ddk-core.AirflowPipelineProps.property.executionRoleArn">executionRoleArn</a></code> | <code>string</code> | The Amazon Resource Name (ARN) of the execution role in IAM that allows MWAA to access AWS resources in your environment. |
+| <code><a href="#aws-ddk-core.AirflowPipelineProps.property.kmsKey">kmsKey</a></code> | <code>string</code> | The AWS Key Management Service (KMS) key to encrypt and decrypt the data in your environment. |
+| <code><a href="#aws-ddk-core.AirflowPipelineProps.property.loggingConfiguration">loggingConfiguration</a></code> | <code>aws-cdk-lib.aws_mwaa.CfnEnvironment.LoggingConfigurationProperty \| aws-cdk-lib.IResolvable</code> | The Apache Airflow logs being sent to CloudWatch Logs: `DagProcessingLogs` , `SchedulerLogs` , `TaskLogs` , `WebserverLogs` , `WorkerLogs` . |
+| <code><a href="#aws-ddk-core.AirflowPipelineProps.property.maxWorkers">maxWorkers</a></code> | <code>number</code> | The maximum number of workers that you want to run in your environment. |
+| <code><a href="#aws-ddk-core.AirflowPipelineProps.property.minWorkers">minWorkers</a></code> | <code>number</code> | The minimum number of workers that you want to run in your environment. |
+| <code><a href="#aws-ddk-core.AirflowPipelineProps.property.networkConfiguration">networkConfiguration</a></code> | <code>aws-cdk-lib.aws_mwaa.CfnEnvironment.NetworkConfigurationProperty \| aws-cdk-lib.IResolvable</code> | The VPC networking components used to secure and enable network traffic between the AWS resources for your environment. |
+| <code><a href="#aws-ddk-core.AirflowPipelineProps.property.pluginsS3ObjectVersion">pluginsS3ObjectVersion</a></code> | <code>string</code> | The version of the plugins.zip file on your Amazon S3 bucket. To learn more, see [Installing custom plugins](https://docs.aws.amazon.com/mwaa/latest/userguide/configuring-dag-import-plugins.html) . |
+| <code><a href="#aws-ddk-core.AirflowPipelineProps.property.pluginsS3Path">pluginsS3Path</a></code> | <code>string</code> | The relative path to the `plugins.zip` file on your Amazon S3 bucket. For example, `plugins.zip` . To learn more, see [Installing custom plugins](https://docs.aws.amazon.com/mwaa/latest/userguide/configuring-dag-import-plugins.html) . |
+| <code><a href="#aws-ddk-core.AirflowPipelineProps.property.requirementsS3ObjectVersion">requirementsS3ObjectVersion</a></code> | <code>string</code> | The version of the requirements.txt file on your Amazon S3 bucket. To learn more, see [Installing Python dependencies](https://docs.aws.amazon.com/mwaa/latest/userguide/working-dags-dependencies.html) . |
+| <code><a href="#aws-ddk-core.AirflowPipelineProps.property.requirementsS3Path">requirementsS3Path</a></code> | <code>string</code> | The relative path to the `requirements.txt` file on your Amazon S3 bucket. For example, `requirements.txt` . To learn more, see [Installing Python dependencies](https://docs.aws.amazon.com/mwaa/latest/userguide/working-dags-dependencies.html) . |
+| <code><a href="#aws-ddk-core.AirflowPipelineProps.property.schedulers">schedulers</a></code> | <code>number</code> | The number of schedulers that you want to run in your environment. Valid values:. |
+| <code><a href="#aws-ddk-core.AirflowPipelineProps.property.sourceBucketArn">sourceBucketArn</a></code> | <code>string</code> | The Amazon Resource Name (ARN) of the Amazon S3 bucket where your DAG code and supporting files are stored. |
+| <code><a href="#aws-ddk-core.AirflowPipelineProps.property.tags">tags</a></code> | <code>any</code> | The key-value tag pairs associated to your environment. |
+| <code><a href="#aws-ddk-core.AirflowPipelineProps.property.webserverAccessMode">webserverAccessMode</a></code> | <code>string</code> | The Apache Airflow *Web server* access mode. |
+| <code><a href="#aws-ddk-core.AirflowPipelineProps.property.weeklyMaintenanceWindowStart">weeklyMaintenanceWindowStart</a></code> | <code>string</code> | The day and time of the week to start weekly maintenance updates of your environment in the following format: `DAY:HH:MM` . |
+| <code><a href="#aws-ddk-core.AirflowPipelineProps.property.additionalPolicyStatements">additionalPolicyStatements</a></code> | <code>aws-cdk-lib.aws_iam.PolicyStatement[]</code> | Additional policy statements to add to the airflow execution role. |
+| <code><a href="#aws-ddk-core.AirflowPipelineProps.property.dagFiles">dagFiles</a></code> | <code>string[]</code> | File(s) to be uploaded to dags location in s3 bucket. |
+| <code><a href="#aws-ddk-core.AirflowPipelineProps.property.dagProcessingLogs">dagProcessingLogs</a></code> | <code>string</code> | Log level for DagProcessing. |
+| <code><a href="#aws-ddk-core.AirflowPipelineProps.property.pluginFile">pluginFile</a></code> | <code>string</code> | Plugin file to be uploaded to plugin path in S3. |
+| <code><a href="#aws-ddk-core.AirflowPipelineProps.property.requirementsFile">requirementsFile</a></code> | <code>string</code> | Requirements file to be uploaded to plugin path in S3. |
+| <code><a href="#aws-ddk-core.AirflowPipelineProps.property.s3Bucket">s3Bucket</a></code> | <code>aws-cdk-lib.aws_s3.IBucket</code> | S3 Bucket. |
+| <code><a href="#aws-ddk-core.AirflowPipelineProps.property.schedulerLogsLevel">schedulerLogsLevel</a></code> | <code>string</code> | Log level for SchedulerLogs. |
+| <code><a href="#aws-ddk-core.AirflowPipelineProps.property.taskLogsLevel">taskLogsLevel</a></code> | <code>string</code> | Log level for TaskLogs. |
+| <code><a href="#aws-ddk-core.AirflowPipelineProps.property.vpcCidr">vpcCidr</a></code> | <code>string</code> | The IP range (CIDR notation) for this VPC. |
+| <code><a href="#aws-ddk-core.AirflowPipelineProps.property.vpcId">vpcId</a></code> | <code>string</code> | Existing vpc id. |
+| <code><a href="#aws-ddk-core.AirflowPipelineProps.property.webserverLogsLevel">webserverLogsLevel</a></code> | <code>string</code> | Log level for WebserverLogs. |
+| <code><a href="#aws-ddk-core.AirflowPipelineProps.property.workerLogsLevel">workerLogsLevel</a></code> | <code>string</code> | Log level for WorkerLogs. |
+
+---
+
+##### `name`<sup>Required</sup> <a name="name" id="aws-ddk-core.AirflowPipelineProps.property.name"></a>
+
+```typescript
+public readonly name: string;
+```
+
+- *Type:* string
+
+The name of your Amazon MWAA environment.
+
+> [http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-mwaa-environment.html#cfn-mwaa-environment-name](http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-mwaa-environment.html#cfn-mwaa-environment-name)
+
+---
+
+##### `airflowConfigurationOptions`<sup>Optional</sup> <a name="airflowConfigurationOptions" id="aws-ddk-core.AirflowPipelineProps.property.airflowConfigurationOptions"></a>
+
+```typescript
+public readonly airflowConfigurationOptions: any;
+```
+
+- *Type:* any
+
+A list of key-value pairs containing the Airflow configuration options for your environment.
+
+For example, `core.default_timezone: utc` . To learn more, see [Apache Airflow configuration options](https://docs.aws.amazon.com/mwaa/latest/userguide/configuring-env-variables.html) .
+
+> [http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-mwaa-environment.html#cfn-mwaa-environment-airflowconfigurationoptions](http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-mwaa-environment.html#cfn-mwaa-environment-airflowconfigurationoptions)
+
+---
+
+##### `airflowVersion`<sup>Optional</sup> <a name="airflowVersion" id="aws-ddk-core.AirflowPipelineProps.property.airflowVersion"></a>
+
+```typescript
+public readonly airflowVersion: string;
+```
+
+- *Type:* string
+
+The version of Apache Airflow to use for the environment.
+
+If no value is specified, defaults to the latest version. Valid values: `2.0.2` , `1.10.12` , `2.2.2` , and `2.4.3` .
+
+> [http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-mwaa-environment.html#cfn-mwaa-environment-airflowversion](http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-mwaa-environment.html#cfn-mwaa-environment-airflowversion)
+
+---
+
+##### `dagS3Path`<sup>Optional</sup> <a name="dagS3Path" id="aws-ddk-core.AirflowPipelineProps.property.dagS3Path"></a>
+
+```typescript
+public readonly dagS3Path: string;
+```
+
+- *Type:* string
+
+The relative path to the DAGs folder on your Amazon S3 bucket.
+
+For example, `dags` . To learn more, see [Adding or updating DAGs](https://docs.aws.amazon.com/mwaa/latest/userguide/configuring-dag-folder.html) .
+
+> [http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-mwaa-environment.html#cfn-mwaa-environment-dags3path](http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-mwaa-environment.html#cfn-mwaa-environment-dags3path)
+
+---
+
+##### `environmentClass`<sup>Optional</sup> <a name="environmentClass" id="aws-ddk-core.AirflowPipelineProps.property.environmentClass"></a>
+
+```typescript
+public readonly environmentClass: string;
+```
+
+- *Type:* string
+
+The environment class type.
+
+Valid values: `mw1.small` , `mw1.medium` , `mw1.large` . To learn more, see [Amazon MWAA environment class](https://docs.aws.amazon.com/mwaa/latest/userguide/environment-class.html) .
+
+> [http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-mwaa-environment.html#cfn-mwaa-environment-environmentclass](http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-mwaa-environment.html#cfn-mwaa-environment-environmentclass)
+
+---
+
+##### `executionRoleArn`<sup>Optional</sup> <a name="executionRoleArn" id="aws-ddk-core.AirflowPipelineProps.property.executionRoleArn"></a>
+
+```typescript
+public readonly executionRoleArn: string;
+```
+
+- *Type:* string
+
+The Amazon Resource Name (ARN) of the execution role in IAM that allows MWAA to access AWS resources in your environment.
+
+For example, `arn:aws:iam::123456789:role/my-execution-role` . To learn more, see [Amazon MWAA Execution role](https://docs.aws.amazon.com/mwaa/latest/userguide/mwaa-create-role.html) .
+
+> [http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-mwaa-environment.html#cfn-mwaa-environment-executionrolearn](http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-mwaa-environment.html#cfn-mwaa-environment-executionrolearn)
+
+---
+
+##### `kmsKey`<sup>Optional</sup> <a name="kmsKey" id="aws-ddk-core.AirflowPipelineProps.property.kmsKey"></a>
+
+```typescript
+public readonly kmsKey: string;
+```
+
+- *Type:* string
+
+The AWS Key Management Service (KMS) key to encrypt and decrypt the data in your environment.
+
+You can use an AWS KMS key managed by MWAA, or a customer-managed KMS key (advanced).
+
+> [http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-mwaa-environment.html#cfn-mwaa-environment-kmskey](http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-mwaa-environment.html#cfn-mwaa-environment-kmskey)
+
+---
+
+##### `loggingConfiguration`<sup>Optional</sup> <a name="loggingConfiguration" id="aws-ddk-core.AirflowPipelineProps.property.loggingConfiguration"></a>
+
+```typescript
+public readonly loggingConfiguration: LoggingConfigurationProperty | IResolvable;
+```
+
+- *Type:* aws-cdk-lib.aws_mwaa.CfnEnvironment.LoggingConfigurationProperty | aws-cdk-lib.IResolvable
+
+The Apache Airflow logs being sent to CloudWatch Logs: `DagProcessingLogs` , `SchedulerLogs` , `TaskLogs` , `WebserverLogs` , `WorkerLogs` .
+
+> [http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-mwaa-environment.html#cfn-mwaa-environment-loggingconfiguration](http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-mwaa-environment.html#cfn-mwaa-environment-loggingconfiguration)
+
+---
+
+##### `maxWorkers`<sup>Optional</sup> <a name="maxWorkers" id="aws-ddk-core.AirflowPipelineProps.property.maxWorkers"></a>
+
+```typescript
+public readonly maxWorkers: number;
+```
+
+- *Type:* number
+
+The maximum number of workers that you want to run in your environment.
+
+MWAA scales the number of Apache Airflow workers up to the number you specify in the `MaxWorkers` field. For example, `20` . When there are no more tasks running, and no more in the queue, MWAA disposes of the extra workers leaving the one worker that is included with your environment, or the number you specify in `MinWorkers` .
+
+> [http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-mwaa-environment.html#cfn-mwaa-environment-maxworkers](http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-mwaa-environment.html#cfn-mwaa-environment-maxworkers)
+
+---
+
+##### `minWorkers`<sup>Optional</sup> <a name="minWorkers" id="aws-ddk-core.AirflowPipelineProps.property.minWorkers"></a>
+
+```typescript
+public readonly minWorkers: number;
+```
+
+- *Type:* number
+
+The minimum number of workers that you want to run in your environment.
+
+MWAA scales the number of Apache Airflow workers up to the number you specify in the `MaxWorkers` field. When there are no more tasks running, and no more in the queue, MWAA disposes of the extra workers leaving the worker count you specify in the `MinWorkers` field. For example, `2` .
+
+> [http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-mwaa-environment.html#cfn-mwaa-environment-minworkers](http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-mwaa-environment.html#cfn-mwaa-environment-minworkers)
+
+---
+
+##### `networkConfiguration`<sup>Optional</sup> <a name="networkConfiguration" id="aws-ddk-core.AirflowPipelineProps.property.networkConfiguration"></a>
+
+```typescript
+public readonly networkConfiguration: NetworkConfigurationProperty | IResolvable;
+```
+
+- *Type:* aws-cdk-lib.aws_mwaa.CfnEnvironment.NetworkConfigurationProperty | aws-cdk-lib.IResolvable
+
+The VPC networking components used to secure and enable network traffic between the AWS resources for your environment.
+
+To learn more, see [About networking on Amazon MWAA](https://docs.aws.amazon.com/mwaa/latest/userguide/networking-about.html) .
+
+> [http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-mwaa-environment.html#cfn-mwaa-environment-networkconfiguration](http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-mwaa-environment.html#cfn-mwaa-environment-networkconfiguration)
+
+---
+
+##### `pluginsS3ObjectVersion`<sup>Optional</sup> <a name="pluginsS3ObjectVersion" id="aws-ddk-core.AirflowPipelineProps.property.pluginsS3ObjectVersion"></a>
+
+```typescript
+public readonly pluginsS3ObjectVersion: string;
+```
+
+- *Type:* string
+
+The version of the plugins.zip file on your Amazon S3 bucket. To learn more, see [Installing custom plugins](https://docs.aws.amazon.com/mwaa/latest/userguide/configuring-dag-import-plugins.html) .
+
+> [http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-mwaa-environment.html#cfn-mwaa-environment-pluginss3objectversion](http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-mwaa-environment.html#cfn-mwaa-environment-pluginss3objectversion)
+
+---
+
+##### `pluginsS3Path`<sup>Optional</sup> <a name="pluginsS3Path" id="aws-ddk-core.AirflowPipelineProps.property.pluginsS3Path"></a>
+
+```typescript
+public readonly pluginsS3Path: string;
+```
+
+- *Type:* string
+
+The relative path to the `plugins.zip` file on your Amazon S3 bucket. For example, `plugins.zip` . To learn more, see [Installing custom plugins](https://docs.aws.amazon.com/mwaa/latest/userguide/configuring-dag-import-plugins.html) .
+
+> [http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-mwaa-environment.html#cfn-mwaa-environment-pluginss3path](http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-mwaa-environment.html#cfn-mwaa-environment-pluginss3path)
+
+---
+
+##### `requirementsS3ObjectVersion`<sup>Optional</sup> <a name="requirementsS3ObjectVersion" id="aws-ddk-core.AirflowPipelineProps.property.requirementsS3ObjectVersion"></a>
+
+```typescript
+public readonly requirementsS3ObjectVersion: string;
+```
+
+- *Type:* string
+
+The version of the requirements.txt file on your Amazon S3 bucket. To learn more, see [Installing Python dependencies](https://docs.aws.amazon.com/mwaa/latest/userguide/working-dags-dependencies.html) .
+
+> [http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-mwaa-environment.html#cfn-mwaa-environment-requirementss3objectversion](http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-mwaa-environment.html#cfn-mwaa-environment-requirementss3objectversion)
+
+---
+
+##### `requirementsS3Path`<sup>Optional</sup> <a name="requirementsS3Path" id="aws-ddk-core.AirflowPipelineProps.property.requirementsS3Path"></a>
+
+```typescript
+public readonly requirementsS3Path: string;
+```
+
+- *Type:* string
+
+The relative path to the `requirements.txt` file on your Amazon S3 bucket. For example, `requirements.txt` . To learn more, see [Installing Python dependencies](https://docs.aws.amazon.com/mwaa/latest/userguide/working-dags-dependencies.html) .
+
+> [http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-mwaa-environment.html#cfn-mwaa-environment-requirementss3path](http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-mwaa-environment.html#cfn-mwaa-environment-requirementss3path)
+
+---
+
+##### `schedulers`<sup>Optional</sup> <a name="schedulers" id="aws-ddk-core.AirflowPipelineProps.property.schedulers"></a>
+
+```typescript
+public readonly schedulers: number;
+```
+
+- *Type:* number
+
+The number of schedulers that you want to run in your environment. Valid values:.
+
+*v2* - Accepts between 2 to 5. Defaults to 2.
+- *v1* - Accepts 1.
+
+> [http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-mwaa-environment.html#cfn-mwaa-environment-schedulers](http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-mwaa-environment.html#cfn-mwaa-environment-schedulers)
+
+---
+
+##### `sourceBucketArn`<sup>Optional</sup> <a name="sourceBucketArn" id="aws-ddk-core.AirflowPipelineProps.property.sourceBucketArn"></a>
+
+```typescript
+public readonly sourceBucketArn: string;
+```
+
+- *Type:* string
+
+The Amazon Resource Name (ARN) of the Amazon S3 bucket where your DAG code and supporting files are stored.
+
+For example, `arn:aws:s3:::my-airflow-bucket-unique-name` . To learn more, see [Create an Amazon S3 bucket for Amazon MWAA](https://docs.aws.amazon.com/mwaa/latest/userguide/mwaa-s3-bucket.html) .
+
+> [http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-mwaa-environment.html#cfn-mwaa-environment-sourcebucketarn](http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-mwaa-environment.html#cfn-mwaa-environment-sourcebucketarn)
+
+---
+
+##### `tags`<sup>Optional</sup> <a name="tags" id="aws-ddk-core.AirflowPipelineProps.property.tags"></a>
+
+```typescript
+public readonly tags: any;
+```
+
+- *Type:* any
+
+The key-value tag pairs associated to your environment.
+
+For example, `"Environment": "Staging"` . To learn more, see [Tagging](https://docs.aws.amazon.com/general/latest/gr/aws_tagging.html) .
+
+> [http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-mwaa-environment.html#cfn-mwaa-environment-tags](http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-mwaa-environment.html#cfn-mwaa-environment-tags)
+
+---
+
+##### `webserverAccessMode`<sup>Optional</sup> <a name="webserverAccessMode" id="aws-ddk-core.AirflowPipelineProps.property.webserverAccessMode"></a>
+
+```typescript
+public readonly webserverAccessMode: string;
+```
+
+- *Type:* string
+
+The Apache Airflow *Web server* access mode.
+
+To learn more, see [Apache Airflow access modes](https://docs.aws.amazon.com/mwaa/latest/userguide/configuring-networking.html) . Valid values: `PRIVATE_ONLY` or `PUBLIC_ONLY` .
+
+> [http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-mwaa-environment.html#cfn-mwaa-environment-webserveraccessmode](http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-mwaa-environment.html#cfn-mwaa-environment-webserveraccessmode)
+
+---
+
+##### `weeklyMaintenanceWindowStart`<sup>Optional</sup> <a name="weeklyMaintenanceWindowStart" id="aws-ddk-core.AirflowPipelineProps.property.weeklyMaintenanceWindowStart"></a>
+
+```typescript
+public readonly weeklyMaintenanceWindowStart: string;
+```
+
+- *Type:* string
+
+The day and time of the week to start weekly maintenance updates of your environment in the following format: `DAY:HH:MM` .
+
+For example: `TUE:03:30` . You can specify a start time in 30 minute increments only. Supported input includes the following:
+
+- MON|TUE|WED|THU|FRI|SAT|SUN:([01]\\d|2[0-3]):(00|30)
+
+> [http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-mwaa-environment.html#cfn-mwaa-environment-weeklymaintenancewindowstart](http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-mwaa-environment.html#cfn-mwaa-environment-weeklymaintenancewindowstart)
+
+---
+
+##### `additionalPolicyStatements`<sup>Optional</sup> <a name="additionalPolicyStatements" id="aws-ddk-core.AirflowPipelineProps.property.additionalPolicyStatements"></a>
+
+```typescript
+public readonly additionalPolicyStatements: PolicyStatement[];
+```
+
+- *Type:* aws-cdk-lib.aws_iam.PolicyStatement[]
+
+Additional policy statements to add to the airflow execution role.
+
+---
+
+##### `dagFiles`<sup>Optional</sup> <a name="dagFiles" id="aws-ddk-core.AirflowPipelineProps.property.dagFiles"></a>
+
+```typescript
+public readonly dagFiles: string[];
+```
+
+- *Type:* string[]
+
+File(s) to be uploaded to dags location in s3 bucket.
+
+---
+
+##### `dagProcessingLogs`<sup>Optional</sup> <a name="dagProcessingLogs" id="aws-ddk-core.AirflowPipelineProps.property.dagProcessingLogs"></a>
+
+```typescript
+public readonly dagProcessingLogs: string;
+```
+
+- *Type:* string
+
+Log level for DagProcessing.
+
+---
+
+##### `pluginFile`<sup>Optional</sup> <a name="pluginFile" id="aws-ddk-core.AirflowPipelineProps.property.pluginFile"></a>
+
+```typescript
+public readonly pluginFile: string;
+```
+
+- *Type:* string
+
+Plugin file to be uploaded to plugin path in S3.
+
+'pluginsS3Path' must be specified as well.
+
+---
+
+##### `requirementsFile`<sup>Optional</sup> <a name="requirementsFile" id="aws-ddk-core.AirflowPipelineProps.property.requirementsFile"></a>
+
+```typescript
+public readonly requirementsFile: string;
+```
+
+- *Type:* string
+
+Requirements file to be uploaded to plugin path in S3.
+
+'requirementsS3Path' must be specified as well.
+
+---
+
+##### `s3Bucket`<sup>Optional</sup> <a name="s3Bucket" id="aws-ddk-core.AirflowPipelineProps.property.s3Bucket"></a>
+
+```typescript
+public readonly s3Bucket: IBucket;
+```
+
+- *Type:* aws-cdk-lib.aws_s3.IBucket
+
+S3 Bucket.
+
+---
+
+##### `schedulerLogsLevel`<sup>Optional</sup> <a name="schedulerLogsLevel" id="aws-ddk-core.AirflowPipelineProps.property.schedulerLogsLevel"></a>
+
+```typescript
+public readonly schedulerLogsLevel: string;
+```
+
+- *Type:* string
+
+Log level for SchedulerLogs.
+
+---
+
+##### `taskLogsLevel`<sup>Optional</sup> <a name="taskLogsLevel" id="aws-ddk-core.AirflowPipelineProps.property.taskLogsLevel"></a>
+
+```typescript
+public readonly taskLogsLevel: string;
+```
+
+- *Type:* string
+
+Log level for TaskLogs.
+
+---
+
+##### `vpcCidr`<sup>Optional</sup> <a name="vpcCidr" id="aws-ddk-core.AirflowPipelineProps.property.vpcCidr"></a>
+
+```typescript
+public readonly vpcCidr: string;
+```
+
+- *Type:* string
+
+The IP range (CIDR notation) for this VPC.
+
+---
+
+##### `vpcId`<sup>Optional</sup> <a name="vpcId" id="aws-ddk-core.AirflowPipelineProps.property.vpcId"></a>
+
+```typescript
+public readonly vpcId: string;
+```
+
+- *Type:* string
+
+Existing vpc id.
+
+---
+
+##### `webserverLogsLevel`<sup>Optional</sup> <a name="webserverLogsLevel" id="aws-ddk-core.AirflowPipelineProps.property.webserverLogsLevel"></a>
+
+```typescript
+public readonly webserverLogsLevel: string;
+```
+
+- *Type:* string
+
+Log level for WebserverLogs.
+
+---
+
+##### `workerLogsLevel`<sup>Optional</sup> <a name="workerLogsLevel" id="aws-ddk-core.AirflowPipelineProps.property.workerLogsLevel"></a>
+
+```typescript
+public readonly workerLogsLevel: string;
+```
+
+- *Type:* string
+
+Log level for WorkerLogs.
+
+---
+
 ### AlarmProps <a name="AlarmProps" id="aws-ddk-core.AlarmProps"></a>
 
 Properties for the alarm being added to the DataStage.
diff --git a/integ/dags/bar.py b/integ/dags/bar.py
new file mode 100644
index 00000000..e69de29b
diff --git a/integ/dags/foo.py b/integ/dags/foo.py
new file mode 100644
index 00000000..e69de29b
diff --git a/src/core/airflow-pipeline.ts b/src/core/airflow-pipeline.ts
new file mode 100644
index 00000000..2195ae86
--- /dev/null
+++ b/src/core/airflow-pipeline.ts
@@ -0,0 +1,289 @@
+import * as cdk from "aws-cdk-lib";
+import * as ec2 from "aws-cdk-lib/aws-ec2";
+import * as iam from "aws-cdk-lib/aws-iam";
+import * as mwaa from "aws-cdk-lib/aws-mwaa";
+import * as s3 from "aws-cdk-lib/aws-s3";
+import * as s3deploy from "aws-cdk-lib/aws-s3-deployment";
+import { Construct } from "constructs";
+
+import { S3Factory } from "./s3-factory";
+
+export interface AirflowPipelineProps extends mwaa.CfnEnvironmentProps {
+  /**
+   * Log level for DagProcessing
+   */
+  readonly dagProcessingLogs?: string;
+  /**
+   * Log level for SchedulerLogs
+   */
+  readonly schedulerLogsLevel?: string;
+  /**
+   * Log level for TaskLogs
+   */
+  readonly taskLogsLevel?: string;
+  /**
+   * Log level for WorkerLogs
+   */
+  readonly workerLogsLevel?: string;
+  /**
+   * Log level for WebserverLogs
+   */
+  readonly webserverLogsLevel?: string;
+  /**
+   * Existing vpc id
+   */
+  readonly vpcId?: string;
+  /**
+   * The IP range (CIDR notation) for this VPC
+   */
+  readonly vpcCidr?: string;
+  /**
+   * S3 Bucket
+   */
+  readonly s3Bucket?: s3.IBucket;
+  /**
+   * File(s) to be uploaded to dags location in s3 bucket.
+   */
+  readonly dagFiles?: string[];
+  /**
+   * Requirements file to be uploaded to plugin path in S3. 'requirementsS3Path' must be specified as well.
+   */
+  readonly requirementsFile?: string;
+  /**
+   * Plugin file to be uploaded to plugin path in S3. 'pluginsS3Path' must be specified as well.
+   */
+  readonly pluginFile?: string;
+  /**
+   * Additional policy statements to add to the airflow execution role
+   */
+  readonly additionalPolicyStatements?: iam.PolicyStatement[];
+}
+
+export class AirflowDataPipeline extends Construct {
+  readonly dagProcessingLogs: string;
+  readonly schedulerLogsLevel: string;
+  readonly taskLogsLevel: string;
+  readonly workerLogsLevel: string;
+  readonly webserverLogsLevel: string;
+  readonly vpc: ec2.IVpc;
+  readonly s3Bucket: s3.IBucket;
+  readonly mwaaEnvironment: mwaa.CfnEnvironment;
+  readonly dagS3Path: string;
+  readonly pluginFile?: s3deploy.BucketDeployment;
+
+  constructor(scope: Construct, id: string, props: AirflowPipelineProps) {
+    super(scope, id);
+
+    this.dagProcessingLogs = props.dagProcessingLogs ?? "INFO";
+    this.schedulerLogsLevel = props.schedulerLogsLevel ?? "INFO";
+    this.taskLogsLevel = props.taskLogsLevel ?? "INFO";
+    this.workerLogsLevel = props.workerLogsLevel ?? "INFO";
+    this.webserverLogsLevel = props.webserverLogsLevel ?? "INFO";
+    this.dagS3Path = props.dagS3Path ?? "dags";
+
+    if (props.vpcId) {
+      this.vpc = ec2.Vpc.fromLookup(scope, "VPC", { vpcId: props.vpcId });
+    } else if (props.vpcCidr) {
+      this.vpc = this.createVpc(scope, props.name, props.vpcCidr);
+    } else {
+      throw new Error("One of 'vpcId' or 'vpcCidr' must be provided");
+    }
+
+    const securityGroup = new ec2.SecurityGroup(scope, `${props.name} Security Group`, {
+      securityGroupName: `${props.name} Security Group`,
+      description: "Security group with a self-referencing inbound rule.",
+      vpc: this.vpc,
+    });
+    securityGroup.addIngressRule(securityGroup, ec2.Port.allTraffic(), "Self referencing rule");
+
+    if (props.s3Bucket) {
+      this.s3Bucket = props.s3Bucket;
+    } else {
+      this.s3Bucket = S3Factory.bucket(this, "Environment Bucket", {
+        versioned: true,
+      });
+    }
+
+    if (props.dagFiles) {
+      var sources: s3deploy.ISource[] = [];
+      props.dagFiles.forEach((location) => {
+        sources.push(s3deploy.Source.asset(location));
+      });
+      new s3deploy.BucketDeployment(this, "Deploy Dag Files", {
+        sources: sources,
+        destinationBucket: this.s3Bucket,
+        destinationKeyPrefix: this.dagS3Path,
+      });
+    }
+
+    if (props.pluginFile) {
+      if (props.pluginsS3Path) {
+        this.pluginFile = new s3deploy.BucketDeployment(this, "Deploy Plugin File", {
+          sources: [s3deploy.Source.asset(props.pluginFile)],
+          destinationBucket: this.s3Bucket,
+          destinationKeyPrefix: props.pluginsS3Path,
+        });
+      } else {
+        throw new Error("'pluginsS3Path' must be specified if a 'pluginFile' is specified.");
+      }
+    }
+
+    if (props.requirementsFile) {
+      if (props.requirementsS3Path) {
+        this.pluginFile = new s3deploy.BucketDeployment(this, "Deploy Requirements File", {
+          sources: [
+            s3deploy.Source.asset(props.requirementsFile.split("/").slice(0, -1).join("/"), {
+              exclude: ["**", `!${props.requirementsFile.split("/")[-1]}`],
+            }),
+          ],
+          destinationBucket: this.s3Bucket,
+          destinationKeyPrefix: props.requirementsS3Path,
+        });
+      } else {
+        throw new Error("'requirementsS3Path' must be specified if a 'requirementsFile' is specified.");
+      }
+    }
+
+    const mwaaExecutionRole = new iam.Role(scope, "MWAA Execution Role", {
+      assumedBy: new iam.CompositePrincipal(
+        new iam.ServicePrincipal("airflow.amazonaws.com"),
+        new iam.ServicePrincipal("airflow-env.amazonaws.com"),
+      ),
+      path: "/service-role/",
+    });
+    mwaaExecutionRole.addManagedPolicy(
+      new iam.ManagedPolicy(this, "MWAA Execution Policy", {
+        statements: [
+          new iam.PolicyStatement({
+            actions: ["airflow:PublishMetrics"],
+            resources: [
+              `arn:aws:airflow:${cdk.Stack.of(scope).region}:${cdk.Stack.of(scope).account}:environment/${props.name}`,
+            ],
+          }),
+          new iam.PolicyStatement({
+            effect: iam.Effect.DENY,
+            actions: ["s3:ListAllMyBuckets"],
+            resources: [this.s3Bucket.bucketArn, `${this.s3Bucket.bucketArn}/*`],
+          }),
+          new iam.PolicyStatement({
+            actions: ["s3:GetObject*", "s3:GetBucket*", "s3:List*"],
+            resources: [this.s3Bucket.bucketArn, `${this.s3Bucket.bucketArn}/*`],
+          }),
+          new iam.PolicyStatement({
+            actions: ["logs:DescribeLogGroups"],
+            resources: ["*"],
+          }),
+          new iam.PolicyStatement({
+            actions: [
+              "logs:CreateLogStream",
+              "logs:CreateLogGroup",
+              "logs:PutLogEvents",
+              "logs:GetLogEvents",
+              "logs:GetLogRecord",
+              "logs:GetLogGroupFields",
+              "logs:GetQueryResults",
+              "logs:DescribeLogGroups",
+            ],
+            resources: [
+              `arn:aws:logs:${cdk.Stack.of(scope).region}:${cdk.Stack.of(scope).account}:log-group:airflow-${
+                cdk.Stack.of(scope).stackName
+              }*`,
+            ],
+          }),
+          new iam.PolicyStatement({
+            actions: ["cloudwatch:PutMetricData"],
+            resources: ["*"],
+          }),
+          new iam.PolicyStatement({
+            actions: [
+              "sqs:ChangeMessageVisibility",
+              "sqs:DeleteMessage",
+              "sqs:GetQueueAttributes",
+              "sqs:GetQueueUrl",
+              "sqs:ReceiveMessage",
+              "sqs:SendMessage",
+            ],
+            resources: [`arn:aws:sqs:${cdk.Stack.of(scope).region}:*:airflow-celery-*`],
+          }),
+          new iam.PolicyStatement({
+            actions: ["kms:Decrypt", "kms:DescribeKey", "kms:GenerateDataKey*", "kms:Encrypt"],
+            notResources: [`arn:aws:kms:*:${cdk.Stack.of(scope).account}:key/*`],
+            conditions: {
+              StringLike: {
+                "kms:ViaService": [`sqs.${cdk.Stack.of(scope).region}.amazonaws.com`],
+              },
+            },
+          }),
+        ],
+      }),
+    );
+
+    if (props.additionalPolicyStatements) {
+      props.additionalPolicyStatements.forEach((statement) => {
+        mwaaExecutionRole.addToPolicy(statement);
+      });
+    }
+
+    this.mwaaEnvironment = new mwaa.CfnEnvironment(this, "MWAA Environment", {
+      sourceBucketArn: this.s3Bucket.bucketArn,
+      executionRoleArn: mwaaExecutionRole.roleArn,
+      dagS3Path: this.dagS3Path,
+      networkConfiguration: {
+        securityGroupIds: [securityGroup.securityGroupId],
+        subnetIds: this.vpc.selectSubnets({ subnetType: ec2.SubnetType.PRIVATE_WITH_EGRESS }).subnetIds,
+      },
+      webserverAccessMode: "PUBLIC_ONLY",
+      loggingConfiguration: {
+        dagProcessingLogs: {
+          enabled: true,
+          logLevel: this.dagProcessingLogs,
+        },
+        schedulerLogs: {
+          enabled: true,
+          logLevel: this.schedulerLogsLevel,
+        },
+        taskLogs: {
+          enabled: true,
+          logLevel: this.taskLogsLevel,
+        },
+        webserverLogs: {
+          enabled: true,
+          logLevel: this.webserverLogsLevel,
+        },
+        workerLogs: {
+          enabled: false,
+          logLevel: this.workerLogsLevel,
+        },
+      },
+      ...props,
+    });
+  }
+  createVpc(scope: Construct, environmentName: string, vpcCidr: string): ec2.IVpc {
+    const resourceName = `${environmentName}-MWAA`;
+    const vpcCIDRMask = +vpcCidr.split("/")[1];
+    if (vpcCIDRMask > 20 || vpcCIDRMask < 16) {
+      throw new Error("Vpc Cidr Range must of size >=16 and <=20");
+    }
+    const subnetCIDRMask = vpcCIDRMask + 4;
+    const vpc = new ec2.Vpc(scope, "Vpc", {
+      ipAddresses: ec2.IpAddresses.cidr(vpcCidr),
+      enableDnsSupport: true,
+      enableDnsHostnames: true,
+      vpcName: resourceName,
+      subnetConfiguration: [
+        {
+          name: "Public",
+          subnetType: ec2.SubnetType.PUBLIC,
+          cidrMask: subnetCIDRMask,
+        },
+        {
+          name: "Private",
+          subnetType: ec2.SubnetType.PRIVATE_WITH_EGRESS,
+          cidrMask: subnetCIDRMask,
+        },
+      ],
+    });
+
+    return vpc;
+  }
+}
diff --git a/src/core/index.ts b/src/core/index.ts
index 1506fd4d..990dcc53 100644
--- a/src/core/index.ts
+++ b/src/core/index.ts
@@ -1,3 +1,4 @@
+export * from "./airflow-pipeline";
 export * from "./glue-factory";
 export * from "./kms-factory";
 export * from "./s3-factory";


Test patch:
diff --git a/integ/integ.airflow-pipeline.test.ts b/integ/integ.airflow-pipeline.test.ts
new file mode 100644
index 00000000..f10c7548
--- /dev/null
+++ b/integ/integ.airflow-pipeline.test.ts
@@ -0,0 +1,43 @@
+import * as integration from "@aws-cdk/integ-tests-alpha";
+import * as cdk from "aws-cdk-lib";
+import { Construct } from "constructs";
+import { RequireApproval } from "aws-cdk-lib/cloud-assembly-schema";
+
+import { AirflowDataPipeline } from "../src";
+
+interface AirflowDataPipelineTestStackProps extends cdk.StackProps {}
+
+class AirflowDataPipelineTestStack extends cdk.Stack {
+  constructor(scope: Construct, id: string, props: AirflowDataPipelineTestStackProps) {
+    super(scope, id, props);
+
+    new AirflowDataPipeline(this, "airflow-pipeline", {
+      vpcCidr: "10.55.0.0/16",
+      name: "integ-test",
+      maxWorkers: 5,
+      dagFiles: ["./dags/"],
+    });
+  }
+}
+
+const app = new cdk.App();
+new integration.IntegTest(app, "Airflow Data Pipeline Stack Integration Tests", {
+    testCases: [
+      new AirflowDataPipelineTestStack(app, "AirflowDataPipeline", {}),
+    ],
+    diffAssets: true,
+    stackUpdateWorkflow: true,
+    cdkCommandOptions: {
+      deploy: {
+        args: {
+          requireApproval: RequireApproval.NEVER,
+          json: true,
+        },
+      },
+      destroy: {
+        args: {
+          force: true,
+        },
+      },
+    },
+});
diff --git a/test/airflow-pipeline.test.ts b/test/airflow-pipeline.test.ts
new file mode 100644
index 00000000..3076102d
--- /dev/null
+++ b/test/airflow-pipeline.test.ts
@@ -0,0 +1,149 @@
+import * as cdk from "aws-cdk-lib";
+import { Match, Template } from "aws-cdk-lib/assertions";
+import * as iam from "aws-cdk-lib/aws-iam";
+import * as s3 from "aws-cdk-lib/aws-s3";
+
+import { AirflowDataPipeline } from "../src";
+
+test("AirflowDataPipeline Basic", () => {
+  const stack = new cdk.Stack();
+
+  new AirflowDataPipeline(stack, "airflow-pipeline", {
+    name: "MWAAEnvironment",
+    vpcCidr: "10.44.0.0/16",
+  });
+  const template = Template.fromStack(stack);
+  template.hasResourceProperties("AWS::MWAA::Environment", {
+    Name: "MWAAEnvironment",
+    NetworkConfiguration: {
+      SecurityGroupIds: Match.anyValue(),
+      SubnetIds: [Match.anyValue(), Match.anyValue()],
+    },
+  });
+  template.resourceCountIs("AWS::IAM::Role", 1);
+  template.hasResourceProperties("AWS::S3::Bucket", {});
+  template.hasResourceProperties("AWS::EC2::VPC", {
+    CidrBlock: "10.44.0.0/16",
+  });
+  template.resourceCountIs("AWS::EC2::Subnet", 4);
+  template.resourceCountIs("AWS::EC2::NatGateway", 2);
+  template.resourceCountIs("AWS::EC2::Route", 4);
+  template.resourceCountIs("AWS::EC2::InternetGateway", 1);
+  template.resourceCountIs("AWS::EC2::VPCGatewayAttachment", 1);
+});
+
+test("AirflowDataPipeline Existing Vpc & Bucket", () => {
+  const app = new cdk.App();
+  const stack = new cdk.Stack(app, "dummy-stack", {
+    env: {
+      account: "123456789012",
+      region: "us-east-1",
+    },
+  });
+  new AirflowDataPipeline(stack, "airflow-pipeline", {
+    name: "MWAAEnvironment",
+    vpcId: "vpc-01234567",
+    s3Bucket: new s3.Bucket(stack, "my-dummy-bucket", {}),
+  });
+
+  const template = Template.fromStack(stack);
+  template.hasResourceProperties("AWS::MWAA::Environment", {});
+  template.hasResourceProperties("AWS::IAM::Role", {});
+  template.resourceCountIs("AWS::EC2::VPC", 0);
+});
+
+test("Vpc Cidr or Id must be provided", () => {
+  const stack = new cdk.Stack();
+  expect(() => {
+    new AirflowDataPipeline(stack, "Stage", { name: "MWAAEnvironment" });
+  }).toThrowError("One of 'vpcId' or 'vpcCidr' must be provided");
+});
+
+test("Invalid Vpc Cidr: Too Large", () => {
+  const stack = new cdk.Stack();
+  expect(() => {
+    new AirflowDataPipeline(stack, "Stage", {
+      name: "MWAAEnvironment",
+      vpcCidr: "10.0.0.0/15",
+    });
+  }).toThrowError("Vpc Cidr Range must of size >=16 and <=20");
+});
+
+test("Invalid Vpc Cidr: Too Small", () => {
+  const stack = new cdk.Stack();
+  expect(() => {
+    new AirflowDataPipeline(stack, "Stage", {
+      name: "MWAAEnvironment",
+      vpcCidr: "10.0.0.0/21",
+    });
+  }).toThrowError("Vpc Cidr Range must of size >=16 and <=20");
+});
+
+test("Plugin Path not specified", () => {
+  const app = new cdk.App();
+  const stack = new cdk.Stack(app, "dummy-stack", {
+    env: {
+      account: "123456789012",
+      region: "us-east-1",
+    },
+  });
+  expect(() => {
+    new AirflowDataPipeline(stack, "Stage", {
+      name: "MWAAEnvironment",
+      vpcId: "vpc-xxxxxx",
+      pluginFile: "./test/mwaa/plugins/plugins.zip",
+    });
+  }).toThrowError("'pluginsS3Path' must be specified if a 'pluginFile' is specified.");
+});
+
+test("Requirements Path not specified", () => {
+  const app = new cdk.App();
+  const stack = new cdk.Stack(app, "dummy-stack", {
+    env: {
+      account: "123456789012",
+      region: "us-east-1",
+    },
+  });
+  expect(() => {
+    new AirflowDataPipeline(stack, "Stage", {
+      name: "MWAAEnvironment",
+      vpcId: "vpc-xxxxxx",
+      requirementsFile: "./test/mwaa/requirements.txt",
+    });
+  }).toThrowError("'requirementsS3Path' must be specified if a 'requirementsFile' is specified.");
+});
+
+test("AirflowDataPipeline All Settings", () => {
+  const stack = new cdk.Stack();
+
+  new AirflowDataPipeline(stack, "airflow-pipeline", {
+    vpcCidr: "10.44.0.0/16",
+    name: "my-environment",
+    maxWorkers: 5,
+    dagProcessingLogs: "ERROR",
+    schedulerLogsLevel: "ERROR",
+    taskLogsLevel: "ERROR",
+    workerLogsLevel: "ERROR",
+    webserverLogsLevel: "ERROR",
+    dagS3Path: "foobar",
+    dagFiles: ["./test/mwaa/dags/"],
+    pluginFile: "./test/mwaa/plugins/plugins.zip",
+    pluginsS3Path: "plugins/plugins.zip",
+    requirementsFile: "./test/mwaa/requirements.txt",
+    requirementsS3Path: "requirements.txt",
+    additionalPolicyStatements: [
+      new iam.PolicyStatement({
+        actions: ["airflow:CreateCliToken"],
+        resources: ["*"],
+      }),
+    ],
+  });
+
+  const template = Template.fromStack(stack);
+  template.hasResourceProperties("AWS::MWAA::Environment", {
+    DagS3Path: "foobar",
+  });
+  template.hasResourceProperties("AWS::IAM::Role", {});
+  template.hasResourceProperties("AWS::S3::Bucket", {});
+  template.hasResourceProperties("AWS::EC2::VPC", {});
+});
diff --git a/test/mwaa/dags/bar.py b/test/mwaa/dags/bar.py
new file mode 100644
index 00000000..fc80254b
--- /dev/null
+++ b/test/mwaa/dags/bar.py
@@ -0,0 +1,1 @@
+pass
\ No newline at end of file
diff --git a/test/mwaa/dags/foo.py b/test/mwaa/dags/foo.py
new file mode 100644
index 00000000..fc80254b
--- /dev/null
+++ b/test/mwaa/dags/foo.py
@@ -0,0 +1,1 @@
+pass
\ No newline at end of file
diff --git a/test/mwaa/plugins/plugins.zip b/test/mwaa/plugins/plugins.zip
new file mode 100644
index 00000000..e69de29b
diff --git a/test/mwaa/requirements.txt b/test/mwaa/requirements.txt
new file mode 100644
index 00000000..e69de29b
diff --git a/test/sqs-lambda-stage.test.ts b/test/sqs-lambda-stage.test.ts
index 3993f6e9..8eaf374b 100644
--- a/test/sqs-lambda-stage.test.ts
+++ b/test/sqs-lambda-stage.test.ts
@@ -209,6 +209,7 @@ test("SQSToLambdaStage additional properties", () => {
       code: lambda.Code.fromAsset(path.join(__dirname, "/../src/")),
       handler: "commons.handlers.lambda_handler",
       runtime: lambda.Runtime.PYTHON_3_8,
+      timeout: cdk.Duration.minutes(8),
     },
     messageGroupId: "dummy-group",
     maxReceiveCount: 2,
@@ -225,6 +226,7 @@ test("SQSToLambdaStage additional properties", () => {
   template.hasResourceProperties("AWS::Lambda::Function", {
     Runtime: "python3.8",
     FunctionName: "dummy-function",
+    Timeout: 480,
   });
   template.resourceCountIs("AWS::SQS::Queue", 2);
   template.hasResourceProperties("AWS::SQS::Queue", {
@@ -232,3 +234,31 @@ test("SQSToLambdaStage additional properties", () => {
     QueueName: "dummy-queue.fifo",
   });
 });
+
+test("SQSToLambdaStage DLQ No FIFO", () => {
+  const stack = new cdk.Stack();
+
+  new SqsToLambdaStage(stack, "Stage", {
+    lambdaFunctionProps: {
+      functionName: "dummy-function",
+      code: lambda.Code.fromAsset(path.join(__dirname, "/../src/")),
+      handler: "commons.handlers.lambda_handler",
+      runtime: lambda.Runtime.PYTHON_3_8,
+    },
+    messageGroupId: "dummy-group",
+    maxReceiveCount: 2,
+    dlqEnabled: true,
+    sqsQueueProps: {
+      queueName: "dummy-queue",
+      visibilityTimeout: cdk.Duration.minutes(5),
+    },
+  });
+
+  const template = Template.fromStack(stack);
+
+  template.resourceCountIs("AWS::SQS::Queue", 2);
+  template.hasResourceProperties("AWS::SQS::Queue", {
+    VisibilityTimeout: 300,
+    QueueName: "dummy-queue",
+  });
+});