mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-12 13:29:04 +00:00
119ff3740c
@task_handler never built or passed cloud_logs_url, so every app using it (incl. modelling_e2e) ran run_subtask with the None default and the CloudWatch deep-link was never saved onto the SubTask. @subtask_handler did this correctly. Extract the URL builder into a shared utilities/aws_lambda/cloud_logs.py (public cloudwatch_url()), use it from both handlers, and pass the URL into run_subtask from @task_handler. Add regression tests. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
27 lines
935 B
Python
27 lines
935 B
Python
"""Build a CloudWatch console deep-link for the running Lambda invocation.
|
|
|
|
Shared by @task_handler and @subtask_handler so both persist the same
|
|
`cloud_logs_url` onto the SubTask they run.
|
|
"""
|
|
|
|
import os
|
|
from typing import Optional
|
|
from urllib.parse import quote
|
|
|
|
|
|
def _console_encode(value: str) -> str:
|
|
return quote(value, safe="").replace("%", "$25")
|
|
|
|
|
|
def cloudwatch_url() -> Optional[str]:
|
|
"""Deep-link to this invocation's log stream, or None outside Lambda."""
|
|
region = os.environ.get("AWS_REGION")
|
|
log_group = os.environ.get("AWS_LAMBDA_LOG_GROUP_NAME")
|
|
log_stream = os.environ.get("AWS_LAMBDA_LOG_STREAM_NAME")
|
|
if not (region and log_group and log_stream):
|
|
return None
|
|
return (
|
|
f"https://{region}.console.aws.amazon.com/cloudwatch/home"
|
|
f"?region={region}#logsV2:log-groups/log-group/"
|
|
f"{_console_encode(log_group)}/log-events/{_console_encode(log_stream)}"
|
|
)
|