chore: fixes linting issues

Author: bencegadanyi1-nhs

packages/slackBotFunction/app/core/config.py

@@ -3,12 +3,13 @@
 Sets up all the AWS and Slack connections we need.
 """
 
-from dataclasses import dataclass
-from functools import lru_cache
-import os
 import json
+import os
 import traceback
-from typing import Tuple
+from dataclasses import dataclass
+from functools import lru_cache
+from typing import tuple
+
 import boto3
 from aws_lambda_powertools import Logger
 from aws_lambda_powertools.logging import utils
@@ -18,7 +19,7 @@
 # we use lru_cache for lots of configs so they are cached
 
 
-@lru_cache()
+@lru_cache
 def get_logger() -> Logger:
     powertools_logger = Logger(service="slackBotFunction")
     utils.copy_config_to_registered_loggers(source_logger=powertools_logger, ignore_log_level=True)
@@ -29,15 +30,15 @@ def get_logger() -> Logger:
 logger = get_logger()
 
 
-@lru_cache()
+@lru_cache
 def get_slack_bot_state_table() -> Table:
     # DynamoDB table for deduplication and session storage
     dynamodb = boto3.resource("dynamodb")
     return dynamodb.Table(os.environ["SLACK_BOT_STATE_TABLE"])
 
 
-@lru_cache()
-def get_ssm_params() -> Tuple[str, str]:
+@lru_cache
+def get_ssm_params() -> tuple[str, str]:
     bot_token_parameter = os.environ["SLACK_BOT_TOKEN_PARAMETER"]
     signing_secret_parameter = os.environ["SLACK_SIGNING_SECRET_PARAMETER"]
     try:
@@ -57,7 +58,7 @@ def get_ssm_params() -> Tuple[str, str]:
             raise ValueError("Missing required parameters: token or secret in Parameter Store values")
 
     except json.JSONDecodeError as e:
-        raise ValueError(f"Invalid JSON in Parameter Store: {e}")
+        raise ValueError(f"Invalid JSON in Parameter Store: {e}") from e
     except Exception:
         logger.error("Configuration error", extra={"error": traceback.format_exc()})
         raise
@@ -71,7 +72,7 @@ def get_bot_token() -> str:
 
 
 @lru_cache
-def get_guardrail_config() -> Tuple[str, str, str, str, str]:
+def get_guardrail_config() -> tuple[str, str, str, str, str]:
     # Bedrock configuration from environment
     KNOWLEDGEBASE_ID = os.environ["KNOWLEDGEBASE_ID"]
     RAG_MODEL_ID = os.environ["RAG_MODEL_ID"]

packages/slackBotFunction/app/handler.py

@@ -6,8 +6,8 @@
 2. Processes async operations when invoked by itself to avoid timeouts
 """
 
-from slack_bolt.adapter.aws_lambda import SlackRequestHandler
 from aws_lambda_powertools.utilities.typing import LambdaContext
+from slack_bolt.adapter.aws_lambda import SlackRequestHandler
 
 from app.core.config import get_logger
 from app.services.app import get_app

packages/slackBotFunction/app/services/app.py

@@ -1,10 +1,11 @@
-from functools import lru_cache
 import logging
+from functools import lru_cache
 
+from aws_lambda_powertools import Logger
 from slack_bolt import App
+
 from app.core.config import get_ssm_params
 from app.slack.slack_handlers import setup_handlers
-from aws_lambda_powertools import Logger
 
 
 @lru_cache

packages/slackBotFunction/app/services/bedrock.py

@@ -1,13 +1,13 @@
 import json
 from typing import Any
+
 import boto3
 from mypy_boto3_bedrock_agent_runtime import AgentsforBedrockRuntimeClient
-from mypy_boto3_bedrock_runtime.client import BedrockRuntimeClient
 from mypy_boto3_bedrock_agent_runtime.type_defs import RetrieveAndGenerateResponseTypeDef
+from mypy_boto3_bedrock_runtime.client import BedrockRuntimeClient
 
 from app.core.config import get_guardrail_config, get_logger
 
-
 logger = get_logger()
 
 

packages/slackBotFunction/app/services/dynamo.py

@@ -1,8 +1,10 @@
-from typing import Any
-from app.core.config import get_logger, get_slack_bot_state_table
 from time import time
+from typing import Any
+
 from mypy_boto3_dynamodb.type_defs import GetItemOutputTableTypeDef
 
+from app.core.config import get_logger, get_slack_bot_state_table
+
 logger = get_logger()
 
 

packages/slackBotFunction/app/services/prompt_loader.py

@@ -1,11 +1,13 @@
 import os
 import traceback
+
 import boto3
 from botocore.exceptions import ClientError
-from app.core.config import get_logger
-from app.services.exceptions import PromptNotFoundError, PromptLoadError
 from mypy_boto3_bedrock_agent import AgentsforBedrockClient
 
+from app.core.config import get_logger
+from app.services.exceptions import PromptLoadError, PromptNotFoundError
+
 logger = get_logger()
 
 
@@ -66,14 +68,14 @@ def load_prompt(prompt_name: str, prompt_version: str = None) -> str:
         )
         raise PromptLoadError(
             f"Failed to load prompt '{prompt_name}' version '{prompt_version}': {error_code} - {error_message}"
-        )
+        ) from e
 
     except Exception as e:
         logger.error(
             "Unexpected error loading prompt",
             extra={"prompt_name": prompt_name, "error_type": type(e).__name__, "error": traceback.format_exc()},
         )
-        raise PromptLoadError(f"Unexpected error loading prompt '{prompt_name}': {e}")
+        raise PromptLoadError(f"Unexpected error loading prompt '{prompt_name}': {e}") from e
 
 
 def get_prompt_id_from_name(client: AgentsforBedrockClient, prompt_name: str) -> str | None:

packages/slackBotFunction/app/services/query_reformulator.py

@@ -1,12 +1,14 @@
 import os
 import traceback
+
 import boto3
+from mypy_boto3_bedrock_runtime.client import BedrockRuntimeClient
 
 from app.core.config import get_logger
 from app.services.bedrock import invoke_model
-from .prompt_loader import load_prompt
+
 from .exceptions import ConfigurationError
-from mypy_boto3_bedrock_runtime.client import BedrockRuntimeClient
+from .prompt_loader import load_prompt
 
 logger = get_logger()
 

packages/slackBotFunction/app/services/slack.py

@@ -1,8 +1,8 @@
 import traceback
 
 from slack_sdk import WebClient
-from app.core.config import bot_messages, get_logger
 
+from app.core.config import bot_messages, get_logger
 
 logger = get_logger()
 

packages/slackBotFunction/app/slack/slack_events.py

@@ -3,13 +3,15 @@
 Handles conversation memory, Bedrock queries, and responding back to Slack
 """
 
+import json
 import re
 import time
 import traceback
-import json
-from typing import Any, Dict
+from typing import Any, dict
+
 from botocore.exceptions import ClientError
 from slack_sdk import WebClient
+
 from app.core.config import (
     bot_messages,
     constants,
@@ -43,7 +45,7 @@
 
 
 def cleanup_previous_unfeedback_qa(
-    conversation_key: str, current_message_ts: str, session_data: Dict[str, Any]
+    conversation_key: str, current_message_ts: str, session_data: dict[str, Any]
 ) -> None:
     """Delete previous Q&A pair if no feedback received using atomic operation"""
     try:
@@ -112,9 +114,9 @@ def _mark_qa_feedback_received(conversation_key: str, message_ts: str) -> None:
 
 def _handle_session_management(
     conversation_key: str,
-    session_data: Dict[str, Any],
+    session_data: dict[str, Any],
     session_id: str,
-    kb_response: Dict[str, Any],
+    kb_response: dict[str, Any],
     user_id: str,
     channel: str,
     thread_ts: str,
@@ -184,7 +186,7 @@ def process_feedback_event(
     channel_id: str,
     thread_root: str,
     client: WebClient,
-    event: Dict[str, Any],
+    event: dict[str, Any],
 ) -> None:
     feedback_text = message_text.split(":", 1)[1].strip() if ":" in message_text else ""
     try:
@@ -208,7 +210,7 @@ def process_feedback_event(
         post_error_message(channel=channel_id, thread_ts=thread_ts, client=client)
 
 
-def process_async_slack_action(body: Dict[str, Any], client: WebClient) -> None:
+def process_async_slack_action(body: dict[str, Any], client: WebClient) -> None:
     try:
         channel_id = body["channel"]["id"]
         action_id = body["actions"][0]["action_id"]
@@ -261,7 +263,7 @@ def process_async_slack_action(body: Dict[str, Any], client: WebClient) -> None:
         logger.error(f"Error handling feedback: {e}", extra={"error": traceback.format_exc()})
 
 
-def process_async_slack_event(event: Dict[str, Any], event_id: str, client: WebClient) -> None:
+def process_async_slack_event(event: dict[str, Any], event_id: str, client: WebClient) -> None:
     logger.debug("Processing async Slack event", extra={"event_id": event_id, "event": event})
     original_message_text = (event.get("text") or "").strip()
     message_text = strip_mentions(message_text=original_message_text)
@@ -293,7 +295,7 @@ def process_async_slack_event(event: Dict[str, Any], event_id: str, client: WebC
     process_slack_message(event=event, event_id=event_id, client=client)
 
 
-def process_slack_message(event: Dict[str, Any], event_id: str, client: WebClient) -> None:
+def process_slack_message(event: dict[str, Any], event_id: str, client: WebClient) -> None:
     """
     Process Slack events asynchronously after initial acknowledgment
 
@@ -369,7 +371,7 @@ def process_slack_message(event: Dict[str, Any], event_id: str, client: WebClien
         post_error_message(channel=channel, thread_ts=thread_ts, client=client)
 
 
-def process_pull_request_slack_event(slack_event_data: Dict[str, Any]) -> None:
+def process_pull_request_slack_event(slack_event_data: dict[str, Any]) -> None:
     # separate function to process pull requests so that we can ensure we store session information
     try:
         event_id = slack_event_data["event_id"]
@@ -384,7 +386,7 @@ def process_pull_request_slack_event(slack_event_data: Dict[str, Any]) -> None:
         logger.error("Error processing message", extra={"event_id": event_id, "error": traceback.format_exc()})
 
 
-def process_pull_request_slack_action(slack_body_data: Dict[str, Any]) -> None:
+def process_pull_request_slack_action(slack_body_data: dict[str, Any]) -> None:
     # separate function to process pull requests so that we can ensure we store session information
     try:
         token = get_bot_token()
@@ -395,7 +397,7 @@ def process_pull_request_slack_action(slack_body_data: Dict[str, Any]) -> None:
         logger.error("Error processing message", extra={"error": traceback.format_exc()})
 
 
-def log_query_stats(user_query: str, event: Dict[str, Any], channel: str, client: WebClient, thread_ts: str) -> None:
+def log_query_stats(user_query: str, event: dict[str, Any], channel: str, client: WebClient, thread_ts: str) -> None:
     query_length = len(user_query)
     start_time = float(event["event_ts"])
     end_time = time.time()
@@ -520,7 +522,7 @@ def get_conversation_session(conversation_key: str) -> str | None:
     return session_data.get("session_id") if session_data else None
 
 
-def get_conversation_session_data(conversation_key: str) -> Dict[str, Any]:
+def get_conversation_session_data(conversation_key: str) -> dict[str, Any]:
     """
     Get full session data for this conversation
     """

packages/slackBotFunction/app/slack/slack_handlers.py

@@ -8,14 +8,17 @@
 """
 
 import json
-from functools import lru_cache
 import traceback
-from typing import Any, Dict
+from functools import lru_cache
+from typing import Any, dict
+
 from slack_bolt import Ack, App
 from slack_sdk import WebClient
+
 from app.core.config import (
     get_logger,
 )
+from app.slack.slack_events import process_async_slack_action, process_async_slack_event
 from app.utils.handler_utils import (
     conversation_key_and_root,
     extract_session_pull_request_id,
@@ -25,7 +28,6 @@
     respond_with_eyes,
     should_reply_to_message,
 )
-from app.slack.slack_events import process_async_slack_action, process_async_slack_event
 
 logger = get_logger()
 
@@ -49,7 +51,7 @@ def setup_handlers(app: App) -> None:
 
 
 # ack function for events where we respond with eyes
-def respond_to_events(event: Dict[str, Any], ack: Ack, client: WebClient):
+def respond_to_events(event: dict[str, Any], ack: Ack, client: WebClient):
     if should_reply_to_message(event):
         respond_with_eyes(event=event, client=client)
     logger.debug("Sending ack response")
@@ -62,7 +64,7 @@ def respond_to_action(ack: Ack):
     ack()
 
 
-def feedback_handler(body: Dict[str, Any], client: WebClient) -> None:
+def feedback_handler(body: dict[str, Any], client: WebClient) -> None:
     """Handle feedback button clicks (both positive and negative)."""
     try:
         feedback_data = json.loads(body["actions"][0]["value"])
@@ -87,7 +89,7 @@ def feedback_handler(body: Dict[str, Any], client: WebClient) -> None:
 # ================================================================
 
 
-def unified_message_handler(client: WebClient, event: Dict[str, Any], body: Dict[str, Any]) -> None:
+def unified_message_handler(client: WebClient, event: dict[str, Any], body: dict[str, Any]) -> None:
     """
     All messages get processed by this code
     If message starts with FEEDBACK_PREFIX then handle feedback message and return