diff --git a/extensions/app-canary/app-canary.qmd b/extensions/app-canary/app-canary.qmd
index a94eddca..e26b4293 100644
--- a/extensions/app-canary/app-canary.qmd
+++ b/extensions/app-canary/app-canary.qmd
@@ -13,40 +13,39 @@ import json
import os
import requests
import datetime
-import pandas as pd
from posit import connect
-from great_tables import GT, style, loc, html
from IPython.display import HTML, display
# Used to display on-screen setup instructions if environment variables are missing
show_instructions = False
instructions = []
-gt_tbl = None
-
-# Read CONNECT_SERVER from environment, this is automatically configured on Connect, set manually for local dev
-connect_server = os.environ.get("CONNECT_SERVER", "")
-if not connect_server:
- show_instructions = True
- instructions.append("Please set the CONNECT_SERVER environment variable.")
-
-# Read CONNECT_API_KEY from environment, this is automatically configured on Connect, set manually for local dev
-api_key = os.environ.get("CONNECT_API_KEY", "")
-if not api_key:
- show_instructions = True
- instructions.append("Please set the CONNECT_API_KEY environment variable.")
-
-# Read CANARY_GUIDS from environment, needs to be manually configured on Connect and for local dev
-canary_guids_str = os.environ.get("CANARY_GUIDS", "")
-if not canary_guids_str:
- show_instructions = True
- instructions.append("Please set the CANARY_GUIDS environment variable. It should be a comma separated list of GUID you wish to monitor.")
- canary_guids = []
-else:
- # Clean up the GUIDs
- canary_guids = [guid.strip() for guid in canary_guids_str.split(',') if guid.strip()]
- if not canary_guids:
+
+# Used to display on-screen error messages if API errors occur
+show_error = False
+error_message = ""
+error_guid = ""
+
+# Variable to store app monitoring result
+app_result = None
+
+# Helper function to read environment variables and add instructions if missing
+def get_env_var(var_name, description=""):
+ """Get environment variable and add instruction if missing"""
+ global show_instructions
+
+ value = os.environ.get(var_name, "")
+ if not value:
show_instructions = True
- instructions.append(f"CANARY_GUIDS environment variable is set but is empty or contains only whitespace. It should be a comma separated list of GUID you wish to monitor. Raw CANARY_GUIDS value: '{canary_guids_str}'")
+ instruction = f"Please set the {var_name} environment variable and re-render the report."
+ if description:
+ instruction += f" {description}"
+ instructions.append(instruction)
+ return value
+
+# Read environment variables
+connect_server = get_env_var("CONNECT_SERVER")
+api_key = get_env_var("CONNECT_API_KEY")
+canary_guid = get_env_var("CANARY_GUID", "It should be the GUID of the content you wish to monitor.")
# Instantiate a Connect client using posit-sdk where api_key and url are automatically read from our environment vars
client = connect.Client()
@@ -56,6 +55,11 @@ client = connect.Client()
#| echo: false
#| label: data-gathering
+# Define status constants
+STATUS_PASS = "PASS"
+STATUS_FAIL = "FAIL"
+ERROR_PREFIX = "ERROR:"
+
if not show_instructions:
# Proceed to validate our monitored GUIDS
@@ -77,6 +81,29 @@ if not show_instructions:
except requests.exceptions.RequestException as e:
raise RuntimeError(f"Connect server at {connect_server} is unavailable: {str(e)}")
+ # Helper function to extract error messages from exceptions
+ def format_error_message(exception):
+ """Extract a clean error message from various exception types"""
+ error_message = str(exception)
+
+ # posit-sdk will return a ClientError if there is a problem getting the guid, parse the error message
+ if isinstance(exception, connect.errors.ClientError):
+ try:
+ # ClientError from posit-connect SDK stores error as string that contains JSON
+ # Convert the string representation to a dict
+ error_data = json.loads(str(exception))
+ if isinstance(error_data, dict):
+ # Extract the specific error message
+ if "error_message" in error_data:
+ error_message = error_data["error_message"]
+ elif "error" in error_data:
+ error_message = error_data["error"]
+ except json.JSONDecodeError:
+ # If parsing fails, keep the original error message
+ pass
+
+ return error_message
+
# Function to get app details from Connect API
def get_content(guid):
try:
@@ -84,28 +111,12 @@ if not show_instructions:
content = client.content.get(guid)
return content
except Exception as e:
- # Initialize default error message
- error_message = str(e)
-
- # posit-sdk will return a ClientError if there is a problem getting the guid, parse the error message
- if isinstance(e, connect.errors.ClientError):
- try:
- # ClientError from posit-connect SDK stores error as string that contains JSON
- # Convert the string representation to a dict
- error_data = json.loads(str(e))
- if isinstance(error_data, dict):
- # Extract the specific error message
- if "error_message" in error_data:
- error_message = error_data["error_message"]
- elif "error" in error_data:
- error_message = error_data["error"]
- except json.JSONDecodeError:
- # If parsing fails, keep the original error message
- pass
+ # Extract error message and return error object
+ error_message = format_error_message(e)
# Return content with error in title
return {
- "title": f"ERROR: {error_message}",
+ "title": f"{ERROR_PREFIX} {error_message}",
"guid": guid
}
@@ -114,7 +125,8 @@ if not show_instructions:
user = client.users.get(user_guid)
return user
except Exception as e:
- raise RuntimeError(f"Error getting user: {str(e)}")
+ error_message = format_error_message(e)
+ raise RuntimeError(f"Error getting user: {error_message}")
# Function to validate app health (simple HTTP 200 check)
def validate_app(guid):
@@ -202,13 +214,32 @@ if not show_instructions:
"http_code": str(e)
}
- # Check all apps and collect results
- results = []
- for guid in canary_guids:
- results.append(validate_app(guid))
-
- # Convert results to DataFrame for easy display
- df = pd.DataFrame(results)
+ # Helper function to check if a result has an error
+ def has_error(result):
+ """Check if a result contains an error message in the name field"""
+ return result["name"].startswith(ERROR_PREFIX) if result and "name" in result else False
+
+ # Helper function to extract error details from a result
+ def extract_error_details(result):
+ """Extract error message and GUID from a result"""
+ if has_error(result):
+ return {
+ "message": result["name"].replace(f"{ERROR_PREFIX} ", ""),
+ "guid": result.get("guid", "")
+ }
+ return None
+
+ # Check the app and get result
+ app_result = None
+ if canary_guid:
+ app_result = validate_app(canary_guid)
+
+ # Check if there was an error with the GUID
+ if has_error(app_result):
+ show_error = True
+ error_details = extract_error_details(app_result)
+ error_message = error_details["message"]
+ error_guid = error_details["guid"]
# Store the current time
check_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
@@ -219,166 +250,250 @@ if not show_instructions:
#| echo: false
#| label: display
-about = """
-This report uses the publisher's API key to monitor a list of application GUIDs, verifies each of the monitored apps are
-reachable at their content URL, and reports the results. If run on a schedule, reports will be emailed when any monitored
-application reports a failure.
-
"""
-
-if not show_instructions and not df.empty:
-
- # Format the canary_guids as a string for display
- canary_guids_str = ", ".join(canary_guids)
-
- # Use HTML to create a callout box
- display(HTML(f"""
-
-
App Canary Monitor
-
{about}
-
Monitored GUIDs:
-
{canary_guids_str}
+# Define CSS styling constants
+CSS_COLORS = {
+ "neutral": {
+ "border": "#ccc",
+ "background": "#f8f9fa",
+ "text": "#000"
+ },
+ "error": {
+ "border": "#cc0000",
+ "background": "#fff8f8",
+ "text": "#cc0000"
+ },
+ "warning": {
+ "border": "#f0ad4e",
+ "background": "#fcf8e3",
+ "text": "#cc0000"
+ },
+ "success": {
+ "border": "#4caf50",
+ "background": "#e8f5e9",
+ "text": "#28a745"
+ },
+ "fail": {
+ "border": "#f44336",
+ "background": "#ffebee",
+ "text": "#dc3545"
+ }
+}
+
+CSS_BOX_STYLE = "border: 1px solid {border}; border-radius: 8px; padding: 10px; margin-bottom: 15px; background-color: {background};"
+CSS_HEADER_STYLE = "margin-top: 0; padding-bottom: 8px; border-bottom: 1px solid #eaecef; font-weight: bold; font-size: 1.2em;"
+CSS_CONTENT_STYLE = "padding: 5px 0;"
+CSS_FOOTER_STYLE = "padding-top: 8px; font-size: 0.9em; border-top: 1px solid #eaecef;"
+CSS_GRID_STYLE = "display: grid; grid-template-columns: 150px auto; grid-gap: 8px; padding: 10px 0;"
+
+# Define helper functions for generating HTML components
+
+def create_about_box(about_content):
+ """Creates the About callout box HTML"""
+ neutral = CSS_COLORS["neutral"]
+ return f"""
+
+
âšī¸ About
+
{about_content}
- """))
-
- # First create links for name and guid columns
- df_display = df.copy()
-
- # Process the DataFrame rows to add HTML links and format owner info
- for i in range(len(df_display)):
- # Format app name links only (not guid)
- if 'dashboard_url' in df_display.columns and not pd.isna(df_display.loc[i, 'dashboard_url']) and df_display.loc[i, 'dashboard_url']:
- url = df_display.loc[i, 'dashboard_url']
- app_name = df_display.loc[i, 'name']
- app_guid = df_display.loc[i, 'guid']
-
- # Create simple markdown link for name only
- df_display.loc[i, 'name'] = f"
{app_name}"
+ """
+
+def create_error_box(guid, error_msg):
+ """Creates the Error callout box HTML"""
+ error = CSS_COLORS["error"]
+ return f"""
+
+
â ī¸ Error Monitoring Content
- # Format logs URL using markdown instead of HTML (similar to owner email)
- if 'logs_url' in df_display.columns and df_display.loc[i, 'logs_url'] is not None and str(df_display.loc[i, 'logs_url']).strip():
- # Use a simple icon since we can't use custom SVG in emails easily
- logs_icon = "đ"
- logs_url = df_display.loc[i, 'logs_url']
- df_display.loc[i, 'logs'] = f"
{logs_icon}"
-
- else:
- df_display.loc[i, 'logs'] = ""
-
- # Format owner name with email icon link
- owner_name = df_display.loc[i, 'owner_name'] if not pd.isna(df_display.loc[i, 'owner_name']) else "Unknown"
- owner_email = df_display.loc[i, 'owner_email'] if not pd.isna(df_display.loc[i, 'owner_email']) else ""
+
+
Content GUID:
+
{guid}
+
+
Error:
+
{error_msg}
+
- if owner_email:
- # Use a simple icon since we can't use custom SVG in emails easily
- email_icon = "âī¸"
- df_display.loc[i, 'owner_display'] = f"{owner_name}
{email_icon}"
- else:
- df_display.loc[i, 'owner_display'] = owner_name
+
+ Please check that your CANARY_GUID environment variable contains a valid content identifier.
+
+
+ """
+
+def create_no_results_box():
+ """Creates the No Results callout box HTML"""
+ warning = CSS_COLORS["warning"]
+ error_text = CSS_COLORS["error"]["text"]
+ return f"""
+
+
â ī¸ No results available
+
+ No monitoring results were found. This could be because:
+
+ - No valid GUID was provided
+ - There was an issue connecting to the specified content
+ - The environment is properly configured but there was an error that caused no data to be returned
+
+
Please check your CANARY_GUID environment variable and ensure it contains a valid content identifier.
+
+
+ """
+
+def create_instructions_box(instructions_html_content):
+ """Creates the Setup Instructions callout box HTML"""
+ error = CSS_COLORS["error"]
+ return f"""
+
+
â ī¸ Setup Instructions
+ {instructions_html_content}
+
+
+ """
+
+# Function to create the report display for a result
+def create_report_display(result_data, check_time_value):
+ # Format app name with link if dashboard_url is available
+ app_name = result_data.get('name', 'Unknown')
+ dashboard_url = result_data.get('dashboard_url', '')
+ if dashboard_url:
+ app_name_display = f"
{app_name}"
+ else:
+ app_name_display = app_name
- # Remove dashboard_url column since the links are embedded in the other columns
- if 'dashboard_url' in df_display.columns:
- df_display = df_display.drop(columns=['dashboard_url'])
+ # Get GUID
+ app_guid = result_data.get('guid', '')
- # Remove raw owner columns in favor of our formatted display column
- if 'owner_name' in df_display.columns:
- df_display = df_display.drop(columns=['owner_name'])
- if 'owner_email' in df_display.columns:
- df_display = df_display.drop(columns=['owner_email'])
+ # Format status with appropriate color
+ status = result_data.get('status', '')
- # Reorder columns to match requested layout
- column_order = ['name', 'guid', 'status', 'http_code', 'logs', 'owner_display']
- # Only include columns that exist in df_display
- ordered_columns = [col for col in column_order if col in df_display.columns]
- df_display = df_display[ordered_columns]
+ if status == STATUS_PASS:
+ status_colors = CSS_COLORS["success"]
+ status_icon = "â
" # Checkmark
+ else:
+ status_colors = CSS_COLORS["fail"]
+ status_icon = "â" # X mark
- # Create GT table
- gt_tbl = GT(df_display)
+ # Get HTTP Code
+ http_code = result_data.get('http_code', '')
- # Tell great_tables to render markdown content in these columns
- gt_tbl = gt_tbl.fmt_markdown(columns=['name', 'owner_display', 'logs'])
+ # Format logs link if available
+ logs_url = result_data.get('logs_url', '')
+ if logs_url:
+ logs_display = f"
đ View Logs"
+ else:
+ logs_display = "No logs available (only available to owner and collaborators)"
- # Apply styling to columns
- gt_tbl = (gt_tbl
- # Status column styling - green for PASS, red for FAIL
- .tab_style(
- style.fill("green"),
- locations=loc.body(columns="status", rows=lambda df: df["status"] == "PASS")
- )
- .tab_style(
- style.fill("red"),
- locations=loc.body(columns="status", rows=lambda df: df["status"] == "FAIL")
- )
- # Set column labels for better presentation
- .cols_label(
- owner_display="Owner",
- name="Name",
- guid="Content GUID",
- status="Status",
- http_code="HTTP Code",
- logs="Logs"
- )
- # Override default column alignment for better presentation
- .cols_align(
- align='center',
- columns=["status", "http_code", "logs"]
- )
- .tab_options(
- container_width="100%"
- )
- )
-
-elif show_instructions:
- # Create a callout box for instructions
- instructions_html = about # Start with the about message
-
- for instruction in instructions:
- instructions_html += f"
{instruction}
"
+ # Format owner information
+ owner_name = result_data.get('owner_name', 'Unknown')
+ owner_email = result_data.get('owner_email', '')
+ if owner_email:
+ owner_display = f"{owner_name}
âī¸"
+ else:
+ owner_display = owner_name
- display(HTML(f"""
-
-
â ī¸ Setup Instructions
- {instructions_html}
-
-
- """))
+ # Create last check time display
+ check_time_display = f"Last checked: {check_time_value}"
- # Set gt_tbl to None since we're using HTML display instead
- gt_tbl = None
-else:
- # We should only hit this catchall if the dataframe is empty (a likely error) and there are no instructions
- display(HTML(f"""
-
-
â ī¸ No results available
-
- No monitoring results were found. This could be because:
-
- - No valid GUIDs were provided
- - There was an issue connecting to the specified content
- - The environment is properly configured but there was an error that caused no data to be returned
-
-
Please check your CANARY_GUIDS environment variable and ensure it contains valid content identifiers.
+ html_output = f"""
+
+
+ {status_icon} {status} - Application Status
+
+
+
+
Name:
+
{app_name_display}
+
+
Content GUID:
+
{app_guid}
+
+
HTTP Code:
+
{http_code}
+
+
Logs:
+
{logs_display}
+
+
Owner:
+
{owner_display}
+
+
+
+ {check_time_display}
- """))
+ """
- # Set gt_tbl to None since we're using HTML display instead
- gt_tbl = None
+ return html_output
+
+# Create the about content text
+about_content = """
+This report uses the publisher's API key to monitor an application GUID, verifies that the monitored app is
+reachable at its content URL, and reports the results. If run on a schedule, reports will be emailed when the monitored
+application reports a failure.
+
"""
+
+# Store HTML components in variables for reuse
+about_box_html = create_about_box(about_content)
+error_box_html = ""
+no_results_box_html = ""
+report_box_html = ""
+
+# Generate the appropriate HTML based on the state
+if show_error:
+ # Create error box HTML
+ error_box_html = create_error_box(error_guid, error_message)
+elif app_result and not has_error(app_result):
+ # Create report box HTML using the function
+ report_box_html = create_report_display(app_result, check_time)
+else:
+ # Create no results box HTML
+ no_results_box_html = create_no_results_box()
+
+# Always display the About callout box
+display(HTML(about_box_html))
-# Compute if we should send an email, only send if at least one app has a failure
-if 'df' in locals() and 'status' in df.columns:
- send_email = bool(not df.empty and (df['status'] == 'FAIL').any())
+# Display the appropriate content based on the state
+if show_instructions:
+ # Create a callout box for instructions
+ instructions_html = ""
+ for instruction in instructions:
+ instructions_html += f"
{instruction}
"
+
+ # Create and display instructions box
+ instructions_box_html = create_instructions_box(instructions_html)
+ display(HTML(instructions_box_html))
+elif show_error:
+ # Display the error box that was already created
+ display(HTML(error_box_html))
+elif report_box_html:
+ # Display the report box that was already created
+ display(HTML(report_box_html))
else:
- send_email = False
+ # Display the no results box that was already created
+ display(HTML(no_results_box_html))
+
+# Helper function to determine if we should send an email
+def should_send_email():
+ """Determine if we should send an email notification"""
+ # Send email if we have an API error
+ if show_error:
+ return True
+
+ # Send email if the monitored app has a failure status
+ if app_result and 'status' in app_result:
+ return app_result['status'] == STATUS_FAIL
+
+ return False
+
+# Set send_email variable for the quarto email mechanism
+send_email = should_send_email()
```
```{python}
#| echo: false
-# Display the table in the rendered document HTML, email is handled separately below
-gt_tbl
+# No need to display anything here since we're displaying HTML directly above
```
@@ -389,11 +504,23 @@ gt_tbl
:::
::: {.subject}
-App Canary - â one or more apps have failed monitoring
+App Canary - â The monitored app has failed
:::
```{python}
#| echo: false
-gt_tbl
+
+# Always show the about section in the email
+display(HTML(about_box_html))
+
+# Display the appropriate content based on the state - use the same HTML as the main report
+if show_instructions:
+ display(HTML(instructions_box_html))
+elif show_error:
+ display(HTML(error_box_html))
+elif report_box_html:
+ display(HTML(report_box_html))
+else:
+ display(HTML(no_results_box_html))
```
:::