We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Use Python 3.8 across the board
We've already encountered some issues with Python 3.6 (cattrs no longer supporting it and less support for type annotations), and we have no reason to support multiple Python versions in production.

To do:

- [x] Set Python version in `.python-version`.
- [ ] Use a single version in CI, removing the need for the strategy matrix.
- [ ] Use the `PYTHON_3_8` runtime for AWS Lambda jobs.
- [ ] Change the version in `pyproject.toml` to `^3.8,<3.9`.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `infra/datalake/api_stack.py`
Content:
```
1 """
2 Data Lake AWS resources definitions.
3 """
4 from aws_cdk import aws_lambda, core
5 from aws_cdk.core import Tags
6 
7 
8 class APIStack(core.Stack):
9     """Data Lake stack definition."""
10 
11     def __init__(self, scope: core.Construct, stack_id: str, datasets_table, **kwargs) -> None:
12         super().__init__(scope, stack_id, **kwargs)
13 
14         ############################################################################################
15         # ### API ENDPOINTS ########################################################################
16         ############################################################################################
17 
18         endpoints = ("datasets",)
19 
20         for endpoint in endpoints:
21             endpoint_function = aws_lambda.Function(
22                 self,
23                 f"{endpoint}-endpoint-function",
24                 function_name=f"{endpoint}-endpoint",
25                 handler=f"endpoints.{endpoint}.entrypoint.lambda_handler",
26                 runtime=aws_lambda.Runtime.PYTHON_3_6,
27                 code=aws_lambda.Code.from_asset(
28                     path="..",
29                     bundling=core.BundlingOptions(
30                         # pylint:disable=no-member
31                         image=aws_lambda.Runtime.PYTHON_3_6.bundling_docker_image,
32                         command=["backend/bundle.bash", f"endpoints/{endpoint}"],
33                     ),
34                 ),
35             )
36 
37             datasets_table.grant_read_write_data(endpoint_function)
38             datasets_table.grant(
39                 endpoint_function, "dynamodb:DescribeTable"
40             )  # required by pynamodb
41 
42             Tags.of(endpoint_function).add("ApplicationLayer", "api")
43
```

Path: `infra/datalake/processing_stack.py`
Content:
```
1 """
2 Data Lake processing stack.
3 """
4 import textwrap
5 
6 from aws_cdk import (
7     aws_batch,
8     aws_ec2,
9     aws_ecs,
10     aws_iam,
11     aws_lambda,
12     aws_stepfunctions,
13     aws_stepfunctions_tasks,
14     core,
15 )
16 from aws_cdk.core import Tags
17 
18 
19 class ProcessingStack(core.Stack):
20     """Data Lake processing stack definition."""
21 
22     # pylint: disable=too-many-locals
23     def __init__(self, scope: core.Construct, stack_id: str, deploy_env, vpc, **kwargs) -> None:
24         super().__init__(scope, stack_id, **kwargs)
25 
26         ############################################################################################
27         # ### DATASET VERSION CREATE ###############################################################
28         ############################################################################################
29 
30         # STATE MACHINE TASKS CONFIGURATION
31         # * type: lambda|batch
32         # * parallel: True|False
33         # * input_path: "$"
34         # * output_path: "$"
35         # * result_path: "$"
36         # * items_path: "$"
37         creation_tasks = {}
38 
39         creation_tasks["content_iterator"] = {"type": "lambda", "result_path": "$.content"}
40 
41         creation_tasks["validation_summary"] = {"type": "lambda", "result_path": "$.validation"}
42 
43         creation_tasks["validation_failure"] = {
44             "type": "lambda",
45             "result_path": aws_stepfunctions.JsonPath.DISCARD,
46         }
47 
48         creation_tasks["check_flat_directory_structure"] = {
49             "type": "batch",
50             "parallel": False,
51             "result_path": aws_stepfunctions.JsonPath.DISCARD,
52         }
53 
54         creation_tasks["check_files_checksums"] = {
55             "type": "batch",
56             "parallel": True,
57             "result_path": aws_stepfunctions.JsonPath.DISCARD,
58         }
59 
60         # AWS BATCH COMPUTE ENVIRONMENT
61         batch_service_role = aws_iam.Role(
62             self,
63             "batch-service-role",
64             assumed_by=aws_iam.ServicePrincipal("batch.amazonaws.com"),
65             managed_policies=[
66                 aws_iam.ManagedPolicy.from_aws_managed_policy_name(
67                     "service-role/AWSBatchServiceRole"
68                 ),
69             ],
70         )
71 
72         batch_instance_role = aws_iam.Role(
73             self,
74             "batch-instance-role",
75             assumed_by=aws_iam.ServicePrincipal("ec2.amazonaws.com"),
76             managed_policies=[
77                 aws_iam.ManagedPolicy.from_aws_managed_policy_name(
78                     "service-role/AmazonEC2ContainerServiceforEC2Role"
79                 ),
80             ],
81         )
82 
83         batch_instance_profile = aws_iam.CfnInstanceProfile(
84             self,
85             "batch-instance-profile",
86             roles=[
87                 batch_instance_role.role_name,
88             ],
89         )
90 
91         batch_launch_template_data = textwrap.dedent(
92             """
93             MIME-Version: 1.0
94             Content-Type: multipart/mixed; boundary="==MYBOUNDARY=="
95 
96             --==MYBOUNDARY==
97             Content-Type: text/x-shellscript; charset="us-ascii"
98 
99             #!/bin/bash
100             echo ECS_IMAGE_PULL_BEHAVIOR=prefer-cached >> /etc/ecs/ecs.config
101 
102             --==MYBOUNDARY==--
103             """
104         )
105 
106         batch_launch_template = aws_ec2.CfnLaunchTemplate(
107             self,
108             "batch-launch-template",
109             launch_template_name="datalake-batch-launch-template",
110             launch_template_data={
111                 "userData": core.Fn.base64(batch_launch_template_data.strip()),
112             },
113         )
114 
115         if deploy_env == "prod":
116             instance_types = [
117                 aws_ec2.InstanceType("c5.xlarge"),
118                 aws_ec2.InstanceType("c5.2xlarge"),
119                 aws_ec2.InstanceType("c5.4xlarge"),
120                 aws_ec2.InstanceType("c5.9xlarge"),
121             ]
122         else:
123             instance_types = [
124                 aws_ec2.InstanceType("m5.large"),
125                 aws_ec2.InstanceType("m5.xlarge"),
126             ]
127 
128         batch_compute_environment = aws_batch.ComputeEnvironment(
129             self,
130             "compute-environment",
131             compute_resources=aws_batch.ComputeResources(
132                 vpc=vpc,
133                 # vpc_subnets=vpc.select_subnets(subnet_group_name="ecs-cluster"),  # TODO
134                 minv_cpus=0,
135                 desiredv_cpus=0,
136                 maxv_cpus=1000,
137                 instance_types=instance_types,
138                 instance_role=batch_instance_profile.instance_profile_name,
139                 allocation_strategy=aws_batch.AllocationStrategy("BEST_FIT_PROGRESSIVE"),
140                 launch_template=aws_batch.LaunchTemplateSpecification(
141                     launch_template_name=batch_launch_template.launch_template_name
142                 ),
143             ),
144             service_role=batch_service_role,
145         )
146 
147         batch_job_queue = aws_batch.JobQueue(
148             self,
149             "dataset-version-creation-queue",
150             compute_environments=[
151                 aws_batch.JobQueueComputeEnvironment(
152                     compute_environment=batch_compute_environment, order=10
153                 ),
154             ],
155             priority=10,
156         )
157 
158         # LAMBDA AND AWS BATCH BUNDLING AND STATE MACHINE TASKS CREATION
159         step_tasks = {}
160         for task_name in creation_tasks:
161 
162             task = creation_tasks[task_name]
163 
164             # lambda functions
165             if task["type"] == "lambda":
166 
167                 lambda_function = aws_lambda.Function(
168                     self,
169                     f"{task_name}-function",
170                     handler=f"processing.{task_name}.task.lambda_handler",
171                     runtime=aws_lambda.Runtime.PYTHON_3_6,
172                     code=aws_lambda.Code.from_asset(
173                         path="..",
174                         bundling=core.BundlingOptions(
175                             # pylint:disable=no-member
176                             image=aws_lambda.Runtime.PYTHON_3_6.bundling_docker_image,
177                             command=["backend/bundle.bash", f"processing/{task_name}"],
178                         ),
179                     ),
180                 )
181 
182                 step_tasks[task_name] = aws_stepfunctions_tasks.LambdaInvoke(
183                     self,
184                     task_name,
185                     lambda_function=lambda_function,
186                     input_path=task.get("input_path", "$"),
187                     output_path=task.get("output_path", "$"),
188                     result_path=task.get("result_path", "$"),
189                     payload_response_only=True,
190                 )
191 
192                 Tags.of(lambda_function).add("ApplicationLayer", "data-processing")
193 
194             # aws batch jobs
195             if task["type"] == "batch":
196 
197                 job_definition = aws_batch.JobDefinition(
198                     self,
199                     f"{task_name}-job",
200                     container=aws_batch.JobDefinitionContainer(
201                         image=aws_ecs.ContainerImage.from_asset(
202                             directory=f"../backend/processing/{task_name}",
203                         ),
204                         memory_limit_mib=3900 if deploy_env == "prod" else 500,
205                         vcpus=1,
206                     ),
207                     retry_attempts=4,
208                 )
209 
210                 job_command = [
211                     "--dataset-id",
212                     "Ref::dataset_id",
213                     "--version-id",
214                     "Ref::version_id",
215                     "--type",
216                     "Ref::type",
217                     "--metadata-url",
218                     "Ref::metadata_url",
219                     "--dataset-id",
220                 ]
221                 job_environment = {"BATCH_JOB_FIRST_ITEM_INDEX": "Ref::first_item"}
222 
223                 job_payload_data = {
224                     "dataset_id.$": "$.dataset_id",
225                     "version_id.$": "$.version_id",
226                     "type.$": "$.type",
227                     "metadata_url.$": "$.metadata_url",
228                 }
229                 job_payload_data_parallel = {"first_item.$": "$.content.first_item"}
230                 job_payload_single = aws_stepfunctions.TaskInput.from_object(job_payload_data)
231                 job_payload_parallel = aws_stepfunctions.TaskInput.from_object(
232                     {**job_payload_data, **job_payload_data_parallel}
233                 )
234 
235                 if task["parallel"]:
236                     step_tasks[task_name] = aws_stepfunctions_tasks.BatchSubmitJob(
237                         self,
238                         task_name,
239                         job_name=f"{task_name}-job",
240                         job_definition=job_definition,
241                         job_queue=batch_job_queue,
242                         array_size=aws_stepfunctions.JsonPath.number_at("$.content.iteration_size"),
243                         input_path=task.get("input_path", "$"),
244                         output_path=task.get("output_path", "$"),
245                         result_path=task.get("result_path", "$"),
246                         container_overrides=aws_stepfunctions_tasks.BatchContainerOverrides(
247                             command=job_command,
248                             environment=job_environment,
249                         ),
250                         payload=job_payload_parallel,
251                     )
252 
253                 else:
254                     step_tasks[task_name] = aws_stepfunctions_tasks.BatchSubmitJob(
255                         self,
256                         task_name,
257                         job_name=f"{task_name}-job",
258                         job_definition=job_definition,
259                         job_queue=batch_job_queue,
260                         input_path=task.get("input_path", "$"),
261                         output_path=task.get("output_path", "$"),
262                         result_path=task.get("result_path", "$"),
263                         container_overrides=aws_stepfunctions_tasks.BatchContainerOverrides(
264                             command=job_command,
265                         ),
266                         payload=job_payload_single,
267                     )
268 
269         # success task
270         step_tasks["success"] = aws_stepfunctions.Succeed(
271             self,
272             "success",
273         )
274 
275         # STATE MACHINE
276         # state machine definition
277         dataset_version_creation_definition = (
278             step_tasks["check_flat_directory_structure"]
279             .next(step_tasks["content_iterator"])
280             .next(step_tasks["check_files_checksums"])
281             .next(
282                 aws_stepfunctions.Choice(self, "content_iteration_finished")
283                 .when(
284                     aws_stepfunctions.Condition.not_(
285                         aws_stepfunctions.Condition.number_equals("$.content.next_item", -1)
286                     ),
287                     step_tasks["content_iterator"],
288                 )
289                 .otherwise(
290                     step_tasks["validation_summary"].next(
291                         aws_stepfunctions.Choice(self, "validation_successful")
292                         .when(
293                             aws_stepfunctions.Condition.boolean_equals(
294                                 "$.validation.success", True
295                             ),
296                             step_tasks["success"],
297                         )
298                         .otherwise(step_tasks["validation_failure"])
299                     ),
300                 )
301             )
302         )
303 
304         # state machine
305         creation_process = aws_stepfunctions.StateMachine(  # pylint:disable=unused-variable
306             self,
307             "dataset-version-creation",
308             definition=dataset_version_creation_definition,
309         )
310
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/infra/datalake/api_stack.py b/infra/datalake/api_stack.py
--- a/infra/datalake/api_stack.py
+++ b/infra/datalake/api_stack.py
@@ -23,12 +23,12 @@
                 f"{endpoint}-endpoint-function",
                 function_name=f"{endpoint}-endpoint",
                 handler=f"endpoints.{endpoint}.entrypoint.lambda_handler",
-                runtime=aws_lambda.Runtime.PYTHON_3_6,
+                runtime=aws_lambda.Runtime.PYTHON_3_8,
                 code=aws_lambda.Code.from_asset(
                     path="..",
                     bundling=core.BundlingOptions(
                         # pylint:disable=no-member
-                        image=aws_lambda.Runtime.PYTHON_3_6.bundling_docker_image,
+                        image=aws_lambda.Runtime.PYTHON_3_8.bundling_docker_image,
                         command=["backend/bundle.bash", f"endpoints/{endpoint}"],
                     ),
                 ),
diff --git a/infra/datalake/processing_stack.py b/infra/datalake/processing_stack.py
--- a/infra/datalake/processing_stack.py
+++ b/infra/datalake/processing_stack.py
@@ -168,12 +168,12 @@
                     self,
                     f"{task_name}-function",
                     handler=f"processing.{task_name}.task.lambda_handler",
-                    runtime=aws_lambda.Runtime.PYTHON_3_6,
+                    runtime=aws_lambda.Runtime.PYTHON_3_8,
                     code=aws_lambda.Code.from_asset(
                         path="..",
                         bundling=core.BundlingOptions(
                             # pylint:disable=no-member
-                            image=aws_lambda.Runtime.PYTHON_3_6.bundling_docker_image,
+                            image=aws_lambda.Runtime.PYTHON_3_8.bundling_docker_image,
                             command=["backend/bundle.bash", f"processing/{task_name}"],
                         ),
                     ),

{"golden_diff": "diff --git a/infra/datalake/api_stack.py b/infra/datalake/api_stack.py\n--- a/infra/datalake/api_stack.py\n+++ b/infra/datalake/api_stack.py\n@@ -23,12 +23,12 @@\n                 f\"{endpoint}-endpoint-function\",\n                 function_name=f\"{endpoint}-endpoint\",\n                 handler=f\"endpoints.{endpoint}.entrypoint.lambda_handler\",\n-                runtime=aws_lambda.Runtime.PYTHON_3_6,\n+                runtime=aws_lambda.Runtime.PYTHON_3_8,\n                 code=aws_lambda.Code.from_asset(\n                     path=\"..\",\n                     bundling=core.BundlingOptions(\n                         # pylint:disable=no-member\n-                        image=aws_lambda.Runtime.PYTHON_3_6.bundling_docker_image,\n+                        image=aws_lambda.Runtime.PYTHON_3_8.bundling_docker_image,\n                         command=[\"backend/bundle.bash\", f\"endpoints/{endpoint}\"],\n                     ),\n                 ),\ndiff --git a/infra/datalake/processing_stack.py b/infra/datalake/processing_stack.py\n--- a/infra/datalake/processing_stack.py\n+++ b/infra/datalake/processing_stack.py\n@@ -168,12 +168,12 @@\n                     self,\n                     f\"{task_name}-function\",\n                     handler=f\"processing.{task_name}.task.lambda_handler\",\n-                    runtime=aws_lambda.Runtime.PYTHON_3_6,\n+                    runtime=aws_lambda.Runtime.PYTHON_3_8,\n                     code=aws_lambda.Code.from_asset(\n                         path=\"..\",\n                         bundling=core.BundlingOptions(\n                             # pylint:disable=no-member\n-                            image=aws_lambda.Runtime.PYTHON_3_6.bundling_docker_image,\n+                            image=aws_lambda.Runtime.PYTHON_3_8.bundling_docker_image,\n                             command=[\"backend/bundle.bash\", f\"processing/{task_name}\"],\n                         ),\n                     ),\n", "issue": "Use Python 3.8 across the board\nWe've already encountered some issues with Python 3.6 (cattrs no longer supporting it and less support for type annotations), and we have no reason to support multiple Python versions in production.\r\n\r\nTo do:\r\n\r\n- [x] Set Python version in `.python-version`.\r\n- [ ] Use a single version in CI, removing the need for the strategy matrix.\r\n- [ ] Use the `PYTHON_3_8` runtime for AWS Lambda jobs.\r\n- [ ] Change the version in `pyproject.toml` to `^3.8,<3.9`.\n", "before_files": [{"content": "\"\"\"\nData Lake AWS resources definitions.\n\"\"\"\nfrom aws_cdk import aws_lambda, core\nfrom aws_cdk.core import Tags\n\n\nclass APIStack(core.Stack):\n    \"\"\"Data Lake stack definition.\"\"\"\n\n    def __init__(self, scope: core.Construct, stack_id: str, datasets_table, **kwargs) -> None:\n        super().__init__(scope, stack_id, **kwargs)\n\n        ############################################################################################\n        # ### API ENDPOINTS ########################################################################\n        ############################################################################################\n\n        endpoints = (\"datasets\",)\n\n        for endpoint in endpoints:\n            endpoint_function = aws_lambda.Function(\n                self,\n                f\"{endpoint}-endpoint-function\",\n                function_name=f\"{endpoint}-endpoint\",\n                handler=f\"endpoints.{endpoint}.entrypoint.lambda_handler\",\n                runtime=aws_lambda.Runtime.PYTHON_3_6,\n                code=aws_lambda.Code.from_asset(\n                    path=\"..\",\n                    bundling=core.BundlingOptions(\n                        # pylint:disable=no-member\n                        image=aws_lambda.Runtime.PYTHON_3_6.bundling_docker_image,\n                        command=[\"backend/bundle.bash\", f\"endpoints/{endpoint}\"],\n                    ),\n                ),\n            )\n\n            datasets_table.grant_read_write_data(endpoint_function)\n            datasets_table.grant(\n                endpoint_function, \"dynamodb:DescribeTable\"\n            )  # required by pynamodb\n\n            Tags.of(endpoint_function).add(\"ApplicationLayer\", \"api\")\n", "path": "infra/datalake/api_stack.py"}, {"content": "\"\"\"\nData Lake processing stack.\n\"\"\"\nimport textwrap\n\nfrom aws_cdk import (\n    aws_batch,\n    aws_ec2,\n    aws_ecs,\n    aws_iam,\n    aws_lambda,\n    aws_stepfunctions,\n    aws_stepfunctions_tasks,\n    core,\n)\nfrom aws_cdk.core import Tags\n\n\nclass ProcessingStack(core.Stack):\n    \"\"\"Data Lake processing stack definition.\"\"\"\n\n    # pylint: disable=too-many-locals\n    def __init__(self, scope: core.Construct, stack_id: str, deploy_env, vpc, **kwargs) -> None:\n        super().__init__(scope, stack_id, **kwargs)\n\n        ############################################################################################\n        # ### DATASET VERSION CREATE ###############################################################\n        ############################################################################################\n\n        # STATE MACHINE TASKS CONFIGURATION\n        # * type: lambda|batch\n        # * parallel: True|False\n        # * input_path: \"$\"\n        # * output_path: \"$\"\n        # * result_path: \"$\"\n        # * items_path: \"$\"\n        creation_tasks = {}\n\n        creation_tasks[\"content_iterator\"] = {\"type\": \"lambda\", \"result_path\": \"$.content\"}\n\n        creation_tasks[\"validation_summary\"] = {\"type\": \"lambda\", \"result_path\": \"$.validation\"}\n\n        creation_tasks[\"validation_failure\"] = {\n            \"type\": \"lambda\",\n            \"result_path\": aws_stepfunctions.JsonPath.DISCARD,\n        }\n\n        creation_tasks[\"check_flat_directory_structure\"] = {\n            \"type\": \"batch\",\n            \"parallel\": False,\n            \"result_path\": aws_stepfunctions.JsonPath.DISCARD,\n        }\n\n        creation_tasks[\"check_files_checksums\"] = {\n            \"type\": \"batch\",\n            \"parallel\": True,\n            \"result_path\": aws_stepfunctions.JsonPath.DISCARD,\n        }\n\n        # AWS BATCH COMPUTE ENVIRONMENT\n        batch_service_role = aws_iam.Role(\n            self,\n            \"batch-service-role\",\n            assumed_by=aws_iam.ServicePrincipal(\"batch.amazonaws.com\"),\n            managed_policies=[\n                aws_iam.ManagedPolicy.from_aws_managed_policy_name(\n                    \"service-role/AWSBatchServiceRole\"\n                ),\n            ],\n        )\n\n        batch_instance_role = aws_iam.Role(\n            self,\n            \"batch-instance-role\",\n            assumed_by=aws_iam.ServicePrincipal(\"ec2.amazonaws.com\"),\n            managed_policies=[\n                aws_iam.ManagedPolicy.from_aws_managed_policy_name(\n                    \"service-role/AmazonEC2ContainerServiceforEC2Role\"\n                ),\n            ],\n        )\n\n        batch_instance_profile = aws_iam.CfnInstanceProfile(\n            self,\n            \"batch-instance-profile\",\n            roles=[\n                batch_instance_role.role_name,\n            ],\n        )\n\n        batch_launch_template_data = textwrap.dedent(\n            \"\"\"\n            MIME-Version: 1.0\n            Content-Type: multipart/mixed; boundary=\"==MYBOUNDARY==\"\n\n            --==MYBOUNDARY==\n            Content-Type: text/x-shellscript; charset=\"us-ascii\"\n\n            #!/bin/bash\n            echo ECS_IMAGE_PULL_BEHAVIOR=prefer-cached >> /etc/ecs/ecs.config\n\n            --==MYBOUNDARY==--\n            \"\"\"\n        )\n\n        batch_launch_template = aws_ec2.CfnLaunchTemplate(\n            self,\n            \"batch-launch-template\",\n            launch_template_name=\"datalake-batch-launch-template\",\n            launch_template_data={\n                \"userData\": core.Fn.base64(batch_launch_template_data.strip()),\n            },\n        )\n\n        if deploy_env == \"prod\":\n            instance_types = [\n                aws_ec2.InstanceType(\"c5.xlarge\"),\n                aws_ec2.InstanceType(\"c5.2xlarge\"),\n                aws_ec2.InstanceType(\"c5.4xlarge\"),\n                aws_ec2.InstanceType(\"c5.9xlarge\"),\n            ]\n        else:\n            instance_types = [\n                aws_ec2.InstanceType(\"m5.large\"),\n                aws_ec2.InstanceType(\"m5.xlarge\"),\n            ]\n\n        batch_compute_environment = aws_batch.ComputeEnvironment(\n            self,\n            \"compute-environment\",\n            compute_resources=aws_batch.ComputeResources(\n                vpc=vpc,\n                # vpc_subnets=vpc.select_subnets(subnet_group_name=\"ecs-cluster\"),  # TODO\n                minv_cpus=0,\n                desiredv_cpus=0,\n                maxv_cpus=1000,\n                instance_types=instance_types,\n                instance_role=batch_instance_profile.instance_profile_name,\n                allocation_strategy=aws_batch.AllocationStrategy(\"BEST_FIT_PROGRESSIVE\"),\n                launch_template=aws_batch.LaunchTemplateSpecification(\n                    launch_template_name=batch_launch_template.launch_template_name\n                ),\n            ),\n            service_role=batch_service_role,\n        )\n\n        batch_job_queue = aws_batch.JobQueue(\n            self,\n            \"dataset-version-creation-queue\",\n            compute_environments=[\n                aws_batch.JobQueueComputeEnvironment(\n                    compute_environment=batch_compute_environment, order=10\n                ),\n            ],\n            priority=10,\n        )\n\n        # LAMBDA AND AWS BATCH BUNDLING AND STATE MACHINE TASKS CREATION\n        step_tasks = {}\n        for task_name in creation_tasks:\n\n            task = creation_tasks[task_name]\n\n            # lambda functions\n            if task[\"type\"] == \"lambda\":\n\n                lambda_function = aws_lambda.Function(\n                    self,\n                    f\"{task_name}-function\",\n                    handler=f\"processing.{task_name}.task.lambda_handler\",\n                    runtime=aws_lambda.Runtime.PYTHON_3_6,\n                    code=aws_lambda.Code.from_asset(\n                        path=\"..\",\n                        bundling=core.BundlingOptions(\n                            # pylint:disable=no-member\n                            image=aws_lambda.Runtime.PYTHON_3_6.bundling_docker_image,\n                            command=[\"backend/bundle.bash\", f\"processing/{task_name}\"],\n                        ),\n                    ),\n                )\n\n                step_tasks[task_name] = aws_stepfunctions_tasks.LambdaInvoke(\n                    self,\n                    task_name,\n                    lambda_function=lambda_function,\n                    input_path=task.get(\"input_path\", \"$\"),\n                    output_path=task.get(\"output_path\", \"$\"),\n                    result_path=task.get(\"result_path\", \"$\"),\n                    payload_response_only=True,\n                )\n\n                Tags.of(lambda_function).add(\"ApplicationLayer\", \"data-processing\")\n\n            # aws batch jobs\n            if task[\"type\"] == \"batch\":\n\n                job_definition = aws_batch.JobDefinition(\n                    self,\n                    f\"{task_name}-job\",\n                    container=aws_batch.JobDefinitionContainer(\n                        image=aws_ecs.ContainerImage.from_asset(\n                            directory=f\"../backend/processing/{task_name}\",\n                        ),\n                        memory_limit_mib=3900 if deploy_env == \"prod\" else 500,\n                        vcpus=1,\n                    ),\n                    retry_attempts=4,\n                )\n\n                job_command = [\n                    \"--dataset-id\",\n                    \"Ref::dataset_id\",\n                    \"--version-id\",\n                    \"Ref::version_id\",\n                    \"--type\",\n                    \"Ref::type\",\n                    \"--metadata-url\",\n                    \"Ref::metadata_url\",\n                    \"--dataset-id\",\n                ]\n                job_environment = {\"BATCH_JOB_FIRST_ITEM_INDEX\": \"Ref::first_item\"}\n\n                job_payload_data = {\n                    \"dataset_id.$\": \"$.dataset_id\",\n                    \"version_id.$\": \"$.version_id\",\n                    \"type.$\": \"$.type\",\n                    \"metadata_url.$\": \"$.metadata_url\",\n                }\n                job_payload_data_parallel = {\"first_item.$\": \"$.content.first_item\"}\n                job_payload_single = aws_stepfunctions.TaskInput.from_object(job_payload_data)\n                job_payload_parallel = aws_stepfunctions.TaskInput.from_object(\n                    {**job_payload_data, **job_payload_data_parallel}\n                )\n\n                if task[\"parallel\"]:\n                    step_tasks[task_name] = aws_stepfunctions_tasks.BatchSubmitJob(\n                        self,\n                        task_name,\n                        job_name=f\"{task_name}-job\",\n                        job_definition=job_definition,\n                        job_queue=batch_job_queue,\n                        array_size=aws_stepfunctions.JsonPath.number_at(\"$.content.iteration_size\"),\n                        input_path=task.get(\"input_path\", \"$\"),\n                        output_path=task.get(\"output_path\", \"$\"),\n                        result_path=task.get(\"result_path\", \"$\"),\n                        container_overrides=aws_stepfunctions_tasks.BatchContainerOverrides(\n                            command=job_command,\n                            environment=job_environment,\n                        ),\n                        payload=job_payload_parallel,\n                    )\n\n                else:\n                    step_tasks[task_name] = aws_stepfunctions_tasks.BatchSubmitJob(\n                        self,\n                        task_name,\n                        job_name=f\"{task_name}-job\",\n                        job_definition=job_definition,\n                        job_queue=batch_job_queue,\n                        input_path=task.get(\"input_path\", \"$\"),\n                        output_path=task.get(\"output_path\", \"$\"),\n                        result_path=task.get(\"result_path\", \"$\"),\n                        container_overrides=aws_stepfunctions_tasks.BatchContainerOverrides(\n                            command=job_command,\n                        ),\n                        payload=job_payload_single,\n                    )\n\n        # success task\n        step_tasks[\"success\"] = aws_stepfunctions.Succeed(\n            self,\n            \"success\",\n        )\n\n        # STATE MACHINE\n        # state machine definition\n        dataset_version_creation_definition = (\n            step_tasks[\"check_flat_directory_structure\"]\n            .next(step_tasks[\"content_iterator\"])\n            .next(step_tasks[\"check_files_checksums\"])\n            .next(\n                aws_stepfunctions.Choice(self, \"content_iteration_finished\")\n                .when(\n                    aws_stepfunctions.Condition.not_(\n                        aws_stepfunctions.Condition.number_equals(\"$.content.next_item\", -1)\n                    ),\n                    step_tasks[\"content_iterator\"],\n                )\n                .otherwise(\n                    step_tasks[\"validation_summary\"].next(\n                        aws_stepfunctions.Choice(self, \"validation_successful\")\n                        .when(\n                            aws_stepfunctions.Condition.boolean_equals(\n                                \"$.validation.success\", True\n                            ),\n                            step_tasks[\"success\"],\n                        )\n                        .otherwise(step_tasks[\"validation_failure\"])\n                    ),\n                )\n            )\n        )\n\n        # state machine\n        creation_process = aws_stepfunctions.StateMachine(  # pylint:disable=unused-variable\n            self,\n            \"dataset-version-creation\",\n            definition=dataset_version_creation_definition,\n        )\n", "path": "infra/datalake/processing_stack.py"}], "after_files": [{"content": "\"\"\"\nData Lake AWS resources definitions.\n\"\"\"\nfrom aws_cdk import aws_lambda, core\nfrom aws_cdk.core import Tags\n\n\nclass APIStack(core.Stack):\n    \"\"\"Data Lake stack definition.\"\"\"\n\n    def __init__(self, scope: core.Construct, stack_id: str, datasets_table, **kwargs) -> None:\n        super().__init__(scope, stack_id, **kwargs)\n\n        ############################################################################################\n        # ### API ENDPOINTS ########################################################################\n        ############################################################################################\n\n        endpoints = (\"datasets\",)\n\n        for endpoint in endpoints:\n            endpoint_function = aws_lambda.Function(\n                self,\n                f\"{endpoint}-endpoint-function\",\n                function_name=f\"{endpoint}-endpoint\",\n                handler=f\"endpoints.{endpoint}.entrypoint.lambda_handler\",\n                runtime=aws_lambda.Runtime.PYTHON_3_8,\n                code=aws_lambda.Code.from_asset(\n                    path=\"..\",\n                    bundling=core.BundlingOptions(\n                        # pylint:disable=no-member\n                        image=aws_lambda.Runtime.PYTHON_3_8.bundling_docker_image,\n                        command=[\"backend/bundle.bash\", f\"endpoints/{endpoint}\"],\n                    ),\n                ),\n            )\n\n            datasets_table.grant_read_write_data(endpoint_function)\n            datasets_table.grant(\n                endpoint_function, \"dynamodb:DescribeTable\"\n            )  # required by pynamodb\n\n            Tags.of(endpoint_function).add(\"ApplicationLayer\", \"api\")\n", "path": "infra/datalake/api_stack.py"}, {"content": "\"\"\"\nData Lake processing stack.\n\"\"\"\nimport textwrap\n\nfrom aws_cdk import (\n    aws_batch,\n    aws_ec2,\n    aws_ecs,\n    aws_iam,\n    aws_lambda,\n    aws_stepfunctions,\n    aws_stepfunctions_tasks,\n    core,\n)\nfrom aws_cdk.core import Tags\n\n\nclass ProcessingStack(core.Stack):\n    \"\"\"Data Lake processing stack definition.\"\"\"\n\n    # pylint: disable=too-many-locals\n    def __init__(self, scope: core.Construct, stack_id: str, deploy_env, vpc, **kwargs) -> None:\n        super().__init__(scope, stack_id, **kwargs)\n\n        ############################################################################################\n        # ### DATASET VERSION CREATE ###############################################################\n        ############################################################################################\n\n        # STATE MACHINE TASKS CONFIGURATION\n        # * type: lambda|batch\n        # * parallel: True|False\n        # * input_path: \"$\"\n        # * output_path: \"$\"\n        # * result_path: \"$\"\n        # * items_path: \"$\"\n        creation_tasks = {}\n\n        creation_tasks[\"content_iterator\"] = {\"type\": \"lambda\", \"result_path\": \"$.content\"}\n\n        creation_tasks[\"validation_summary\"] = {\"type\": \"lambda\", \"result_path\": \"$.validation\"}\n\n        creation_tasks[\"validation_failure\"] = {\n            \"type\": \"lambda\",\n            \"result_path\": aws_stepfunctions.JsonPath.DISCARD,\n        }\n\n        creation_tasks[\"check_flat_directory_structure\"] = {\n            \"type\": \"batch\",\n            \"parallel\": False,\n            \"result_path\": aws_stepfunctions.JsonPath.DISCARD,\n        }\n\n        creation_tasks[\"check_files_checksums\"] = {\n            \"type\": \"batch\",\n            \"parallel\": True,\n            \"result_path\": aws_stepfunctions.JsonPath.DISCARD,\n        }\n\n        # AWS BATCH COMPUTE ENVIRONMENT\n        batch_service_role = aws_iam.Role(\n            self,\n            \"batch-service-role\",\n            assumed_by=aws_iam.ServicePrincipal(\"batch.amazonaws.com\"),\n            managed_policies=[\n                aws_iam.ManagedPolicy.from_aws_managed_policy_name(\n                    \"service-role/AWSBatchServiceRole\"\n                ),\n            ],\n        )\n\n        batch_instance_role = aws_iam.Role(\n            self,\n            \"batch-instance-role\",\n            assumed_by=aws_iam.ServicePrincipal(\"ec2.amazonaws.com\"),\n            managed_policies=[\n                aws_iam.ManagedPolicy.from_aws_managed_policy_name(\n                    \"service-role/AmazonEC2ContainerServiceforEC2Role\"\n                ),\n            ],\n        )\n\n        batch_instance_profile = aws_iam.CfnInstanceProfile(\n            self,\n            \"batch-instance-profile\",\n            roles=[\n                batch_instance_role.role_name,\n            ],\n        )\n\n        batch_launch_template_data = textwrap.dedent(\n            \"\"\"\n            MIME-Version: 1.0\n            Content-Type: multipart/mixed; boundary=\"==MYBOUNDARY==\"\n\n            --==MYBOUNDARY==\n            Content-Type: text/x-shellscript; charset=\"us-ascii\"\n\n            #!/bin/bash\n            echo ECS_IMAGE_PULL_BEHAVIOR=prefer-cached >> /etc/ecs/ecs.config\n\n            --==MYBOUNDARY==--\n            \"\"\"\n        )\n\n        batch_launch_template = aws_ec2.CfnLaunchTemplate(\n            self,\n            \"batch-launch-template\",\n            launch_template_name=\"datalake-batch-launch-template\",\n            launch_template_data={\n                \"userData\": core.Fn.base64(batch_launch_template_data.strip()),\n            },\n        )\n\n        if deploy_env == \"prod\":\n            instance_types = [\n                aws_ec2.InstanceType(\"c5.xlarge\"),\n                aws_ec2.InstanceType(\"c5.2xlarge\"),\n                aws_ec2.InstanceType(\"c5.4xlarge\"),\n                aws_ec2.InstanceType(\"c5.9xlarge\"),\n            ]\n        else:\n            instance_types = [\n                aws_ec2.InstanceType(\"m5.large\"),\n                aws_ec2.InstanceType(\"m5.xlarge\"),\n            ]\n\n        batch_compute_environment = aws_batch.ComputeEnvironment(\n            self,\n            \"compute-environment\",\n            compute_resources=aws_batch.ComputeResources(\n                vpc=vpc,\n                # vpc_subnets=vpc.select_subnets(subnet_group_name=\"ecs-cluster\"),  # TODO\n                minv_cpus=0,\n                desiredv_cpus=0,\n                maxv_cpus=1000,\n                instance_types=instance_types,\n                instance_role=batch_instance_profile.instance_profile_name,\n                allocation_strategy=aws_batch.AllocationStrategy(\"BEST_FIT_PROGRESSIVE\"),\n                launch_template=aws_batch.LaunchTemplateSpecification(\n                    launch_template_name=batch_launch_template.launch_template_name\n                ),\n            ),\n            service_role=batch_service_role,\n        )\n\n        batch_job_queue = aws_batch.JobQueue(\n            self,\n            \"dataset-version-creation-queue\",\n            compute_environments=[\n                aws_batch.JobQueueComputeEnvironment(\n                    compute_environment=batch_compute_environment, order=10\n                ),\n            ],\n            priority=10,\n        )\n\n        # LAMBDA AND AWS BATCH BUNDLING AND STATE MACHINE TASKS CREATION\n        step_tasks = {}\n        for task_name in creation_tasks:\n\n            task = creation_tasks[task_name]\n\n            # lambda functions\n            if task[\"type\"] == \"lambda\":\n\n                lambda_function = aws_lambda.Function(\n                    self,\n                    f\"{task_name}-function\",\n                    handler=f\"processing.{task_name}.task.lambda_handler\",\n                    runtime=aws_lambda.Runtime.PYTHON_3_8,\n                    code=aws_lambda.Code.from_asset(\n                        path=\"..\",\n                        bundling=core.BundlingOptions(\n                            # pylint:disable=no-member\n                            image=aws_lambda.Runtime.PYTHON_3_8.bundling_docker_image,\n                            command=[\"backend/bundle.bash\", f\"processing/{task_name}\"],\n                        ),\n                    ),\n                )\n\n                step_tasks[task_name] = aws_stepfunctions_tasks.LambdaInvoke(\n                    self,\n                    task_name,\n                    lambda_function=lambda_function,\n                    input_path=task.get(\"input_path\", \"$\"),\n                    output_path=task.get(\"output_path\", \"$\"),\n                    result_path=task.get(\"result_path\", \"$\"),\n                    payload_response_only=True,\n                )\n\n                Tags.of(lambda_function).add(\"ApplicationLayer\", \"data-processing\")\n\n            # aws batch jobs\n            if task[\"type\"] == \"batch\":\n\n                job_definition = aws_batch.JobDefinition(\n                    self,\n                    f\"{task_name}-job\",\n                    container=aws_batch.JobDefinitionContainer(\n                        image=aws_ecs.ContainerImage.from_asset(\n                            directory=f\"../backend/processing/{task_name}\",\n                        ),\n                        memory_limit_mib=3900 if deploy_env == \"prod\" else 500,\n                        vcpus=1,\n                    ),\n                    retry_attempts=4,\n                )\n\n                job_command = [\n                    \"--dataset-id\",\n                    \"Ref::dataset_id\",\n                    \"--version-id\",\n                    \"Ref::version_id\",\n                    \"--type\",\n                    \"Ref::type\",\n                    \"--metadata-url\",\n                    \"Ref::metadata_url\",\n                    \"--dataset-id\",\n                ]\n                job_environment = {\"BATCH_JOB_FIRST_ITEM_INDEX\": \"Ref::first_item\"}\n\n                job_payload_data = {\n                    \"dataset_id.$\": \"$.dataset_id\",\n                    \"version_id.$\": \"$.version_id\",\n                    \"type.$\": \"$.type\",\n                    \"metadata_url.$\": \"$.metadata_url\",\n                }\n                job_payload_data_parallel = {\"first_item.$\": \"$.content.first_item\"}\n                job_payload_single = aws_stepfunctions.TaskInput.from_object(job_payload_data)\n                job_payload_parallel = aws_stepfunctions.TaskInput.from_object(\n                    {**job_payload_data, **job_payload_data_parallel}\n                )\n\n                if task[\"parallel\"]:\n                    step_tasks[task_name] = aws_stepfunctions_tasks.BatchSubmitJob(\n                        self,\n                        task_name,\n                        job_name=f\"{task_name}-job\",\n                        job_definition=job_definition,\n                        job_queue=batch_job_queue,\n                        array_size=aws_stepfunctions.JsonPath.number_at(\"$.content.iteration_size\"),\n                        input_path=task.get(\"input_path\", \"$\"),\n                        output_path=task.get(\"output_path\", \"$\"),\n                        result_path=task.get(\"result_path\", \"$\"),\n                        container_overrides=aws_stepfunctions_tasks.BatchContainerOverrides(\n                            command=job_command,\n                            environment=job_environment,\n                        ),\n                        payload=job_payload_parallel,\n                    )\n\n                else:\n                    step_tasks[task_name] = aws_stepfunctions_tasks.BatchSubmitJob(\n                        self,\n                        task_name,\n                        job_name=f\"{task_name}-job\",\n                        job_definition=job_definition,\n                        job_queue=batch_job_queue,\n                        input_path=task.get(\"input_path\", \"$\"),\n                        output_path=task.get(\"output_path\", \"$\"),\n                        result_path=task.get(\"result_path\", \"$\"),\n                        container_overrides=aws_stepfunctions_tasks.BatchContainerOverrides(\n                            command=job_command,\n                        ),\n                        payload=job_payload_single,\n                    )\n\n        # success task\n        step_tasks[\"success\"] = aws_stepfunctions.Succeed(\n            self,\n            \"success\",\n        )\n\n        # STATE MACHINE\n        # state machine definition\n        dataset_version_creation_definition = (\n            step_tasks[\"check_flat_directory_structure\"]\n            .next(step_tasks[\"content_iterator\"])\n            .next(step_tasks[\"check_files_checksums\"])\n            .next(\n                aws_stepfunctions.Choice(self, \"content_iteration_finished\")\n                .when(\n                    aws_stepfunctions.Condition.not_(\n                        aws_stepfunctions.Condition.number_equals(\"$.content.next_item\", -1)\n                    ),\n                    step_tasks[\"content_iterator\"],\n                )\n                .otherwise(\n                    step_tasks[\"validation_summary\"].next(\n                        aws_stepfunctions.Choice(self, \"validation_successful\")\n                        .when(\n                            aws_stepfunctions.Condition.boolean_equals(\n                                \"$.validation.success\", True\n                            ),\n                            step_tasks[\"success\"],\n                        )\n                        .otherwise(step_tasks[\"validation_failure\"])\n                    ),\n                )\n            )\n        )\n\n        # state machine\n        creation_process = aws_stepfunctions.StateMachine(  # pylint:disable=unused-variable\n            self,\n            \"dataset-version-creation\",\n            definition=dataset_version_creation_definition,\n        )\n", "path": "infra/datalake/processing_stack.py"}]}