An ASGI proxy service.

proxy.py

import httpx
from starlette.requests import Request
from starlette.responses import StreamingResponse


class Proxy:
    def __init__(self, hostname):
        self.hostname = hostname
        self.client = httpx.AsyncClient()

    async def __call__(self, scope, receive, send):
        assert scope['type'] == 'http'
        request = Request(scope, receive=receive)
        async with self.client.stream(
            method=self.get_method(request),
            url=self.get_url(request),
            headers=self.get_headers(request),
            data=self.get_body(request),
            allow_redirects=False
        ) as response:
            app = StreamingResponse(
                status_code=response.status_code,
                headers=response.headers,
                content=response.aiter_raw()
            )
            await app(scope, receive, send)

    def get_method(self, request):
        return request.method

    def get_url(self, request):
        return str(request.url.replace(scheme="https", netloc=self.hostname))

    def get_headers(self, request):
        return [(key, value) for key, value in request.headers.raw if key != b'host']

    def get_body(self, request):
        if 'content-length' in request.headers or 'transfer-encoding' in request.headers:
            return request.stream()
        return None


app = Proxy(hostname="www.google.com")

xref: encode/starlette#241