| | from contextlib import asynccontextmanager |
| | from typing import AsyncIterator |
| |
|
| | from fastapi import FastAPI |
| | from fastapi_proxy_lib.core.http import ForwardHttpProxy |
| | from fastapi_proxy_lib.core.tool import default_proxy_filter |
| | from httpx import AsyncClient |
| | from starlette.requests import Request |
| |
|
| | proxy = ForwardHttpProxy(AsyncClient(), proxy_filter=default_proxy_filter) |
| |
|
| | @asynccontextmanager |
| | async def close_proxy_event(_: FastAPI) -> AsyncIterator[None]: |
| | """Close proxy.""" |
| | yield |
| | await proxy.aclose() |
| |
|
| | app = FastAPI(lifespan=close_proxy_event) |
| |
|
| | @app.api_route("/{path:path}", methods=["GET", "POST", "PUT", "DELETE", "PATCH", "HEAD", "OPTIONS", "CONNECT"]) |
| | async def _(request: Request, path: str = ""): |
| | return await proxy.proxy(request=request, path=path) |
| |
|
| |
|
| | if __name__ == '__main__': |
| | import uvicorn |
| | uvicorn.run("run:app", host="0.0.0.0", port="8080") |
| |
|