Skip to content

Commit df0920f

Browse files
committed
fix flake8 errors in config module
1 parent b31dffd commit df0920f

File tree

2 files changed

+45
-30
lines changed

2 files changed

+45
-30
lines changed

aws_sra_examples/solutions/genai/bedrock_org/lambda/src/app.py

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -255,7 +255,7 @@ def load_sra_cloudwatch_dashboard() -> dict:
255255
s3 = sra_s3.sra_s3()
256256
lambdas = sra_lambda.sra_lambda()
257257
sns = sra_sns.sra_sns()
258-
config = sra_config.sra_config()
258+
config = sra_config.SRAConfig()
259259
cloudwatch = sra_cloudwatch.SRACloudWatch()
260260
kms = sra_kms.sra_kms()
261261

aws_sra_examples/solutions/genai/bedrock_org/lambda/src/sra_config.py

Lines changed: 44 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -12,30 +12,25 @@
1212

1313
import logging
1414
import os
15-
from time import sleep
1615

17-
from typing import TYPE_CHECKING, Literal, Optional
18-
from typing import cast
16+
from typing import TYPE_CHECKING, Literal
1917

2018
import boto3
2119
from botocore.config import Config
2220
from botocore.exceptions import ClientError
2321

24-
import urllib.parse
2522
import json
2623

27-
import cfnresponse
2824

2925
if TYPE_CHECKING:
30-
from mypy_boto3_cloudformation import CloudFormationClient
3126
from mypy_boto3_organizations import OrganizationsClient
3227
from mypy_boto3_config import ConfigServiceClient
33-
from mypy_boto3_config.type_defs import DescribeConfigRulesResponseTypeDef, ConfigRuleTypeDef, ScopeTypeDef
34-
from mypy_boto3_iam.client import IAMClient
35-
from mypy_boto3_iam.type_defs import CreatePolicyResponseTypeDef, CreateRoleResponseTypeDef, EmptyResponseMetadataTypeDef
28+
from mypy_boto3_config.type_defs import DescribeConfigRulesResponseTypeDef
3629

3730

38-
class sra_config:
31+
class SRAConfig:
32+
"""Class to setup SRA Config resources in the organization."""
33+
3934
# Setup Default Logger
4035
LOGGER = logging.getLogger(__name__)
4136
log_level: str = os.environ.get("LOG_LEVEL", "INFO")
@@ -53,48 +48,56 @@ class sra_config:
5348
raise ValueError("Unexpected error executing Lambda function. Review CloudWatch logs for details.") from None
5449

5550
def get_organization_config_rules(self) -> dict:
56-
"""Get Organization Config Rules."""
51+
"""Get Organization Config Rules.
52+
53+
Returns:
54+
dict: Organization Config Rules
55+
"""
5756
# Get the Organization ID
5857
org_id: str = self.ORG_CLIENT.describe_organization()["Organization"]["Id"]
5958

6059
# Get the Organization Config Rules
61-
response = self.ORG_CLIENT.describe_organization_config_rules( # type: ignore
60+
response = self.ORG_CLIENT.describe_organization_config_rules( # type: ignore
6261
OrganizationConfigRuleNames=["sra_config_rule"],
6362
OrganizationId=org_id,
6463
)
6564

6665
# Log the response
67-
sra_config.LOGGER.info(response)
66+
self.LOGGER.info(response)
6867

6968
# Return the response
7069
return response
7170

7271
def put_organization_config_rule(self) -> dict:
73-
"""Put Organization Config Rule."""
72+
"""Put Organization Config Rule.
73+
74+
Returns:
75+
dict: Organization Config Rule
76+
"""
7477
# Get the Organization ID
7578
org_id: str = self.ORG_CLIENT.describe_organization()["Organization"]["Id"]
7679

7780
# Put the Organization Config Rule
78-
response = self.ORG_CLIENT.put_organization_config_rule( # type: ignore
81+
response = self.ORG_CLIENT.put_organization_config_rule( # type: ignore
7982
OrganizationConfigRuleName="sra_config_rule",
8083
OrganizationId=org_id,
8184
ConfigRuleName="sra_config_rule",
8285
)
8386

8487
# Log the response
85-
sra_config.LOGGER.info(response)
88+
self.LOGGER.info(response)
8689

8790
# Return the response
8891
return response
8992

9093
def find_config_rule(self, rule_name: str) -> tuple[bool, dict | DescribeConfigRulesResponseTypeDef]:
91-
"""Get config rule
94+
"""Get config rule.
9295
9396
Args:
9497
rule_name (str): Config rule name
9598
9699
Raises:
97-
ValueError: If the config rule is not found
100+
ValueError: Unexpected error executing Lambda function. Review CloudWatch logs for details.
98101
99102
Returns:
100103
tuple[bool, dict | DescribeConfigRulesResponseTypeDef]: True if the config rule is found, False if not, and the response
@@ -110,19 +113,27 @@ def find_config_rule(self, rule_name: str) -> tuple[bool, dict | DescribeConfigR
110113
if e.response["Error"]["Code"] == "NoSuchConfigRuleException":
111114
self.LOGGER.info(f"No such config rule: {rule_name}")
112115
return False, {}
113-
else:
114-
self.LOGGER.info(f"Unexpected error: {e}")
115-
raise e
116-
# Log the response
116+
self.LOGGER.info(f"Unexpected error: {e}")
117+
raise ValueError(f"Unexpected error executing Lambda function. Review CloudWatch logs for details. {e}") from None
117118
self.LOGGER.info(f"Config rule {rule_name} exists: {response}")
118119
return True, response
119120

120-
121-
def create_config_rule(self, rule_name: str, lambda_arn: str,
122-
max_frequency: Literal["One_Hour", "Three_Hours", "Six_Hours", "Twelve_Hours", "TwentyFour_Hours"],
123-
owner: Literal["CUSTOM_LAMBDA", "AWS"], description: str, input_params: dict,
121+
def create_config_rule(self, rule_name: str, lambda_arn: str, # noqa: CFQ002
122+
max_frequency: Literal["One_Hour", "Three_Hours", "Six_Hours", "Twelve_Hours", "TwentyFour_Hours"],
123+
owner: Literal["CUSTOM_LAMBDA", "AWS"], description: str, input_params: dict,
124124
eval_mode: Literal["DETECTIVE", "PROACTIVE"], solution_name: str) -> None:
125-
"""Create Config Rule."""
125+
"""Create Config Rule.
126+
127+
Args:
128+
rule_name (str): Config rule name
129+
lambda_arn (str): Lambda ARN
130+
max_frequency (Literal["One_Hour", "Three_Hours", "Six_Hours", "Twelve_Hours", "TwentyFour_Hours"]): Config rule max frequency
131+
owner (Literal["CUSTOM_LAMBDA", "AWS"]): Config rule owner
132+
description (str): Config rule description
133+
input_params (dict): Config rule input parameters
134+
eval_mode (Literal["DETECTIVE", "PROACTIVE"]): Config rule evaluation mode
135+
solution_name (str): SRA solution name
136+
"""
126137
self.CONFIG_CLIENT.put_config_rule(
127138
ConfigRule={
128139
"ConfigRuleName": rule_name,
@@ -153,7 +164,11 @@ def create_config_rule(self, rule_name: str, lambda_arn: str,
153164
self.LOGGER.info(f"{rule_name} config rule created...")
154165

155166
def delete_config_rule(self, rule_name: str) -> None:
156-
"""Delete Config Rule."""
167+
"""Delete Config Rule.
168+
169+
Args:
170+
rule_name (str): Config rule name
171+
"""
157172
# Delete the Config Rule
158173
try:
159174
self.CONFIG_CLIENT.delete_config_rule(
@@ -166,4 +181,4 @@ def delete_config_rule(self, rule_name: str) -> None:
166181
if e.response["Error"]["Code"] == "NoSuchConfigRuleException":
167182
self.LOGGER.info(f"No such config rule: {rule_name}")
168183
else:
169-
self.LOGGER.info(f"Unexpected error: {e}")
184+
self.LOGGER.info(f"Unexpected error: {e}")

0 commit comments

Comments
 (0)