Skip to content

Instantly share code, notes, and snippets.

@simonw

simonw/main.py Secret

Created March 26, 2023 16:16
Show Gist options
  • Save simonw/e0a5368d8b465febefe6d4ae6f47f2da to your computer and use it in GitHub Desktop.
Save simonw/e0a5368d8b465febefe6d4ae6f47f2da to your computer and use it in GitHub Desktop.
import os
from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.middleware.cors import CORSMiddleware
from starlette.routing import Route
from starlette.requests import Request
from starlette.responses import JSONResponse
import httpx
TARGET_SERVER_HOST = os.environ.get("TARGET_SERVER_HOST")
middleware = [
Middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
]
app = Starlette(middleware=middleware)
async def forward_request(request: Request):
try:
async with httpx.AsyncClient() as client:
target_url = f"{TARGET_SERVER_HOST}{request.url.path}"
method = request.method
headers = dict(request.headers)
if "host" in headers:
del headers["host"]
if "content-length" in headers:
del headers["content-length"]
if request.method in ["POST", "PUT", "PATCH"]:
request_body = await request.json()
else:
request_body = None
response = await client.request(
method, target_url, headers=headers, json=request_body
)
return JSONResponse(response.json(), status_code=response.status_code)
except httpx.HTTPError as e:
return JSONResponse({"error": str(e)}, status_code=500)
app.add_route(
"/{path:path}", forward_request, methods=["GET", "POST", "PUT", "PATCH", "DELETE"]
)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment