URLTypes,
)
from .status_codes import StatusCode, codes
+
+__all__ = [
+ "__description__",
+ "__title__",
+ "__version__",
+ "delete",
+ "get",
+ "head",
+ "options",
+ "patch",
+ "post",
+ "patch",
+ "put",
+ "request",
+ "AsyncClient",
+ "Client",
+ "AsyncioBackend",
+ "USER_AGENT",
+ "CertTypes",
+ "PoolLimits",
+ "SSLConfig",
+ "TimeoutConfig",
+ "VerifyTypes",
+ "HTTPConnection",
+ "ConnectionPool",
+ "ConnectTimeout",
+ "CookieConflict",
+ "DecodingError",
+ "InvalidURL",
+ "PoolTimeout",
+ "ProtocolError",
+ "ReadTimeout",
+ "RedirectBodyUnavailable",
+ "RedirectLoop",
+ "ResponseClosed",
+ "ResponseNotRead",
+ "StreamConsumed",
+ "Timeout",
+ "TooManyRedirects",
+ "WriteTimeout",
+ "AsyncDispatcher",
+ "BaseReader",
+ "BaseWriter",
+ "ConcurrencyBackend",
+ "Dispatcher",
+ "Protocol",
+ "URL",
+ "URLTypes",
+ "StatusCode",
+ "codes",
+ "TimeoutTypes",
+ "AsyncRequest",
+ "AsyncRequestData",
+ "AsyncResponse",
+ "AsyncResponseContent",
+ "AuthTypes",
+ "Cookies",
+ "CookieTypes",
+ "Headers",
+ "HeaderTypes",
+ "Origin",
+ "QueryParams",
+ "QueryParamTypes",
+ "Request",
+ "RequestData",
+ "Response",
+ "ResponseContent",
+ "RequestFiles",
+]
Headers,
HeaderTypes,
QueryParamTypes,
- Request,
RequestData,
RequestFiles,
Response,
if response.is_redirect:
async def send_next() -> AsyncResponse:
- nonlocal request, response, verify, cert, allow_redirects, timeout, history
+ nonlocal request, response, verify, cert
+ nonlocal allow_redirects, timeout, history
request = self.build_redirect_request(request, response)
response = await self.send_handling_redirects(
request,
import typing
from types import TracebackType
-from .config import DEFAULT_TIMEOUT_CONFIG, PoolLimits, TimeoutConfig
+from .config import PoolLimits, TimeoutConfig
from .exceptions import ConnectTimeout, PoolTimeout, ReadTimeout, WriteTimeout
from .interfaces import (
BaseBackgroundManager,
None, self.load_ssl_context_verify
)
+ assert self.ssl_context is not None
return self.ssl_context
def load_ssl_context_no_verify(self) -> ssl.SSLContext:
# Signal to server support for PHA in TLS 1.3. Raises an
# AttributeError if only read-only access is implemented.
try:
- context.post_handshake_auth = True
- except AttributeError:
+ context.post_handshake_auth = True # type: ignore
+ except AttributeError: # pragma: nocover
pass
# Disable using 'commonName' for SSLContext.check_hostname
# when the 'subjectAltName' extension isn't available.
try:
- context.hostname_checks_common_name = False
- except AttributeError:
+ context.hostname_checks_common_name = False # type: ignore
+ except AttributeError: # pragma: nocover
pass
if os.path.isfile(ca_bundle_path):
context.load_cert_chain(certfile=self.cert)
elif isinstance(self.cert, tuple) and len(self.cert) == 2:
context.load_cert_chain(certfile=self.cert[0], keyfile=self.cert[1])
- else:
+ elif isinstance(self.cert, tuple) and len(self.cert) == 3:
context.load_cert_chain(
- certfile=self.cert[0], keyfile=self.cert[1], password=self.cert[2]
+ certfile=self.cert[0],
+ keyfile=self.cert[1],
+ password=self.cert[2], # type: ignore
)
return context
class_name = self.__class__.__name__
if len({self.connect_timeout, self.read_timeout, self.write_timeout}) == 1:
return f"{class_name}(timeout={self.connect_timeout})"
- return f"{class_name}(connect_timeout={self.connect_timeout}, read_timeout={self.read_timeout}, write_timeout={self.write_timeout})"
+ return (
+ f"{class_name}(connect_timeout={self.connect_timeout}, "
+ f"read_timeout={self.read_timeout}, write_timeout={self.write_timeout})"
+ )
class PoolLimits:
def __repr__(self) -> str:
class_name = self.__class__.__name__
- return f"{class_name}(soft_limit={self.soft_limit}, hard_limit={self.hard_limit}, pool_timeout={self.pool_timeout})"
+ return (
+ f"{class_name}(soft_limit={self.soft_limit}, "
+ f"hard_limit={self.hard_limit}, pool_timeout={self.pool_timeout})"
+ )
DEFAULT_SSL_CONFIG = SSLConfig(cert=None, verify=True)
import functools
import typing
-import h2.connection
-import h11
-
from ..concurrency import AsyncioBackend
from ..config import (
- DEFAULT_SSL_CONFIG,
DEFAULT_TIMEOUT_CONFIG,
CertTypes,
SSLConfig,
TimeoutTypes,
VerifyTypes,
)
-from ..exceptions import ConnectTimeout
from ..interfaces import AsyncDispatcher, ConcurrencyBackend, Protocol
from ..models import AsyncRequest, AsyncResponse, Origin
from .http2 import HTTP2Connection
from ..concurrency import AsyncioBackend
from ..config import (
- DEFAULT_CA_BUNDLE_PATH,
DEFAULT_POOL_LIMITS,
DEFAULT_TIMEOUT_CONFIG,
CertTypes,
TimeoutTypes,
VerifyTypes,
)
-from ..decoders import ACCEPT_ENCODING
-from ..exceptions import NotConnected, PoolTimeout
+from ..exceptions import NotConnected
from ..interfaces import AsyncDispatcher, ConcurrencyBackend
from ..models import AsyncRequest, AsyncResponse, Origin
from .connection import HTTPConnection
import h11
from ..concurrency import TimeoutFlag
-from ..config import DEFAULT_TIMEOUT_CONFIG, TimeoutConfig, TimeoutTypes
-from ..exceptions import ConnectTimeout, NotConnected, ReadTimeout
+from ..config import TimeoutConfig, TimeoutTypes
+from ..exceptions import NotConnected
from ..interfaces import BaseReader, BaseWriter, ConcurrencyBackend
from ..models import AsyncRequest, AsyncResponse
event = h11.ConnectionClosed()
try:
self.h11_state.send(event)
- except h11.LocalProtocolError as exc: # pragma: no cover
+ except h11.LocalProtocolError: # pragma: no cover
# Premature client disconnect
pass
await self.writer.close()
import h2.events
from ..concurrency import TimeoutFlag
-from ..config import DEFAULT_TIMEOUT_CONFIG, TimeoutConfig, TimeoutTypes
-from ..exceptions import ConnectTimeout, NotConnected, ReadTimeout
+from ..config import TimeoutConfig, TimeoutTypes
+from ..exceptions import NotConnected
from ..interfaces import BaseReader, BaseWriter, ConcurrencyBackend
from ..models import AsyncRequest, AsyncResponse
from .config import CertTypes, PoolLimits, TimeoutConfig, TimeoutTypes, VerifyTypes
from .models import (
- URL,
AsyncRequest,
AsyncRequestData,
AsyncResponse,
- Headers,
HeaderTypes,
QueryParamTypes,
Request,
@classmethod
def is_redirect(cls, value: int) -> bool:
return value in (
- StatusCode.MOVED_PERMANENTLY, # 301 (Cacheable redirect. Method may change to GET.)
- StatusCode.FOUND, # 302 (Uncacheable redirect. Method may change to GET.)
- StatusCode.SEE_OTHER, # 303 (Client should make a GET or HEAD request.)
- StatusCode.TEMPORARY_REDIRECT, # 307 (Equiv. 302, but retain method)
- StatusCode.PERMANENT_REDIRECT, # 308 (Equiv. 301, but retain method)
+ # 301 (Cacheable redirect. Method may change to GET.)
+ StatusCode.MOVED_PERMANENTLY,
+ # 302 (Uncacheable redirect. Method may change to GET.)
+ StatusCode.FOUND,
+ # 303 (Client should make a GET or HEAD request.)
+ StatusCode.SEE_OTHER,
+ # 307 (Equiv. 302, but retain method)
+ StatusCode.TEMPORARY_REDIRECT,
+ # 308 (Equiv. 301, but retain method)
+ StatusCode.PERMANENT_REDIRECT,
)
@classmethod
autoflake
black
cryptography
+flake8
isort
mypy
pytest
set -x
-${PREFIX}autoflake --in-place --recursive http3 tests
-${PREFIX}black http3 tests
-${PREFIX}isort --multi-line=3 --trailing-comma --force-grid-wrap=0 --combine-as --line-width 88 --recursive --apply http3 tests
+${PREFIX}autoflake --in-place --recursive http3 tests setup.py
+${PREFIX}isort --multi-line=3 --trailing-comma --force-grid-wrap=0 --combine-as --line-width 88 --recursive --apply http3 tests setup.py
+${PREFIX}black http3 tests setup.py
+${PREFIX}flake8 --max-line-length=88 --ignore=W503,E203 http3 tests setup.py
${PREFIX}mypy http3 --ignore-missing-imports --disallow-untyped-defs
scripts/clean
"h11==0.8.*",
"h2==3.*",
"idna==2.*",
- "rfc3986==1.*"
+ "rfc3986==1.*",
],
classifiers=[
"Development Status :: 3 - Alpha",
import json
-import pytest
-
from http3 import (
- URL,
AsyncDispatcher,
AsyncRequest,
AsyncResponse,
import json
from http.cookiejar import Cookie, CookieJar
-import pytest
-
from http3 import (
- URL,
AsyncDispatcher,
AsyncRequest,
AsyncResponse,
CertTypes,
RedirectBodyUnavailable,
RedirectLoop,
- Request,
- Response,
TimeoutTypes,
TooManyRedirects,
VerifyTypes,
import pytest
-from http3 import HTTPConnection, Request, SSLConfig
+from http3 import HTTPConnection
@pytest.mark.asyncio
import json
-import pytest
-
from http3 import Client, Response
from .utils import MockHTTP2Backend
import json
-import pytest
-
from http3 import (
CertTypes,
Client,
def test_asgi_exc():
client = http3.Client(app=raise_exc)
with pytest.raises(ValueError):
- response = client.get("http://www.example.org/")
+ client.get("http://www.example.org/")
def test_asgi_exc_after_response():
client = http3.Client(app=raise_exc_after_response)
with pytest.raises(ValueError):
- response = client.get("http://www.example.org/")
+ client.get("http://www.example.org/")
import cgi
import io
-import pytest
-
from http3 import (
CertTypes,
Client,
pdict = {"boundary": boundary.encode("ascii"), "CONTENT-LENGTH": content_length}
multipart = cgi.parse_multipart(io.BytesIO(response.content), pdict)
- # Note that the expected return type for text fields appears to differs from 3.6 to 3.7+
+ # Note that the expected return type for text fields
+ # appears to differs from 3.6 to 3.7+
assert multipart["text"] == ["abc"] or multipart["text"] == [b"abc"]
assert multipart["file"] == [b"<file content>"]
pdict = {"boundary": boundary.encode("ascii"), "CONTENT-LENGTH": content_length}
multipart = cgi.parse_multipart(io.BytesIO(response.content), pdict)
- # Note that the expected return type for text fields appears to differs from 3.6 to 3.7+
+ # Note that the expected return type for text fields
+ # appears to differs from 3.6 to 3.7+
assert multipart["text"] == ["abc"] or multipart["text"] == [b"abc"]
assert multipart["file"] == [b"<file content>"]
try:
raise ValueError()
- except:
+ except ValueError:
exc_info = sys.exc_info()
start_response(status, response_headers, exc_info=exc_info)
def test_wsgi_exc():
client = http3.Client(app=raise_exc)
with pytest.raises(ValueError):
- response = client.get("http://www.example.org/")
+ client.get("http://www.example.org/")
projects / thirdparty / httpx.git / commitdiff
