mirror of
https://github.com/Sea-Haven-Industries/meal-order-manager.git
synced 2026-09-30 22:53:12 +00:00
88399fe153
* feat(api): add IAM-authenticated menu publication Keep weekly menu writes behind Lambda so the GitHub runtime role cannot access the shared DynamoDB table directly. * refactor(workflow): publish weekly menus through API Use SigV4 requests for settings and menu publication so the scheduled workflow no longer needs direct DynamoDB access. * fix: address review comments * style(python): apply Ruff formatting
102 lines
2.9 KiB
Python
102 lines
2.9 KiB
Python
"""Call the IAM-protected weekly-menu publication API with SigV4."""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
import boto3
|
|
from botocore.auth import SigV4Auth
|
|
from botocore.awsrequest import AWSRequest
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
|
OUTPUT_DIR = PROJECT_ROOT / "output"
|
|
|
|
|
|
def signed_request(
|
|
api_url: str,
|
|
path: str,
|
|
method: str = "GET",
|
|
body: dict | None = None,
|
|
region: str = "us-east-1",
|
|
) -> dict:
|
|
if not api_url.startswith(("http://", "https://")):
|
|
raise ValueError(f"api_url must use http(s) scheme: {api_url!r}")
|
|
data = json.dumps(body).encode() if body is not None else None
|
|
headers = {"Content-Type": "application/json"} if data is not None else {}
|
|
request = AWSRequest(
|
|
method=method,
|
|
url=f"{api_url.rstrip('/')}{path}",
|
|
data=data,
|
|
headers=headers,
|
|
)
|
|
credentials = boto3.Session().get_credentials()
|
|
if credentials is None:
|
|
raise RuntimeError("AWS credentials are required")
|
|
SigV4Auth(credentials.get_frozen_credentials(), "execute-api", region).add_auth(
|
|
request
|
|
)
|
|
|
|
prepared = request.prepare()
|
|
http_request = urllib.request.Request(
|
|
prepared.url,
|
|
data=prepared.body,
|
|
headers=dict(prepared.headers),
|
|
method=method,
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(http_request, timeout=30) as response:
|
|
return json.loads(response.read())
|
|
except urllib.error.HTTPError as exc:
|
|
detail = exc.read().decode(errors="replace")
|
|
raise RuntimeError(
|
|
f"Publication API returned HTTP {exc.code}: {detail}"
|
|
) from exc
|
|
|
|
|
|
def get_settings(api_url: str, region: str) -> dict:
|
|
return signed_request(api_url, "/api/publish/settings", method="GET", region=region)
|
|
|
|
|
|
def publish_menu(api_url: str, region: str) -> dict:
|
|
files = sorted(OUTPUT_DIR.glob("menu-*.json"), reverse=True)
|
|
if not files:
|
|
raise RuntimeError("No menu JSON found. Run scrape_menu.py first.")
|
|
|
|
with files[0].open() as menu_file:
|
|
menu = json.load(menu_file)
|
|
return signed_request(
|
|
api_url,
|
|
"/api/publish/menu",
|
|
method="POST",
|
|
body=menu,
|
|
region=region,
|
|
)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("action", choices=("settings", "publish"))
|
|
parser.add_argument("--api-url", required=True)
|
|
parser.add_argument("--region", default=os.environ.get("AWS_REGION", "us-east-1"))
|
|
args = parser.parse_args()
|
|
|
|
try:
|
|
result = (
|
|
get_settings(args.api_url, args.region)
|
|
if args.action == "settings"
|
|
else publish_menu(args.api_url, args.region)
|
|
)
|
|
except (OSError, RuntimeError, ValueError, json.JSONDecodeError) as exc:
|
|
print(f"Error: {exc}", file=sys.stderr)
|
|
return 1
|
|
|
|
print(json.dumps(result))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|