From 7c80f857d3eb054410c6f42544559ff3e5eab809 Mon Sep 17 00:00:00 2001 From: Nicholas Mei Date: Mon, 6 May 2024 11:00:15 -0700 Subject: [PATCH] Fix pre-existing linting errors --- .../constructs_/efs/file_system.py | 12 +++--- .../constructs_/monitoring.py | 16 +++++--- .../constructs_/ssm/tools.py | 4 +- .../stacks/data_sync.py | 39 ++++++++++--------- 4 files changed, 37 insertions(+), 34 deletions(-) diff --git a/src/aibs_informatics_cdk_lib/constructs_/efs/file_system.py b/src/aibs_informatics_cdk_lib/constructs_/efs/file_system.py index cc1c4f5..4aa6081 100644 --- a/src/aibs_informatics_cdk_lib/constructs_/efs/file_system.py +++ b/src/aibs_informatics_cdk_lib/constructs_/efs/file_system.py @@ -61,9 +61,9 @@ def __init__( scope, id, vpc=vpc, - file_system_name=self.get_name_with_env(file_system_name) - if file_system_name - else None, + file_system_name=( + self.get_name_with_env(file_system_name) if file_system_name else None + ), allow_anonymous_access=allow_anonymous_access, enable_automatic_backups=enable_automatic_backups, encrypted=encrypted, @@ -158,9 +158,9 @@ def __init__( self, "fs", env_base=self.env_base, - file_system_name=self.get_name_with_env(file_system_name) - if file_system_name - else None, + file_system_name=( + self.get_name_with_env(file_system_name) if file_system_name else None + ), lifecycle_policy=efs.LifecyclePolicy.AFTER_7_DAYS, out_of_infrequent_access_policy=efs.OutOfInfrequentAccessPolicy.AFTER_1_ACCESS, enable_automatic_backups=False, diff --git a/src/aibs_informatics_cdk_lib/constructs_/monitoring.py b/src/aibs_informatics_cdk_lib/constructs_/monitoring.py index f3fb535..f953a87 100644 --- a/src/aibs_informatics_cdk_lib/constructs_/monitoring.py +++ b/src/aibs_informatics_cdk_lib/constructs_/monitoring.py @@ -78,9 +78,11 @@ def add_function_widgets( self.dashboard_tools.add_text_widget(f"{group_name} Lambda Functions", 1) function_names_groups: List[Tuple[str, List[str]]] = [ - cast(Tuple[str, List[str]], _) - if isinstance(_, tuple) and len(_) == 2 and isinstance(_[-1], list) - else ("/".join(_), _) + ( + cast(Tuple[str, List[str]], _) + if isinstance(_, tuple) and len(_) == 2 and isinstance(_[-1], list) + else ("/".join(_), _) + ) for _ in function_names_groups ] @@ -184,9 +186,11 @@ def add_state_machine_widgets( *grouped_state_machine_names: Union[Tuple[str, List[str]], List[str]], ): grouped_state_machine_names: List[Tuple[str, List[str]]] = [ - cast(Tuple[str, List[str]], _) - if isinstance(_, tuple) and len(_) == 2 and isinstance(_[-1], list) - else ("/".join(_), _) + ( + cast(Tuple[str, List[str]], _) + if isinstance(_, tuple) and len(_) == 2 and isinstance(_[-1], list) + else ("/".join(_), _) + ) for _ in grouped_state_machine_names ] self.dashboard_tools.add_text_widget(f"{group_name} State Machines", 1) diff --git a/src/aibs_informatics_cdk_lib/constructs_/ssm/tools.py b/src/aibs_informatics_cdk_lib/constructs_/ssm/tools.py index a10ece1..e08ba18 100644 --- a/src/aibs_informatics_cdk_lib/constructs_/ssm/tools.py +++ b/src/aibs_informatics_cdk_lib/constructs_/ssm/tools.py @@ -85,9 +85,7 @@ def upload_file( ) # Store the S3 URI location in an SSM parameter - s3_uri = S3URI.build( - destination_bucket.bucket_name, destination_key, full_validate=False - ) + s3_uri = S3URI.build(destination_bucket.bucket_name, destination_key, full_validate=False) parameter_name = self.env_base.get_ssm_param_name(*param_name_components) cmd_wrapper_param = ssm.StringParameter( diff --git a/src/aibs_informatics_cdk_lib/stacks/data_sync.py b/src/aibs_informatics_cdk_lib/stacks/data_sync.py index 2212c81..c3d7e8c 100644 --- a/src/aibs_informatics_cdk_lib/stacks/data_sync.py +++ b/src/aibs_informatics_cdk_lib/stacks/data_sync.py @@ -155,7 +155,6 @@ def code_asset(self) -> CodeAsset: return self._code_asset def create_lambda_functions(self) -> None: - # ---------------------------------------------------------- # Data Transfer functions # ---------------------------------------------------------- @@ -251,7 +250,6 @@ def create_lambda_functions(self) -> None: self.add_managed_policies(self.prepare_batch_data_sync_fn.role, "AmazonS3ReadOnlyAccess") def create_step_functions(self): - start_pass_state = sfn.Pass( self, "Data Sync: Start", @@ -300,14 +298,16 @@ def create_step_functions(self): }, ) ], - platform_capabilities=["FARGATE"] - if any( - [ - isinstance(ce.compute_environment, batch.FargateComputeEnvironment) - for ce in self._job_queue.compute_environments - ] - ) - else None, + platform_capabilities=( + ["FARGATE"] + if any( + [ + isinstance(ce.compute_environment, batch.FargateComputeEnvironment) + for ce in self._job_queue.compute_environments + ] + ) + else None + ), ) ) # fmt: off @@ -360,7 +360,6 @@ def create_step_functions(self): ) def create_rclone_step_function(self): - start_pass_state = sfn.Pass( self, "Rclone: Start", @@ -413,14 +412,16 @@ def create_rclone_step_function(self): }, ) ], - platform_capabilities=["FARGATE"] - if any( - [ - isinstance(ce.compute_environment, batch.FargateComputeEnvironment) - for ce in self._job_queue.compute_environments - ] - ) - else None, + platform_capabilities=( + ["FARGATE"] + if any( + [ + isinstance(ce.compute_environment, batch.FargateComputeEnvironment) + for ce in self._job_queue.compute_environments + ] + ) + else None + ), ) ) # fmt: off