(ELI-597) linting

Author: TOEL2

infrastructure/stacks/api-layer/scripts/create_pending_secret.py

@@ -1,75 +1,71 @@
-import boto3
+import json
+import logging
+import os
 import secrets
 import string
-import os
-import logging
-import json
 
-SECRET_NAME = os.environ.get('SECRET_NAME')
-REGION_NAME = os.environ.get('AWS_REGION')
+import boto3
+
+SECRET_NAME = os.environ.get("SECRET_NAME")
+REGION_NAME = os.environ.get("AWS_REGION")
 
 logger = logging.getLogger()
 logger.setLevel(logging.INFO)
 
+
 def generate_password(length=32):
     """Generates a secure random password."""
     alphabet = string.ascii_letters + string.digits + "!@#$%^&*"
-    return ''.join(secrets.choice(alphabet) for i in range(length))
+    return "".join(secrets.choice(alphabet) for i in range(length))
+
 
 def lambda_handler(event, context):
-    sm_client = boto3.client('secretsmanager', region_name=REGION_NAME)
+    sm_client = boto3.client("secretsmanager", region_name=REGION_NAME)
 
-    logger.info(json.dumps({
-        'event': 'rotation_started',
-        'request_id': context.aws_request_id,
-        'secret_name': SECRET_NAME,
-        'function': 'create_pending_secret'
-    }))
+    logger.info(
+        json.dumps(
+            {
+                "event": "rotation_started",
+                "request_id": context.aws_request_id,
+                "secret_name": SECRET_NAME,
+                "function": "create_pending_secret",
+            }
+        )
+    )
 
     try:
         metadata = sm_client.describe_secret(SecretId=SECRET_NAME)
         # Check if any version currently has the 'AWSPENDING' label
-        for version_id, stages in metadata.get('VersionIdsToStages', {}).items():
-            if 'AWSPENDING' in stages:
+        for version_id, stages in metadata.get("VersionIdsToStages", {}).items():
+            if "AWSPENDING" in stages:
                 msg = f"Pending version already exists with version_id: {version_id}."
 
-                logger.warning(json.dumps({
-                    'event': 'rotation_aborted',
-                    'reason': 'pending_version_exists',
-                    'pending_version_id': version_id
-                }))
+                logger.warning(
+                    json.dumps(
+                        {
+                            "event": "rotation_aborted",
+                            "reason": "pending_version_exists",
+                            "pending_version_id": version_id,
+                        }
+                    )
+                )
 
                 raise Exception(msg)
     except sm_client.exceptions.ResourceNotFoundException:
         logger.info("Secret not found. Proceeding to create (assuming it will be initialized).")
-        pass
 
     new_password = generate_password()
 
     try:
-        resp = sm_client.put_secret_value(
-            SecretId=SECRET_NAME,
-            SecretString=new_password,
-            VersionStages=['AWSPENDING']
-        )
+        resp = sm_client.put_secret_value(SecretId=SECRET_NAME, SecretString=new_password, VersionStages=["AWSPENDING"])
 
-        logger.info(json.dumps({
-            'event': 'pending_version_created',
-            'version_id': resp['VersionId'],
-            'status': 'success'
-        }))
-        return {
-            "status": "success",
-            "secret_name": SECRET_NAME,
-            "version_id": resp['VersionId']
-        }
+        logger.info(
+            json.dumps({"event": "pending_version_created", "version_id": resp["VersionId"], "status": "success"})
+        )
+        return {"status": "success", "secret_name": SECRET_NAME, "version_id": resp["VersionId"]}
 
     except sm_client.exceptions.ResourceNotFoundException:
         raise Exception(f"The secret '{SECRET_NAME}' was not found in region '{REGION_NAME}'.")
     except Exception as e:
-        logger.error(json.dumps({
-            'event': 'rotation_failed',
-            'error': str(e),
-            'type': type(e).__name__
-        }))
-        raise Exception(f"Error creating pending secret: {str(e)}")
+        logger.error(json.dumps({"event": "rotation_failed", "error": str(e), "type": type(e).__name__}))
+        raise Exception(f"Error creating pending secret: {e!s}")

infrastructure/stacks/api-layer/scripts/promote_to_current.py

@@ -1,74 +1,66 @@
-import os
+import json
 import logging
+import os
+
 import boto3
-import json
 
-SECRET_NAME = os.environ.get('SECRET_NAME')
-REGION_NAME = os.environ.get('AWS_REGION')
+SECRET_NAME = os.environ.get("SECRET_NAME")
+REGION_NAME = os.environ.get("AWS_REGION")
 
 logger = logging.getLogger()
 logger.setLevel(logging.INFO)
 
+
 def lambda_handler(event, context):
-    sm_client = boto3.client('secretsmanager', region_name=REGION_NAME)
-    logger.info(json.dumps({
-        'event': 'promotion_started',
-        'request_id': context.aws_request_id,
-        'secret_name': SECRET_NAME,
-        'function': 'promote_to_current'
-    }))
+    sm_client = boto3.client("secretsmanager", region_name=REGION_NAME)
+    logger.info(
+        json.dumps(
+            {
+                "event": "promotion_started",
+                "request_id": context.aws_request_id,
+                "secret_name": SECRET_NAME,
+                "function": "promote_to_current",
+            }
+        )
+    )
 
     try:
         metadata = sm_client.describe_secret(SecretId=SECRET_NAME)
         pending_version = None
 
-        for version_id, stages in metadata['VersionIdsToStages'].items():
-            if 'AWSPENDING' in stages:
+        for version_id, stages in metadata["VersionIdsToStages"].items():
+            if "AWSPENDING" in stages:
                 pending_version = version_id
                 break
 
         if not pending_version:
-            logger.warning(json.dumps({
-                'event': 'promotion_skipped',
-                'reason': 'no_pending_version_found',
-                'secret_name': SECRET_NAME
-            }))
+            logger.warning(
+                json.dumps(
+                    {"event": "promotion_skipped", "reason": "no_pending_version_found", "secret_name": SECRET_NAME}
+                )
+            )
             return {"status": "skipped", "reason": "no_pending_version"}
 
-        logger.info(json.dumps({
-            'event': 'promoting_version',
-            'pending_version_id': pending_version,
-            'action': 'swap_AWSCURRENT'
-        }))
+        logger.info(
+            json.dumps(
+                {"event": "promoting_version", "pending_version_id": pending_version, "action": "swap_AWSCURRENT"}
+            )
+        )
 
         sm_client.update_secret_version_stage(
-            SecretId=SECRET_NAME,
-            VersionStage='AWSCURRENT',
-            MoveToVersionId=pending_version
+            SecretId=SECRET_NAME, VersionStage="AWSCURRENT", MoveToVersionId=pending_version
         )
 
         sm_client.update_secret_version_stage(
-            SecretId=SECRET_NAME,
-            VersionStage='AWSPENDING',
-            RemoveFromVersionId=pending_version
+            SecretId=SECRET_NAME, VersionStage="AWSPENDING", RemoveFromVersionId=pending_version
         )
 
-        logger.info(json.dumps({
-            'event': 'promotion_complete',
-            'new_current_version': pending_version,
-            'status': 'success'
-        }))
+        logger.info(
+            json.dumps({"event": "promotion_complete", "new_current_version": pending_version, "status": "success"})
+        )
 
-        return {
-            'status': 'success',
-            'action': 'promoted_and_cleaned',
-            'new_current_version': pending_version
-        }
+        return {"status": "success", "action": "promoted_and_cleaned", "new_current_version": pending_version}
 
     except Exception as e:
-        logger.error(json.dumps({
-            'event': 'promotion_failed',
-            'error': str(e),
-            'type': type(e).__name__
-        }))
+        logger.error(json.dumps({"event": "promotion_failed", "error": str(e), "type": type(e).__name__}))
         raise e

src/eligibility_signposting_api/model/campaign_config.py

@@ -158,7 +158,7 @@ class IterationRule(BaseModel):
     model_config = {"populate_by_name": True, "extra": "ignore"}
 
     @field_validator("rule_stop", mode="before")
-    def parse_yn_to_bool(cls, v: str | bool) -> bool:  # noqa: N805, FBT001
+    def parse_yn_to_bool(cls, v: str | bool) -> bool:  # noqa: N805
         if isinstance(v, str):
             return v.upper() == "Y"
         return v