94 lines
2.5 KiB
Python
94 lines
2.5 KiB
Python
"""
|
|
Example Shunt - Template for creating fake service connectors.
|
|
|
|
Copy this to create a new shunt:
|
|
cp -r shunts/example shunts/mercadopago
|
|
|
|
Then customize:
|
|
- Update responses.json with fake responses
|
|
- Add endpoints matching the real vein's API
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
|
|
from fastapi import FastAPI, Request
|
|
from fastapi.responses import HTMLResponse, JSONResponse
|
|
|
|
app = FastAPI(title="Example Shunt", description="Fake service for testing")
|
|
|
|
BASE_DIR = Path(__file__).parent
|
|
DEPOT_DIR = Path(os.getenv("SHUNT_DEPOT", str(BASE_DIR / "depot")))
|
|
|
|
# Load responses
|
|
RESPONSES_FILE = DEPOT_DIR / "responses.json"
|
|
responses = {}
|
|
if RESPONSES_FILE.exists():
|
|
responses = json.loads(RESPONSES_FILE.read_text())
|
|
|
|
|
|
@app.get("/health")
|
|
def health():
|
|
return {"status": "ok", "shunt": "example"}
|
|
|
|
|
|
@app.get("/", response_class=HTMLResponse)
|
|
def config_ui():
|
|
"""Simple UI to view/edit responses."""
|
|
return f"""
|
|
<!DOCTYPE html>
|
|
<html>
|
|
<head>
|
|
<title>Example Shunt</title>
|
|
<style>
|
|
body {{ font-family: system-ui; background: #111827; color: #f3f4f6; padding: 2rem; }}
|
|
h1 {{ color: #60a5fa; }}
|
|
pre {{ background: #1f2937; padding: 1rem; border-radius: 8px; overflow-x: auto; }}
|
|
.info {{ color: #9ca3af; }}
|
|
</style>
|
|
</head>
|
|
<body>
|
|
<h1>Example Shunt</h1>
|
|
<p class="info">This shunt returns configurable fake responses for testing.</p>
|
|
<h2>Current Responses</h2>
|
|
<pre>{json.dumps(responses, indent=2)}</pre>
|
|
<p class="info">Edit {RESPONSES_FILE} to change responses.</p>
|
|
</body>
|
|
</html>
|
|
"""
|
|
|
|
|
|
@app.get("/api/{endpoint:path}")
|
|
def fake_get(endpoint: str):
|
|
"""Return configured response for GET requests."""
|
|
key = f"GET /{endpoint}"
|
|
if key in responses:
|
|
return responses[key]
|
|
return {"error": "not configured", "endpoint": endpoint, "method": "GET"}
|
|
|
|
|
|
@app.post("/api/{endpoint:path}")
|
|
async def fake_post(endpoint: str, request: Request):
|
|
"""Return configured response for POST requests."""
|
|
key = f"POST /{endpoint}"
|
|
if key in responses:
|
|
return responses[key]
|
|
body = (
|
|
await request.json()
|
|
if request.headers.get("content-type") == "application/json"
|
|
else {}
|
|
)
|
|
return {
|
|
"error": "not configured",
|
|
"endpoint": endpoint,
|
|
"method": "POST",
|
|
"received": body,
|
|
}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
|
|
uvicorn.run(app, host="0.0.0.0", port=8099)
|