mirror of
https://github.com/Sea-Haven-Industries/meal-order-manager.git
synced 2026-10-07 16:18:54 +00:00
Adam Moussa
f0912e56df
Use SigV4 requests for settings and menu publication so the scheduled workflow no longer needs direct DynamoDB access.
102 lines
2.8 KiB
Python
102 lines
2.8 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:
|
|
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())
|