generated from cds-snc/project-template
-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Feat/isolate incident roles, docs and folders concept into modules (#617
) * feat: split folder related functions into own module * feat: migrate folder related functions into own module * fix: remove commented code * chore: fmt * fix: add unit tests for update spreadsheet incident status * fix: remove alias * feat: migrate roles functions to own module * fix: update docstring for additional context * feat: migrate docs related functions into own module * chore: fmt
- Loading branch information
Showing
8 changed files
with
941 additions
and
659 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,43 @@ | ||
"""Module to manage the incident document used to track the details.""" | ||
|
||
from integrations.google_workspace import google_docs | ||
|
||
|
||
def update_incident_document_status(document_id, new_status="Closed"): | ||
"""Update the status of the incident document. | ||
Args: | ||
document_id (str): The ID of the document to update. | ||
new_status (str, optional): The new status to set. Defaults to "Closed". | ||
Returns: | ||
bool: True if the status was updated, False otherwise. | ||
""" | ||
# List of possible statuses to be replaced | ||
possible_statuses = [ | ||
"In Progress", | ||
"Open", | ||
"Ready to be Reviewed", | ||
"Reviewed", | ||
"Closed", | ||
] | ||
|
||
if new_status not in possible_statuses: | ||
raise ValueError(f"Invalid status: {new_status}") | ||
|
||
# Replace all possible statuses with the new status | ||
changes = [ | ||
{ | ||
"replaceAllText": { | ||
"containsText": {"text": f"Status: {status}", "matchCase": "false"}, | ||
"replaceText": f"Status: {new_status}", | ||
} | ||
} | ||
for status in possible_statuses | ||
if status != new_status | ||
] | ||
replies = google_docs.batch_update(document_id, changes)["replies"] | ||
return any( | ||
reply.get("replaceAllText", {}).get("occurrencesChanged", 0) > 0 | ||
for reply in replies | ||
) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,246 @@ | ||
"""Module for managing SRE incident folders in Google Drive. | ||
Includes functions to manage the folders, the metadata, and the list of incidents in a Google Sheets spreadsheet. | ||
""" | ||
|
||
import os | ||
from slack_sdk.web import WebClient | ||
from slack_bolt import Ack | ||
from integrations.google_workspace import google_drive, sheets | ||
import logging | ||
|
||
SRE_INCIDENT_FOLDER = os.environ.get("SRE_INCIDENT_FOLDER") | ||
INCIDENT_LIST = os.environ.get("INCIDENT_LIST") | ||
|
||
|
||
def list_folders(client: WebClient, body, ack: Ack): | ||
ack() | ||
folders = google_drive.list_folders_in_folder( | ||
SRE_INCIDENT_FOLDER, "not name contains 'Templates'" | ||
) | ||
folders.sort(key=lambda x: x["name"]) | ||
blocks = { | ||
"type": "modal", | ||
"callback_id": "list_folders_view", | ||
"title": {"type": "plain_text", "text": "SRE - Listing folders"}, | ||
"close": {"type": "plain_text", "text": "Close"}, | ||
"blocks": [ | ||
item for sublist in list(map(folder_item, folders)) for item in sublist | ||
], | ||
} | ||
client.views_open(trigger_id=body["trigger_id"], view=blocks) | ||
|
||
|
||
def delete_folder_metadata(client: WebClient, body, ack): | ||
ack() | ||
folder_id = body["view"]["private_metadata"] | ||
key = body["actions"][0]["value"] | ||
response = google_drive.delete_metadata(folder_id, key) | ||
if not response: | ||
logging.info(f"Failed to delete metadata `{key}` for folder `{folder_id}`") | ||
else: | ||
logging.info(f"Deleted metadata for key `{key}`") | ||
body["actions"] = [{"value": folder_id}] | ||
view_folder_metadata(client, body, ack) | ||
|
||
|
||
def save_metadata(client: WebClient, body, ack, view): | ||
ack() | ||
folder_id = view["private_metadata"] | ||
key = view["state"]["values"]["key"]["key"]["value"] | ||
value = view["state"]["values"]["value"]["value"]["value"] | ||
google_drive.add_metadata(folder_id, key, value) | ||
body["actions"] = [{"value": folder_id}] | ||
del body["view"] | ||
view_folder_metadata(client, body, ack) | ||
|
||
|
||
def view_folder_metadata(client, body, ack): | ||
ack() | ||
folder_id = body["actions"][0]["value"] | ||
logging.info(f"Viewing metadata for folder {folder_id}") | ||
folder = google_drive.list_metadata(folder_id) | ||
blocks = { | ||
"type": "modal", | ||
"callback_id": "view_folder_metadata_modal", | ||
"title": {"type": "plain_text", "text": "SRE - Showing metadata"}, | ||
"submit": {"type": "plain_text", "text": "Return to folders"}, | ||
"private_metadata": folder_id, | ||
"blocks": ( | ||
[ | ||
{ | ||
"type": "section", | ||
"text": { | ||
"type": "plain_text", | ||
"text": folder["name"], | ||
}, | ||
"accessory": { | ||
"type": "button", | ||
"text": {"type": "plain_text", "text": "Add metadata"}, | ||
"value": folder_id, | ||
"action_id": "add_folder_metadata", | ||
}, | ||
}, | ||
{"type": "divider"}, | ||
] | ||
+ metadata_items(folder) | ||
), | ||
} | ||
if "view" in body: | ||
client.views_update( | ||
view_id=body["view"]["id"], | ||
view=blocks, | ||
) | ||
else: | ||
client.views_open(trigger_id=body["trigger_id"], view=blocks) | ||
|
||
|
||
def add_folder_metadata(client: WebClient, body, ack): | ||
ack() | ||
folder_id = body["actions"][0]["value"] | ||
blocks = { | ||
"type": "modal", | ||
"callback_id": "add_metadata_view", | ||
"title": {"type": "plain_text", "text": "SRE - Add metadata"}, | ||
"submit": {"type": "plain_text", "text": "Save metadata"}, | ||
"close": {"type": "plain_text", "text": "Cancel"}, | ||
"private_metadata": folder_id, | ||
"blocks": [ | ||
{ | ||
"type": "section", | ||
"text": { | ||
"type": "mrkdwn", | ||
"text": "*Add metadata*", | ||
}, | ||
}, | ||
{ | ||
"type": "input", | ||
"block_id": "key", | ||
"element": { | ||
"type": "plain_text_input", | ||
"action_id": "key", | ||
"placeholder": {"type": "plain_text", "text": "Key"}, | ||
}, | ||
"label": { | ||
"type": "plain_text", | ||
"text": "Key", | ||
}, | ||
}, | ||
{ | ||
"type": "input", | ||
"block_id": "value", | ||
"element": { | ||
"type": "plain_text_input", | ||
"action_id": "value", | ||
"placeholder": {"type": "plain_text", "text": "Value"}, | ||
}, | ||
"label": { | ||
"type": "plain_text", | ||
"text": "Value", | ||
}, | ||
}, | ||
], | ||
} | ||
client.views_update( | ||
view_id=body["view"]["id"], | ||
view=blocks, | ||
) | ||
|
||
|
||
def folder_item(folder): | ||
return [ | ||
{ | ||
"type": "section", | ||
"text": {"type": "mrkdwn", "text": f"*{folder['name']}*"}, | ||
"accessory": { | ||
"type": "button", | ||
"text": { | ||
"type": "plain_text", | ||
"text": "Manage metadata", | ||
"emoji": True, | ||
}, | ||
"value": f"{folder['id']}", | ||
"action_id": "view_folder_metadata", | ||
}, | ||
}, | ||
{ | ||
"type": "context", | ||
"elements": [ | ||
{ | ||
"type": "mrkdwn", | ||
"text": f"<https://drive.google.com/drive/u/0/folders/{folder['id']}|View in Google Drive>", | ||
} | ||
], | ||
}, | ||
{"type": "divider"}, | ||
] | ||
|
||
|
||
def metadata_items(folder): | ||
if "appProperties" not in folder or len(folder["appProperties"]) == 0: | ||
return [ | ||
{ | ||
"type": "section", | ||
"text": { | ||
"type": "mrkdwn", | ||
"text": "*No metadata found. Click the button above to add metadata.*", | ||
}, | ||
}, | ||
] | ||
else: | ||
return [ | ||
{ | ||
"type": "section", | ||
"text": { | ||
"type": "mrkdwn", | ||
"text": f"*{key}*\n{value}", | ||
}, | ||
"accessory": { | ||
"type": "button", | ||
"text": { | ||
"type": "plain_text", | ||
"text": "Delete metadata", | ||
"emoji": True, | ||
}, | ||
"value": key, | ||
"style": "danger", | ||
"action_id": "delete_folder_metadata", | ||
}, | ||
} | ||
for key, value in folder["appProperties"].items() | ||
] | ||
|
||
|
||
def update_spreadsheet_incident_status(channel_name, status="Closed"): | ||
"""Update the status of an incident in the incident list spreadsheet. | ||
Args: | ||
channel_name (str): The name of the channel to update. | ||
status (str): The status to update the incident to. | ||
Returns: | ||
bool: True if the status was updated successfully, False otherwise. | ||
""" | ||
valid_statuses = ["Open", "Closed", "In Progress", "Resolved"] | ||
if status not in valid_statuses: | ||
logging.warning("Invalid status %s", status) | ||
return False | ||
sheet_name = "Sheet1" | ||
sheet = sheets.get_values(INCIDENT_LIST, range=sheet_name) | ||
values = sheet.get("values", []) | ||
if len(values) == 0: | ||
logging.warning("No incident found for channel %s", channel_name) | ||
return False | ||
# Find the row with the search value | ||
for i, row in enumerate(values): | ||
if channel_name in row: | ||
# Update the 4th column (index 3) of the found row | ||
update_range = ( | ||
f"{sheet_name}!D{i+1}" # Column D, Rows are 1-indexed in Sheets | ||
) | ||
updated_sheet = sheets.batch_update_values( | ||
INCIDENT_LIST, update_range, [[status]] | ||
) | ||
if updated_sheet: | ||
return True | ||
return False |
Oops, something went wrong.