mirror of
https://github.com/Sea-Haven-Industries/meal-order-manager.git
synced 2026-09-30 17:03:13 +00:00
Adam Moussa
3e9a1c60bf
Keep the Flask app always-on with in-process jobs so CloudFront no longer fronts a cold-start API Gateway.
94 lines
2.6 KiB
Python
94 lines
2.6 KiB
Python
"""Call the HMAC-protected weekly-menu publication API."""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
|
OUTPUT_DIR = PROJECT_ROOT / "output"
|
|
|
|
|
|
def api_request(
|
|
api_url: str,
|
|
path: str,
|
|
method: str = "GET",
|
|
body: dict | None = None,
|
|
publish_key: str = "",
|
|
) -> 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 = {}
|
|
if data is not None:
|
|
headers["Content-Type"] = "application/json"
|
|
if publish_key:
|
|
headers["X-Meals-Publish-Key"] = publish_key
|
|
http_request = urllib.request.Request(
|
|
f"{api_url.rstrip('/')}{path}",
|
|
data=data,
|
|
headers=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, publish_key: str) -> dict:
|
|
return api_request(
|
|
api_url, "/api/publish/settings", method="GET", publish_key=publish_key
|
|
)
|
|
|
|
|
|
def publish_menu(api_url: str, publish_key: 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 api_request(
|
|
api_url,
|
|
"/api/publish/menu",
|
|
method="POST",
|
|
body=menu,
|
|
publish_key=publish_key,
|
|
)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("action", choices=("settings", "publish"))
|
|
parser.add_argument("--api-url", required=True)
|
|
parser.add_argument(
|
|
"--publish-key",
|
|
default=os.environ.get("MEALS_PUBLISH_KEY", ""),
|
|
help="Shared secret for /api/publish/* (or MEALS_PUBLISH_KEY)",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
try:
|
|
result = (
|
|
get_settings(args.api_url, args.publish_key)
|
|
if args.action == "settings"
|
|
else publish_menu(args.api_url, args.publish_key)
|
|
)
|
|
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())
|