113 lines
3.5 KiB
Python
113 lines
3.5 KiB
Python
"""Tiny Python proxy for the LASUCA shared endpoint."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import time
|
|
from typing import Any, Dict, Optional
|
|
|
|
import requests
|
|
from flask import Flask, Response, abort, jsonify
|
|
|
|
DEFAULT_ENDPOINT = "http://localhost/shared-endpoint/public/index.php"
|
|
DEFAULT_CACHE_SECONDS = 1
|
|
DEFAULT_TIMEOUT_SECONDS = 3
|
|
|
|
app = Flask(__name__)
|
|
|
|
|
|
def _bool_env(var_name: str, default: bool = False) -> bool:
|
|
value = os.getenv(var_name)
|
|
if value is None:
|
|
return default
|
|
return value.strip().lower() in {"1", "true", "yes", "on"}
|
|
|
|
|
|
def get_settings() -> Dict[str, Any]:
|
|
return {
|
|
"endpoint": os.getenv("LASUCA_PHP_FEED", DEFAULT_ENDPOINT),
|
|
"cache_seconds": int(os.getenv("LASUCA_PROXY_CACHE_SECONDS", DEFAULT_CACHE_SECONDS)),
|
|
"timeout": float(os.getenv("LASUCA_PROXY_TIMEOUT", DEFAULT_TIMEOUT_SECONDS)),
|
|
"forward_headers": _bool_env("LASUCA_PROXY_FORWARD_HEADERS", False),
|
|
}
|
|
|
|
|
|
class CacheEntry:
|
|
__slots__ = ("payload", "timestamp", "etag")
|
|
|
|
def __init__(self, payload: Dict[str, Any], etag: Optional[str]) -> None:
|
|
self.payload = payload
|
|
self.timestamp = time.time()
|
|
self.etag = etag
|
|
|
|
|
|
_cache: Dict[str, CacheEntry] = {}
|
|
|
|
|
|
def _build_cache_key(settings: Dict[str, Any]) -> str:
|
|
endpoint = settings["endpoint"]
|
|
forward_headers = settings["forward_headers"]
|
|
return f"{endpoint}|fh={int(forward_headers)}"
|
|
|
|
|
|
def _http_get(settings: Dict[str, Any]) -> requests.Response:
|
|
headers = {}
|
|
if settings["forward_headers"]:
|
|
for header in ("Authorization", "X-Forwarded-For", "Cookie"):
|
|
value = os.getenv(f"LASUCA_PROXY_HEADER_{header.replace('-', '_').upper()}")
|
|
if value:
|
|
headers[header] = value
|
|
response = requests.get(settings["endpoint"], timeout=settings["timeout"], headers=headers)
|
|
response.raise_for_status()
|
|
return response
|
|
|
|
|
|
@app.route("/healthz", methods=["GET"])
|
|
def health_check() -> Response:
|
|
return jsonify({"status": "ok"})
|
|
|
|
|
|
@app.route("/items", methods=["GET"])
|
|
def items() -> Response:
|
|
settings = get_settings()
|
|
cache_key = _build_cache_key(settings)
|
|
now = time.time()
|
|
cache_entry = _cache.get(cache_key)
|
|
|
|
if cache_entry:
|
|
age = now - cache_entry.timestamp
|
|
if age <= settings["cache_seconds"]:
|
|
response = jsonify(cache_entry.payload)
|
|
if cache_entry.etag is not None:
|
|
response.headers["ETag"] = cache_entry.etag
|
|
response.headers["X-Cache-Hit"] = "1"
|
|
response.headers["Cache-Control"] = f"max-age={settings['cache_seconds']}"
|
|
return response
|
|
|
|
try:
|
|
upstream_response = _http_get(settings)
|
|
payload = upstream_response.json()
|
|
except requests.RequestException as exc:
|
|
abort(502, f"Shared endpoint failed: {exc}")
|
|
except ValueError as exc:
|
|
abort(502, f"Invalid JSON from shared endpoint: {exc}")
|
|
|
|
if payload.get("status") != "ok":
|
|
abort(502, payload.get("message", "Shared endpoint error"))
|
|
|
|
etag = upstream_response.headers.get("ETag")
|
|
_cache[cache_key] = CacheEntry(payload, etag)
|
|
|
|
response = jsonify(payload)
|
|
if etag is not None:
|
|
response.headers["ETag"] = etag
|
|
response.headers["X-Cache-Hit"] = "0"
|
|
response.headers["Cache-Control"] = f"max-age={settings['cache_seconds']}"
|
|
return response
|
|
|
|
|
|
if __name__ == "__main__":
|
|
host = os.getenv("LASUCA_PROXY_HOST", "0.0.0.0")
|
|
port = int(os.getenv("LASUCA_PROXY_PORT", "5050"))
|
|
debug = _bool_env("LASUCA_PROXY_DEBUG", False)
|
|
app.run(host=host, port=port, debug=debug)
|