Replies: 2 comments
-
import httpx
from httpx._transports.default import (
httpcore, typing, map_httpcore_exceptions, ResponseStream,
HTTPTransport, SyncByteStream, Request, Response
)
class CustomHostsTransport(HTTPTransport):
def __init__(self, *args, domains={}, **kwargs) -> None:
self.domains = {}
for k, v in domains.items():
self.domains[k.encode()] = v.encode()
super().__init__(*args, **kwargs)
def handle_request(self, request: Request) -> Response:
assert isinstance(request.stream, SyncByteStream)
host = request.url.raw_host
req = httpcore.Request(
method=request.method,
url=httpcore.URL(
scheme=request.url.raw_scheme,
host=self.domains.get(host, host),
port=request.url.port,
target=request.url.raw_path,
),
headers=request.headers.raw,
content=request.stream,
extensions=request.extensions,
)
with map_httpcore_exceptions():
resp = self._pool.handle_request(req)
assert isinstance(resp.stream, typing.Iterable)
return Response(
status_code=resp.status,
headers=resp.headers,
stream=ResponseStream(resp.stream),
extensions=resp.extensions,
)
with httpx.Client(transport=CustomHostsTransport(domains={'dns.google': '8.8.8.8'})) as c:
c.get('https://dns.google') |
Beta Was this translation helpful? Give feedback.
0 replies
-
Thanks to encode/httpcore#696 from httpx import Request, Response, HTTPTransport, AsyncHTTPTransport
class NameSolver:
def get(self, name: str) -> str:
if name.endswith('.google'):
return '8.8.8.8'
return ''
def resolve(self, request: Request) -> Request:
host = request.url.host
ip = self.get(host)
if ip:
request.extensions["sni_hostname"] = host
request.url = request.url.copy_with(host=ip)
return request
class CustomHost(HTTPTransport):
def __init__(self, solver: NameSolver, *args, **kwargs) -> None:
self.solver = solver
super().__init__(*args, **kwargs)
def handle_request(self, request: Request) -> Response:
request = self.solver.resolve(request)
return super().handle_request(request)
class AsyncCustomHost(AsyncHTTPTransport):
def __init__(self, solver: NameSolver, *args, **kwargs) -> None:
self.solver = solver
super().__init__(*args, **kwargs)
async def handle_async_request(self, request: Request) -> Response:
request = self.solver.resolve(request)
return await super().handle_async_request(request) usage: import httpx
with httpx.Client(transport=CustomHost(NameSolver())) as client:
r = client.get("https://dns.google") you can create subclass of |
Beta Was this translation helpful? Give feedback.
0 replies
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
-
I want to set specified IP for few hosts directly using httpx for sending requests.
My ISP filtered dns requests for Google's DoH
dns.google
.I want to address this issue to automatically resolve as
8.8.8.8
when I sending requests through httpx.I tried edit Hosts file of OS and it seems it working, but isn't there any direct way?
Beta Was this translation helpful? Give feedback.
All reactions