- name: Install dependencies
run: |
python -m pip install --upgrade pip
- python -m pip install cryptography trio idna httpx h2 aioquic pytest ruff pyright ty
+ python -m pip install cryptography trio idna httpx2 h2 aioquic pytest ruff pyright ty
- name: Typecheck
run: |
pyright dns
if dns._features.have("doh"):
import anyio
- import httpcore
- import httpcore._backends.anyio
- import httpx
+ import httpcore2
+ import httpcore2._backends.anyio
+ import httpx2
- _CoreAsyncNetworkBackend = httpcore.AsyncNetworkBackend
- _CoreAnyIOStream = httpcore._backends.anyio.AnyIOStream # pyright: ignore
+ _CoreAsyncNetworkBackend = httpcore2.AsyncNetworkBackend
+ _CoreAnyIOStream = httpcore2._backends.anyio.AnyIOStream # pyright: ignore
from dns.query import _compute_times, _expiration_for_this_attempt, _remaining
self._family = family
if local_port != 0:
raise NotImplementedError(
- "the asyncio transport for HTTPX cannot set the local port"
+ "the asyncio transport for httpx2 cannot set the local port"
)
async def connect_tcp(
return _CoreAnyIOStream(stream)
except Exception:
pass
- raise httpcore.ConnectError
+ raise httpcore2.ConnectError
async def connect_unix_socket(
self, path, timeout=None, socket_options=None
async def sleep(self, seconds): # pylint: disable=signature-differs
await anyio.sleep(seconds)
- class _HTTPTransport(httpx.AsyncHTTPTransport):
+ class _HTTPTransport(httpx2.AsyncHTTPTransport):
def __init__(
self,
*args,
_requirements: dict[str, list[str]] = {
### BEGIN generated requirements
"dnssec": ["cryptography>=45"],
- "doh": ["httpcore>=1.0.0", "httpx>=0.28.0", "h2>=4.2.0"],
+ "doh": ["httpcore2>=2.4", "httpx2>=2.4", "h2>=4.3.0"],
"doq": ["aioquic>=1.2.0"],
"idna": ["idna>=3.10"],
"trio": ["trio>=0.30"],
if dns._features.have("doh"):
- import httpcore
- import httpcore._backends.trio
- import httpx
+ import httpcore2
+ import httpcore2._backends.trio
+ import httpx2
- _CoreAsyncNetworkBackend = httpcore.AsyncNetworkBackend
- _CoreTrioStream = httpcore._backends.trio.TrioStream
+ _CoreAsyncNetworkBackend = httpcore2.AsyncNetworkBackend
+ _CoreTrioStream = httpcore2._backends.trio.TrioStream
from dns.query import _compute_times, _expiration_for_this_attempt, _remaining
return _CoreTrioStream(sock.stream)
except Exception:
continue
- raise httpcore.ConnectError
+ raise httpcore2.ConnectError
async def connect_unix_socket(
self, path, timeout=None, socket_options=None
async def sleep(self, seconds): # pylint: disable=signature-differs
await trio.sleep(seconds)
- class _HTTPTransport(httpx.AsyncHTTPTransport):
+ class _HTTPTransport(httpx2.AsyncHTTPTransport):
def __init__(
self,
*args,
import dns._no_ssl as ssl # pyright: ignore
if have_doh:
- import httpx
+ import httpx2
# for brevity
_lltuple = dns.inet.low_level_address_tuple
source_port: int = 0, # pylint: disable=W0613
one_rr_per_rrset: bool = False,
ignore_trailing: bool = False,
- client: "httpx.AsyncClient|dns.quic.AsyncQuicConnection | None" = None,
+ client: "httpx2.AsyncClient|dns.quic.AsyncQuicConnection | None" = None,
path: str = "/dns-query",
post: bool = True,
verify: bool | str | ssl.SSLContext = True,
:param client: If provided, the client to use for the query. Unlike the
other dnspython async functions, a backend cannot be provided here
- because httpx always auto-detects the async backend.
- :type client: ``httpx.AsyncClient`` or ``None``
+ because httpx2 always auto-detects the async backend.
+ :type client: ``httpx2.AsyncClient`` or ``None``
See :py:func:`dns.query.https()` for the documentation of the other
parameters, exceptions, and return type of this method.
if not have_doh:
raise NoDOH # pragma: no cover
# pylint: disable=possibly-used-before-assignment
- if client and not isinstance(client, httpx.AsyncClient): # pyright: ignore
- raise ValueError("client parameter must be an httpx.AsyncClient")
+ if client and not isinstance(client, httpx2.AsyncClient): # pyright: ignore
+ raise ValueError("client parameter must be an httpx2.AsyncClient")
# pylint: enable=possibly-used-before-assignment
wire = q.to_wire()
family=family,
)
- cm = httpx.AsyncClient( # pyright: ignore
+ cm = httpx2.AsyncClient( # pyright: ignore
http1=h1, http2=h2, verify=verify, transport=transport # type: ignore
)
)
else:
wire = base64.urlsafe_b64encode(wire).rstrip(b"=")
- twire = wire.decode() # httpx does a repr() if we give it bytes
+ twire = wire.decode() # httpx2 does a repr() if we give it bytes
response = await backend.wait_for(
the_client.get( # pyright: ignore
url,
return min(time.time() + timeout, expiration)
-_have_httpx = dns._features.have("doh")
-if _have_httpx:
- import httpcore._backends.sync
- import httpx
+_have_httpx2 = dns._features.have("doh")
+if _have_httpx2:
+ import httpcore2._backends.sync
+ import httpx2
- _CoreNetworkBackend = httpcore.NetworkBackend
- _CoreSyncStream = httpcore._backends.sync.SyncStream
+ _CoreNetworkBackend = httpcore2.NetworkBackend
+ _CoreSyncStream = httpcore2._backends.sync.SyncStream
class _NetworkBackend(_CoreNetworkBackend):
def __init__(self, resolver, local_port, bootstrap_address, family):
return _CoreSyncStream(sock)
except Exception:
pass
- raise httpcore.ConnectError
+ raise httpcore2.ConnectError
def connect_unix_socket(
self, path, timeout=None, socket_options=None
): # pylint: disable=signature-differs
raise NotImplementedError
- class _HTTPTransport(httpx.HTTPTransport): # pyright: ignore
+ class _HTTPTransport(httpx2.HTTPTransport): # pyright: ignore
def __init__(
self,
*args,
raise NotImplementedError
-have_doh = _have_httpx
+have_doh = _have_httpx2
def default_socket_factory(
class NoDOH(dns.exception.DNSException):
- """DNS over HTTPS (DOH) was requested but the httpx module is not
+ """DNS over HTTPS (DOH) was requested but the httpx2 module is not
available."""
:type ignore_trailing: bool
:param session: If provided, the client session to use to send the
queries.
- :type session: ``httpx.Client`` or ``None``
+ :type session: ``httpx2.Client`` or ``None``
:param path: If *where* is an IP address, *path* is used to construct the
query URL.
:type path: str
:param resolver: Resolver to use for hostname resolution in URLs. If
``None``, a new resolver with default configuration is used (not the
default resolver, to avoid a DoH chicken-and-egg problem). Only
- effective when using httpx.
+ effective when using httpx2.
:type resolver: :py:class:`dns.resolver.Resolver` or ``None``
:param family: Address family. ``socket.AF_UNSPEC`` (the default)
retrieves both A and AAAA records.
if not have_doh:
raise NoDOH # pragma: no cover
- if session and not isinstance(session, httpx.Client): # pyright: ignore
- raise ValueError("session parameter must be an httpx.Client")
+ if session and not isinstance(session, httpx2.Client): # pyright: ignore
+ raise ValueError("session parameter must be an httpx2.Client")
wire = q.to_wire()
headers = {"accept": "application/dns-message"}
family=family, # pyright: ignore
)
- cm = httpx.Client( # pyright: ignore
+ cm = httpx2.Client( # pyright: ignore
http1=h1, http2=h2, verify=verify, transport=transport # type: ignore
)
with cm as session:
)
else:
wire = base64.urlsafe_b64encode(wire).rstrip(b"=")
- twire = wire.decode() # httpx does a repr() if we give it bytes
+ twire = wire.decode() # httpx2 does a repr() if we give it bytes
response = session.get(
url,
headers=headers,
intersphinx_mapping = {
"python": ("https://docs.python.org/3", None),
- "cryptography": ("https://cryptography.io/en/latest/", None)
+ "cryptography": ("https://cryptography.io/en/latest/", None),
}
nitpick_ignore = [
# socket module class (from dns.query sock parameters)
("py:class", "socket"),
# External libraries not in intersphinx
- ("py:class", "httpx.AsyncClient"),
+ ("py:class", "httpx2.AsyncClient"),
("py:class", "dns.quic.AsyncQuicConnection"),
]
The following modules are optional, but recommended for full functionality.
-If ``httpx`` is installed, then DNS-over-HTTPS will be available.
+If ``httpx2`` is installed, then DNS-over-HTTPS will be available.
If ``cryptography`` is installed, then dnspython will be
able to do low-level DNSSEC signature generation and validation.
currently dns.message.CopyMode.QUESTION for all opcodes.
* If an IP address is used as the hostname in a URL, the https query code now passes
- the sni_hostname to httpx as this is required to get httpx to validate the certificate
+ the sni_hostname to httpx2 as this is required to get httpx2 to validate the certificate
and check for an IP subject alternative name.
* The minimum supported aioquic version is now 1.0.0.
* Dnspython now looks for version metadata for optional packages and will not
use them if they are too old. This prevents possible exceptions when a
- feature like DoH is not desired in dnspython, but an old httpx is installed
+ feature like DoH is not desired in dnspython, but an old httpx2 is installed
along with dnspython for some other purpose.
* The DoHNameserver class now allows GET to be used instead of the default POST,
* The DNS-over-HTTPS bootstrap address no longer causes URL rewriting.
-* DNS-over-HTTPS now only uses httpx; support for requests has been dropped. A source
- port may now be supplied when using httpx.
+* DNS-over-HTTPS now only uses httpx2; support for requests has been dropped. A source
+ port may now be supplied when using httpx2.
* DNSSEC zone signing with NSEC records is now supported. Thank you
very much (again!) Jakob Schlyter!
an error trace like the NoNameservers exception. This class is a subclass of
dns.exception.Timeout for backwards compatibility.
-* DNS-over-HTTPS will try to use HTTP/2 if the httpx and h2 packages
+* DNS-over-HTTPS will try to use HTTP/2 if the httpx2 and h2 packages
are installed.
* DNS-over-HTTPS is now supported for asynchronous queries and resolutions.
import copy
import json
-import httpx
+import httpx2
import dns.flags
import dns.message
a = dns.resolver.resolve("www.dnspython.org", "a")
p = to_doh_simple(a.response)
print(json.dumps(p, indent=4))
-response = httpx.get(
+response = httpx2.get(
"https://dns.google/resolve?",
verify=True,
params={"name": "www.dnspython.org", "type": 1},
#!/usr/bin/env python3
#
# This is an example of sending DNS queries over HTTPS (DoH) with dnspython.
-import httpx
+import httpx2
import dns.message
import dns.query
def main():
where = "https://dns.google/dns-query"
qname = "example.com."
- with httpx.Client() as client:
+ with httpx2.Client() as client:
q = dns.message.make_query(qname, dns.rdatatype.A)
r = dns.query.https(q, where, session=client)
for answer in r.answer:
"ty>=0.0.14",
]
dnssec = ["cryptography>=46"]
-doh = ["httpcore>=1.0.0", "httpx>=0.28.0", "h2>=4.3.0"]
+doh = ["httpcore22>=2.4", "httpx2>=2.4", "h2>=4.3.0"]
doq = ["aioquic>=1.3.0"]
idna = ["idna>=3.11"]
trio = ["trio>=0.30"]
self.assertRaises(dns.exception.Timeout, run)
- @unittest.skipIf(not dns.query._have_httpx, "httpx not available")
+ @unittest.skipIf(not dns.query._have_httpx2, "httpx2 not available")
@tests.util.retry_on_timeout
def testDOHGetRequest(self):
async def run():
self.async_run(run)
- @unittest.skipIf(not dns.query._have_httpx, "httpx not available")
+ @unittest.skipIf(not dns.query._have_httpx2, "httpx2 not available")
@tests.util.retry_on_timeout
def testDOHPostRequest(self):
async def run():
self.async_run(run)
- @unittest.skipIf(not dns.query._have_httpx, "httpx not available")
+ @unittest.skipIf(not dns.query._have_httpx2, "httpx2 not available")
@tests.util.retry_on_timeout
def testResolverDOH(self):
async def run():
import dns.rdatatype
import dns.resolver
-if dns.query._have_httpx:
- import httpx
+if dns.query._have_httpx2:
+ import httpx2
import tests.util
@unittest.skipUnless(
- dns.query._have_httpx and tests.util.is_internet_reachable() and _have_ssl,
- "Python httpx cannot be imported; no DNS over HTTPS (DOH)",
+ dns.query._have_httpx2 and tests.util.is_internet_reachable() and _have_ssl,
+ "Python httpx2 cannot be imported; no DNS over HTTPS (DOH)",
)
-class DNSOverHTTPSTestCaseHttpx(unittest.TestCase):
+class DNSOverHTTPSTestCasehttpx2(unittest.TestCase):
def setUp(self):
- self.session = httpx.Client(http1=True, http2=True, verify=True)
+ self.session = httpx2.Client(http1=True, http2=True, verify=True)
def tearDown(self):
self.session.close()
# q = dns.message.make_query("example.com.", dns.rdatatype.A)
# # make sure CleanBrowsing's IP address will fail TLS certificate
# # check.
- # with self.assertRaises(httpx.ConnectError):
+ # with self.assertRaises(httpx2.ConnectError):
# dns.query.https(q, invalid_tls_url, session=self.session, timeout=4)
# # And if we don't mangle the URL, it should work.
# r = dns.query.https(
projects / thirdparty / dnspython.git / commitdiff
