Skip to content
Docs Pydantic Docs

Developer Interface

Helper Functions

request 

def request(
    method: str,
    url: URL | str,
    *,
    params: QueryParamTypes | None = None,
    content: RequestContent | None = None,
    data: RequestData | None = None,
    files: RequestFiles | None = None,
    json: typing.Any | None = None,
    headers: HeaderTypes | None = None,
    cookies: CookieTypes | None = None,
    auth: AuthTypes | None = None,
    proxy: ProxyTypes | None = None,
    timeout: TimeoutTypes = DEFAULT_TIMEOUT_CONFIG,
    follow_redirects: bool = False,
    verify: ssl.SSLContext | str | bool = True,
    trust_env: bool = True,
) -> Response

Sends an HTTP request.

Usage:

>>> import httpx2
>>> response = httpx2.request('GET', 'https://httpbin.org/get')
>>> response
<Response [200 OK]>

Returns

Response — The Response object.

Parameters

method : str

HTTP method for the new Request object: GET, OPTIONS, HEAD, POST, PUT, PATCH, DELETE, or QUERY.

url : URL | str

URL for the new Request object.

params : QueryParamTypes | None Default: None

(optional) Query parameters to include in the URL, as a string, dictionary, or sequence of two-tuples.

content : RequestContent | None Default: None

(optional) Binary content to include in the body of the request, as bytes or a byte iterator.

data : RequestData | None Default: None

(optional) Form data to include in the body of the request, as a dictionary.

files : RequestFiles | None Default: None

(optional) A dictionary of upload files to include in the body of the request.

json : typing.Any | None Default: None

(optional) A JSON serializable object to include in the body of the request.

headers : HeaderTypes | None Default: None

(optional) Dictionary of HTTP headers to include in the request.

cookies : CookieTypes | None Default: None

(optional) Dictionary of Cookie items to include in the request.

auth : AuthTypes | None Default: None

(optional) An authentication class to use when sending the request.

proxy : ProxyTypes | None Default: None

(optional) A proxy URL where all the traffic should be routed.

timeout : TimeoutTypes Default: DEFAULT_TIMEOUT_CONFIG

(optional) The timeout configuration to use when sending the request.

follow_redirects : bool Default: False

(optional) Enables or disables HTTP redirects.

verify : ssl.SSLContext | str | bool Default: True

(optional) Either True to use an SSL context with the default CA bundle, False to disable verification, or an instance of ssl.SSLContext to use a custom context.

trust_env : bool Default: True

(optional) Enables or disables usage of environment variables for configuration.

get 

def get(
    url: URL | str,
    *,
    params: QueryParamTypes | None = None,
    headers: HeaderTypes | None = None,
    cookies: CookieTypes | None = None,
    auth: AuthTypes | None = None,
    proxy: ProxyTypes | None = None,
    follow_redirects: bool = False,
    verify: ssl.SSLContext | str | bool = True,
    timeout: TimeoutTypes = DEFAULT_TIMEOUT_CONFIG,
    trust_env: bool = True,
) -> Response

Sends a GET request.

Parameters: See httpx2.request.

Note that the data, files, json and content parameters are not available on this function, as GET requests should not include a request body.

Returns

Response

options 

def options(
    url: URL | str,
    *,
    params: QueryParamTypes | None = None,
    headers: HeaderTypes | None = None,
    cookies: CookieTypes | None = None,
    auth: AuthTypes | None = None,
    proxy: ProxyTypes | None = None,
    follow_redirects: bool = False,
    verify: ssl.SSLContext | str | bool = True,
    timeout: TimeoutTypes = DEFAULT_TIMEOUT_CONFIG,
    trust_env: bool = True,
) -> Response

Sends an OPTIONS request.

Parameters: See httpx2.request.

Note that the data, files, json and content parameters are not available on this function, as OPTIONS requests should not include a request body.

Returns

Response

head 

def head(
    url: URL | str,
    *,
    params: QueryParamTypes | None = None,
    headers: HeaderTypes | None = None,
    cookies: CookieTypes | None = None,
    auth: AuthTypes | None = None,
    proxy: ProxyTypes | None = None,
    follow_redirects: bool = False,
    verify: ssl.SSLContext | str | bool = True,
    timeout: TimeoutTypes = DEFAULT_TIMEOUT_CONFIG,
    trust_env: bool = True,
) -> Response

Sends a HEAD request.

Parameters: See httpx2.request.

Note that the data, files, json and content parameters are not available on this function, as HEAD requests should not include a request body.

Returns

Response

post 

def post(
    url: URL | str,
    *,
    content: RequestContent | None = None,
    data: RequestData | None = None,
    files: RequestFiles | None = None,
    json: typing.Any | None = None,
    params: QueryParamTypes | None = None,
    headers: HeaderTypes | None = None,
    cookies: CookieTypes | None = None,
    auth: AuthTypes | None = None,
    proxy: ProxyTypes | None = None,
    follow_redirects: bool = False,
    verify: ssl.SSLContext | str | bool = True,
    timeout: TimeoutTypes = DEFAULT_TIMEOUT_CONFIG,
    trust_env: bool = True,
) -> Response

Sends a POST request.

Parameters: See httpx2.request.

Returns

Response

put 

def put(
    url: URL | str,
    *,
    content: RequestContent | None = None,
    data: RequestData | None = None,
    files: RequestFiles | None = None,
    json: typing.Any | None = None,
    params: QueryParamTypes | None = None,
    headers: HeaderTypes | None = None,
    cookies: CookieTypes | None = None,
    auth: AuthTypes | None = None,
    proxy: ProxyTypes | None = None,
    follow_redirects: bool = False,
    verify: ssl.SSLContext | str | bool = True,
    timeout: TimeoutTypes = DEFAULT_TIMEOUT_CONFIG,
    trust_env: bool = True,
) -> Response

Sends a PUT request.

Parameters: See httpx2.request.

Returns

Response

patch 

def patch(
    url: URL | str,
    *,
    content: RequestContent | None = None,
    data: RequestData | None = None,
    files: RequestFiles | None = None,
    json: typing.Any | None = None,
    params: QueryParamTypes | None = None,
    headers: HeaderTypes | None = None,
    cookies: CookieTypes | None = None,
    auth: AuthTypes | None = None,
    proxy: ProxyTypes | None = None,
    follow_redirects: bool = False,
    verify: ssl.SSLContext | str | bool = True,
    timeout: TimeoutTypes = DEFAULT_TIMEOUT_CONFIG,
    trust_env: bool = True,
) -> Response

Sends a PATCH request.

Parameters: See httpx2.request.

Returns

Response

delete 

def delete(
    url: URL | str,
    *,
    params: QueryParamTypes | None = None,
    headers: HeaderTypes | None = None,
    cookies: CookieTypes | None = None,
    auth: AuthTypes | None = None,
    proxy: ProxyTypes | None = None,
    follow_redirects: bool = False,
    timeout: TimeoutTypes = DEFAULT_TIMEOUT_CONFIG,
    verify: ssl.SSLContext | str | bool = True,
    trust_env: bool = True,
) -> Response

Sends a DELETE request.

Parameters: See httpx2.request.

Note that the data, files, json and content parameters are not available on this function, as DELETE requests should not include a request body.

Returns

Response

query 

def query(
    url: URL | str,
    *,
    content: RequestContent | None = None,
    data: RequestData | None = None,
    files: RequestFiles | None = None,
    json: typing.Any | None = None,
    params: QueryParamTypes | None = None,
    headers: HeaderTypes | None = None,
    cookies: CookieTypes | None = None,
    auth: AuthTypes | None = None,
    proxy: ProxyTypes | None = None,
    follow_redirects: bool = False,
    verify: ssl.SSLContext | str | bool = True,
    timeout: TimeoutTypes = DEFAULT_TIMEOUT_CONFIG,
    trust_env: bool = True,
) -> Response

Sends a QUERY request.

Parameters: See httpx2.request.

Returns

Response

stream 

def stream(
    method: str,
    url: URL | str,
    *,
    params: QueryParamTypes | None = None,
    content: RequestContent | None = None,
    data: RequestData | None = None,
    files: RequestFiles | None = None,
    json: typing.Any | None = None,
    headers: HeaderTypes | None = None,
    cookies: CookieTypes | None = None,
    auth: AuthTypes | None = None,
    proxy: ProxyTypes | None = None,
    timeout: TimeoutTypes = DEFAULT_TIMEOUT_CONFIG,
    follow_redirects: bool = False,
    verify: ssl.SSLContext | str | bool = True,
    trust_env: bool = True,
) -> Generator[Response]

Alternative to httpx2.request() that streams the response body instead of loading it into memory at once.

Parameters: See httpx2.request.

See also: Streaming Responses

Returns

Generator[Response]

websocket 

def websocket(
    url: URL | str,
    *,
    params: QueryParamTypes | None = None,
    headers: HeaderTypes | None = None,
    cookies: CookieTypes | None = None,
    auth: AuthTypes | None = None,
    proxy: ProxyTypes | None = None,
    follow_redirects: bool = False,
    timeout: TimeoutTypes = DEFAULT_TIMEOUT_CONFIG,
    verify: ssl.SSLContext | str | bool = True,
    trust_env: bool = True,
    subprotocols: list[str] | None = None,
    max_message_size_bytes: int = DEFAULT_MAX_MESSAGE_SIZE_BYTES,
    queue_size: int = DEFAULT_QUEUE_SIZE,
    keepalive_ping_interval_seconds: float | None = DEFAULT_KEEPALIVE_PING_INTERVAL_SECONDS,
    keepalive_ping_timeout_seconds: float | None = DEFAULT_KEEPALIVE_PING_TIMEOUT_SECONDS,
) -> Generator[WebSocketSession]

Open a WebSocket session.

The session is closed automatically when exiting the context manager.

with httpx2.websocket("ws://localhost:8000/ws") as ws:
    ws.send_text("Hello!")
    message = ws.receive_text()

Parameters: See httpx2.request and httpx2.Client.websocket.

Returns

Generator[WebSocketSession]

Client

Client

Bases: BaseClient

An HTTP client, with connection pooling, HTTP/2, redirects, cookie persistence, etc.

It can be shared between threads.

Usage:

>>> client = httpx2.Client()
>>> response = client.get('https://example.org')

Parameters:

  • auth - (optional) An authentication class to use when sending requests.
  • params - (optional) Query parameters to include in request URLs, as a string, dictionary, or sequence of two-tuples.
  • headers - (optional) Dictionary of HTTP headers to include when sending requests.
  • cookies - (optional) Dictionary of Cookie items to include when sending requests.
  • verify - (optional) Either True to use an SSL context with the default CA bundle, False to disable verification, or an instance of ssl.SSLContext to use a custom context.
  • http2 - (optional) A boolean indicating if HTTP/2 support should be enabled. Defaults to False.
  • proxy - (optional) A proxy URL where all the traffic should be routed.
  • mounts - (optional) A dictionary mapping URL patterns to transports, used to route requests through specific transports based on the URL.
  • timeout - (optional) The timeout configuration to use when sending requests.
  • limits - (optional) The limits configuration to use.
  • max_redirects - (optional) The maximum number of redirect responses that should be followed.
  • base_url - (optional) A URL to use as the base when building request URLs.
  • transport - (optional) A transport class to use for sending requests over the network.
  • trust_env - (optional) Enables or disables usage of environment variables for configuration.
  • default_encoding - (optional) The default encoding to use for decoding response text, if no charset information is included in a response Content-Type header. Set to a callable for automatic character set detection. Default: “utf-8”.

Attributes

headers

HTTP headers to include when sending requests.

Type: Headers

cookies

Cookie values to include when sending requests.

Type: Cookies

params

Query parameters to include in the URL when sending requests.

Type: QueryParams

auth

Authentication class used when none is passed at the request-level.

See also Authentication.

Type: Auth | None

Methods

request 
def request(
    method: str,
    url: URL | str,
    *,
    content: RequestContent | None = None,
    data: RequestData | None = None,
    files: RequestFiles | None = None,
    json: typing.Any | None = None,
    params: QueryParamTypes | None = None,
    headers: HeaderTypes | None = None,
    cookies: CookieTypes | None = None,
    auth: AuthTypes | UseClientDefault | None = USE_CLIENT_DEFAULT,
    follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
    timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
    extensions: RequestExtensions | None = None,
) -> Response

Build and send a request.

Equivalent to:

request = client.build_request(...)
response = client.send(request, ...)

See Client.build_request(), Client.send() and Merging of configuration for how the various parameters are merged with client-level configuration.

Returns

Response

get 
def get(
    url: URL | str,
    *,
    params: QueryParamTypes | None = None,
    headers: HeaderTypes | None = None,
    cookies: CookieTypes | None = None,
    auth: AuthTypes | UseClientDefault | None = USE_CLIENT_DEFAULT,
    follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
    timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
    extensions: RequestExtensions | None = None,
) -> Response

Send a GET request.

Parameters: See httpx2.request.

Returns

Response

head 
def head(
    url: URL | str,
    *,
    params: QueryParamTypes | None = None,
    headers: HeaderTypes | None = None,
    cookies: CookieTypes | None = None,
    auth: AuthTypes | UseClientDefault = USE_CLIENT_DEFAULT,
    follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
    timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
    extensions: RequestExtensions | None = None,
) -> Response

Send a HEAD request.

Parameters: See httpx2.request.

Returns

Response

options 
def options(
    url: URL | str,
    *,
    params: QueryParamTypes | None = None,
    headers: HeaderTypes | None = None,
    cookies: CookieTypes | None = None,
    auth: AuthTypes | UseClientDefault = USE_CLIENT_DEFAULT,
    follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
    timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
    extensions: RequestExtensions | None = None,
) -> Response

Send an OPTIONS request.

Parameters: See httpx2.request.

Returns

Response

post 
def post(
    url: URL | str,
    *,
    content: RequestContent | None = None,
    data: RequestData | None = None,
    files: RequestFiles | None = None,
    json: typing.Any | None = None,
    params: QueryParamTypes | None = None,
    headers: HeaderTypes | None = None,
    cookies: CookieTypes | None = None,
    auth: AuthTypes | UseClientDefault = USE_CLIENT_DEFAULT,
    follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
    timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
    extensions: RequestExtensions | None = None,
) -> Response

Send a POST request.

Parameters: See httpx2.request.

Returns

Response

put 
def put(
    url: URL | str,
    *,
    content: RequestContent | None = None,
    data: RequestData | None = None,
    files: RequestFiles | None = None,
    json: typing.Any | None = None,
    params: QueryParamTypes | None = None,
    headers: HeaderTypes | None = None,
    cookies: CookieTypes | None = None,
    auth: AuthTypes | UseClientDefault = USE_CLIENT_DEFAULT,
    follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
    timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
    extensions: RequestExtensions | None = None,
) -> Response

Send a PUT request.

Parameters: See httpx2.request.

Returns

Response

patch 
def patch(
    url: URL | str,
    *,
    content: RequestContent | None = None,
    data: RequestData | None = None,
    files: RequestFiles | None = None,
    json: typing.Any | None = None,
    params: QueryParamTypes | None = None,
    headers: HeaderTypes | None = None,
    cookies: CookieTypes | None = None,
    auth: AuthTypes | UseClientDefault = USE_CLIENT_DEFAULT,
    follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
    timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
    extensions: RequestExtensions | None = None,
) -> Response

Send a PATCH request.

Parameters: See httpx2.request.

Returns

Response

delete 
def delete(
    url: URL | str,
    *,
    params: QueryParamTypes | None = None,
    headers: HeaderTypes | None = None,
    cookies: CookieTypes | None = None,
    auth: AuthTypes | UseClientDefault = USE_CLIENT_DEFAULT,
    follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
    timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
    extensions: RequestExtensions | None = None,
) -> Response

Send a DELETE request.

Parameters: See httpx2.request.

Returns

Response

query 
def query(
    url: URL | str,
    *,
    content: RequestContent | None = None,
    data: RequestData | None = None,
    files: RequestFiles | None = None,
    json: typing.Any | None = None,
    params: QueryParamTypes | None = None,
    headers: HeaderTypes | None = None,
    cookies: CookieTypes | None = None,
    auth: AuthTypes | UseClientDefault = USE_CLIENT_DEFAULT,
    follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
    timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
    extensions: RequestExtensions | None = None,
) -> Response

Send a QUERY request.

Parameters: See httpx2.request.

Returns

Response

stream 
def stream(
    method: str,
    url: URL | str,
    *,
    content: RequestContent | None = None,
    data: RequestData | None = None,
    files: RequestFiles | None = None,
    json: typing.Any | None = None,
    params: QueryParamTypes | None = None,
    headers: HeaderTypes | None = None,
    cookies: CookieTypes | None = None,
    auth: AuthTypes | UseClientDefault | None = USE_CLIENT_DEFAULT,
    follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
    timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
    extensions: RequestExtensions | None = None,
) -> Generator[Response]

Alternative to httpx2.request() that streams the response body instead of loading it into memory at once.

Parameters: See httpx2.request.

See also: Streaming Responses

Returns

Generator[Response]

sse 
def sse(
    url: URL | str,
    *,
    method: str = 'GET',
    content: RequestContent | None = None,
    data: RequestData | None = None,
    files: RequestFiles | None = None,
    json: typing.Any | None = None,
    params: QueryParamTypes | None = None,
    headers: HeaderTypes | None = None,
    cookies: CookieTypes | None = None,
    auth: AuthTypes | UseClientDefault | None = USE_CLIENT_DEFAULT,
    follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
    timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
    extensions: RequestExtensions | None = None,
    max_event_size: int | None = DEFAULT_MAX_EVENT_SIZE_BYTES,
) -> Generator[EventSource]

Connect to a server-sent events endpoint and yield an EventSource.

Iterating the EventSource yields ServerSentEvent instances.

Parameters: See httpx2.request.

Returns

Generator[EventSource]

websocket 
def websocket(
    url: URL | str,
    *,
    max_message_size_bytes: int = DEFAULT_MAX_MESSAGE_SIZE_BYTES,
    queue_size: int = DEFAULT_QUEUE_SIZE,
    keepalive_ping_interval_seconds: float | None = DEFAULT_KEEPALIVE_PING_INTERVAL_SECONDS,
    keepalive_ping_timeout_seconds: float | None = DEFAULT_KEEPALIVE_PING_TIMEOUT_SECONDS,
    subprotocols: list[str] | None = None,
    params: QueryParamTypes | None = None,
    headers: HeaderTypes | None = None,
    cookies: CookieTypes | None = None,
    auth: AuthTypes | UseClientDefault | None = USE_CLIENT_DEFAULT,
    follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
    timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
    extensions: RequestExtensions | None = None,
) -> Generator[WebSocketSession]

Open a WebSocket session, using this client’s configuration.

The session is closed automatically when exiting the context manager.

with httpx2.Client() as client:
    with client.websocket("ws://localhost:8000/ws") as ws:
        ws.send_text("Hello!")
        message = ws.receive_text()
Returns

Generator[WebSocketSession]

build_request 
def build_request(
    method: str,
    url: URL | str,
    *,
    content: RequestContent | None = None,
    data: RequestData | None = None,
    files: RequestFiles | None = None,
    json: typing.Any | None = None,
    params: QueryParamTypes | None = None,
    headers: HeaderTypes | None = None,
    cookies: CookieTypes | None = None,
    timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
    extensions: RequestExtensions | None = None,
) -> Request

Build and return a request instance.

  • The params, headers and cookies arguments are merged with any values set on the client.
  • The url argument is merged with any base_url set on the client.

See also: Request instances

Returns

Request

send 
def send(
    request: Request,
    *,
    stream: bool = False,
    auth: AuthTypes | UseClientDefault | None = USE_CLIENT_DEFAULT,
    follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
) -> Response

Send a request.

The request is sent as-is, unmodified.

Typically you’ll want to build one with Client.build_request() so that any client-level configuration is merged into the request, but passing an explicit httpx2.Request() is supported as well.

See also: Request instances

Returns

Response

close 
def close() -> None

Close transport and proxies.

Returns

None

AsyncClient

AsyncClient

Bases: BaseClient

An asynchronous HTTP client, with connection pooling, HTTP/2, redirects, cookie persistence, etc.

It can be shared between tasks.

Usage:

>>> async with httpx2.AsyncClient() as client:
>>>     response = await client.get('https://example.org')

Parameters:

  • auth - (optional) An authentication class to use when sending requests.
  • params - (optional) Query parameters to include in request URLs, as a string, dictionary, or sequence of two-tuples.
  • headers - (optional) Dictionary of HTTP headers to include when sending requests.
  • cookies - (optional) Dictionary of Cookie items to include when sending requests.
  • verify - (optional) Either True to use an SSL context with the default CA bundle, False to disable verification, or an instance of ssl.SSLContext to use a custom context.
  • http2 - (optional) A boolean indicating if HTTP/2 support should be enabled. Defaults to False.
  • proxy - (optional) A proxy URL where all the traffic should be routed.
  • mounts - (optional) A dictionary mapping URL patterns to transports, used to route requests through specific transports based on the URL.
  • timeout - (optional) The timeout configuration to use when sending requests.
  • limits - (optional) The limits configuration to use.
  • max_redirects - (optional) The maximum number of redirect responses that should be followed.
  • base_url - (optional) A URL to use as the base when building request URLs.
  • transport - (optional) A transport class to use for sending requests over the network.
  • trust_env - (optional) Enables or disables usage of environment variables for configuration.
  • default_encoding - (optional) The default encoding to use for decoding response text, if no charset information is included in a response Content-Type header. Set to a callable for automatic character set detection. Default: “utf-8”.

Attributes

headers

HTTP headers to include when sending requests.

Type: Headers

cookies

Cookie values to include when sending requests.

Type: Cookies

params

Query parameters to include in the URL when sending requests.

Type: QueryParams

auth

Authentication class used when none is passed at the request-level.

See also Authentication.

Type: Auth | None

Methods

request

@async

def request(
    method: str,
    url: URL | str,
    *,
    content: RequestContent | None = None,
    data: RequestData | None = None,
    files: RequestFiles | None = None,
    json: typing.Any | None = None,
    params: QueryParamTypes | None = None,
    headers: HeaderTypes | None = None,
    cookies: CookieTypes | None = None,
    auth: AuthTypes | UseClientDefault | None = USE_CLIENT_DEFAULT,
    follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
    timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
    extensions: RequestExtensions | None = None,
) -> Response

Build and send a request.

Equivalent to:

request = client.build_request(...)
response = await client.send(request, ...)

See AsyncClient.build_request(), AsyncClient.send() and Merging of configuration for how the various parameters are merged with client-level configuration.

Returns

Response

get

@async

def get(
    url: URL | str,
    *,
    params: QueryParamTypes | None = None,
    headers: HeaderTypes | None = None,
    cookies: CookieTypes | None = None,
    auth: AuthTypes | UseClientDefault | None = USE_CLIENT_DEFAULT,
    follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
    timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
    extensions: RequestExtensions | None = None,
) -> Response

Send a GET request.

Parameters: See httpx2.request.

Returns

Response

head

@async

def head(
    url: URL | str,
    *,
    params: QueryParamTypes | None = None,
    headers: HeaderTypes | None = None,
    cookies: CookieTypes | None = None,
    auth: AuthTypes | UseClientDefault = USE_CLIENT_DEFAULT,
    follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
    timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
    extensions: RequestExtensions | None = None,
) -> Response

Send a HEAD request.

Parameters: See httpx2.request.

Returns

Response

options

@async

def options(
    url: URL | str,
    *,
    params: QueryParamTypes | None = None,
    headers: HeaderTypes | None = None,
    cookies: CookieTypes | None = None,
    auth: AuthTypes | UseClientDefault = USE_CLIENT_DEFAULT,
    follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
    timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
    extensions: RequestExtensions | None = None,
) -> Response

Send an OPTIONS request.

Parameters: See httpx2.request.

Returns

Response

post

@async

def post(
    url: URL | str,
    *,
    content: RequestContent | None = None,
    data: RequestData | None = None,
    files: RequestFiles | None = None,
    json: typing.Any | None = None,
    params: QueryParamTypes | None = None,
    headers: HeaderTypes | None = None,
    cookies: CookieTypes | None = None,
    auth: AuthTypes | UseClientDefault = USE_CLIENT_DEFAULT,
    follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
    timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
    extensions: RequestExtensions | None = None,
) -> Response

Send a POST request.

Parameters: See httpx2.request.

Returns

Response

put

@async

def put(
    url: URL | str,
    *,
    content: RequestContent | None = None,
    data: RequestData | None = None,
    files: RequestFiles | None = None,
    json: typing.Any | None = None,
    params: QueryParamTypes | None = None,
    headers: HeaderTypes | None = None,
    cookies: CookieTypes | None = None,
    auth: AuthTypes | UseClientDefault = USE_CLIENT_DEFAULT,
    follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
    timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
    extensions: RequestExtensions | None = None,
) -> Response

Send a PUT request.

Parameters: See httpx2.request.

Returns

Response

patch

@async

def patch(
    url: URL | str,
    *,
    content: RequestContent | None = None,
    data: RequestData | None = None,
    files: RequestFiles | None = None,
    json: typing.Any | None = None,
    params: QueryParamTypes | None = None,
    headers: HeaderTypes | None = None,
    cookies: CookieTypes | None = None,
    auth: AuthTypes | UseClientDefault = USE_CLIENT_DEFAULT,
    follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
    timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
    extensions: RequestExtensions | None = None,
) -> Response

Send a PATCH request.

Parameters: See httpx2.request.

Returns

Response

delete

@async

def delete(
    url: URL | str,
    *,
    params: QueryParamTypes | None = None,
    headers: HeaderTypes | None = None,
    cookies: CookieTypes | None = None,
    auth: AuthTypes | UseClientDefault = USE_CLIENT_DEFAULT,
    follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
    timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
    extensions: RequestExtensions | None = None,
) -> Response

Send a DELETE request.

Parameters: See httpx2.request.

Returns

Response

query

@async

def query(
    url: URL | str,
    *,
    content: RequestContent | None = None,
    data: RequestData | None = None,
    files: RequestFiles | None = None,
    json: typing.Any | None = None,
    params: QueryParamTypes | None = None,
    headers: HeaderTypes | None = None,
    cookies: CookieTypes | None = None,
    auth: AuthTypes | UseClientDefault = USE_CLIENT_DEFAULT,
    follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
    timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
    extensions: RequestExtensions | None = None,
) -> Response

Send a QUERY request.

Parameters: See httpx2.request.

Returns

Response

stream

@async

def stream(
    method: str,
    url: URL | str,
    *,
    content: RequestContent | None = None,
    data: RequestData | None = None,
    files: RequestFiles | None = None,
    json: typing.Any | None = None,
    params: QueryParamTypes | None = None,
    headers: HeaderTypes | None = None,
    cookies: CookieTypes | None = None,
    auth: AuthTypes | UseClientDefault | None = USE_CLIENT_DEFAULT,
    follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
    timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
    extensions: RequestExtensions | None = None,
) -> AsyncGenerator[Response]

Alternative to httpx2.request() that streams the response body instead of loading it into memory at once.

Parameters: See httpx2.request.

See also: Streaming Responses

Returns

AsyncGenerator[Response]

sse

@async

def sse(
    url: URL | str,
    *,
    method: str = 'GET',
    content: RequestContent | None = None,
    data: RequestData | None = None,
    files: RequestFiles | None = None,
    json: typing.Any | None = None,
    params: QueryParamTypes | None = None,
    headers: HeaderTypes | None = None,
    cookies: CookieTypes | None = None,
    auth: AuthTypes | UseClientDefault | None = USE_CLIENT_DEFAULT,
    follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
    timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
    extensions: RequestExtensions | None = None,
    max_event_size: int | None = DEFAULT_MAX_EVENT_SIZE_BYTES,
) -> AsyncGenerator[EventSource]

Connect to a server-sent events endpoint and yield an EventSource.

Iterating the EventSource yields ServerSentEvent instances.

Parameters: See httpx2.request.

Returns

AsyncGenerator[EventSource]

websocket

@async

def websocket(
    url: URL | str,
    *,
    max_message_size_bytes: int = DEFAULT_MAX_MESSAGE_SIZE_BYTES,
    queue_size: int = DEFAULT_QUEUE_SIZE,
    keepalive_ping_interval_seconds: float | None = DEFAULT_KEEPALIVE_PING_INTERVAL_SECONDS,
    keepalive_ping_timeout_seconds: float | None = DEFAULT_KEEPALIVE_PING_TIMEOUT_SECONDS,
    subprotocols: list[str] | None = None,
    params: QueryParamTypes | None = None,
    headers: HeaderTypes | None = None,
    cookies: CookieTypes | None = None,
    auth: AuthTypes | UseClientDefault | None = USE_CLIENT_DEFAULT,
    follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
    timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
    extensions: RequestExtensions | None = None,
) -> AsyncGenerator[AsyncWebSocketSession]

Open a WebSocket session, using this client’s configuration.

The session is closed automatically when exiting the context manager.

async with httpx2.AsyncClient() as client:
    async with client.websocket("ws://localhost:8000/ws") as ws:
        await ws.send_text("Hello!")
        message = await ws.receive_text()
Returns

AsyncGenerator[AsyncWebSocketSession]

build_request 
def build_request(
    method: str,
    url: URL | str,
    *,
    content: RequestContent | None = None,
    data: RequestData | None = None,
    files: RequestFiles | None = None,
    json: typing.Any | None = None,
    params: QueryParamTypes | None = None,
    headers: HeaderTypes | None = None,
    cookies: CookieTypes | None = None,
    timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
    extensions: RequestExtensions | None = None,
) -> Request

Build and return a request instance.

  • The params, headers and cookies arguments are merged with any values set on the client.
  • The url argument is merged with any base_url set on the client.

See also: Request instances

Returns

Request

send

@async

def send(
    request: Request,
    *,
    stream: bool = False,
    auth: AuthTypes | UseClientDefault | None = USE_CLIENT_DEFAULT,
    follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
) -> Response

Send a request.

The request is sent as-is, unmodified.

Typically you’ll want to build one with AsyncClient.build_request() so that any client-level configuration is merged into the request, but passing an explicit httpx2.Request() is supported as well.

See also: Request instances

Returns

Response

aclose

@async

def aclose() -> None

Close transport and proxies.

Returns

None

Response

An HTTP response.

  • def __init__(...)
  • .status_code - int
  • .reason_phrase - str
  • .http_version - "HTTP/2" or "HTTP/1.1"
  • .url - URL
  • .headers - Headers
  • .content - bytes
  • .text - str
  • .encoding - str
  • .is_redirect - bool
  • .request - Request
  • .next_request - Optional[Request]
  • .cookies - Cookies
  • .history - List[Response]
  • .elapsed - timedelta
    • The amount of time elapsed between sending the request and calling close() on the corresponding response received for that request. total_seconds() to correctly get the total elapsed seconds.
  • def .raise_for_status() - Response
  • def .json() - Any
  • def .read() - bytes
  • def .iter_raw([chunk_size]) - bytes iterator
  • def .iter_bytes([chunk_size]) - bytes iterator
  • def .iter_text([chunk_size]) - text iterator
  • def .iter_lines() - text iterator
  • def .close() - None
  • def .next() - Response
  • def .aread() - bytes
  • def .aiter_raw([chunk_size]) - async bytes iterator
  • def .aiter_bytes([chunk_size]) - async bytes iterator
  • def .aiter_text([chunk_size]) - async text iterator
  • def .aiter_lines() - async text iterator
  • def .aclose() - None
  • def .anext() - Response

Request

An HTTP request. Can be constructed explicitly for more control over exactly what gets sent over the wire.

>>> request = httpx2.Request("GET", "https://example.org", headers={'host': 'example.org'})
>>> response = client.send(request)
  • def __init__(method, url, [params], [headers], [cookies], [content], [data], [files], [json], [stream])
  • .method - str
  • .url - URL
  • .content - byte, byte iterator, or byte async iterator
  • .headers - Headers
  • .cookies - Cookies

URL

A normalized, IDNA supporting URL.

>>> url = URL("https://example.org/")
>>> url.host
'example.org'
  • def __init__(url, **kwargs)
  • .scheme - str
  • .authority - str
  • .host - str
  • .port - int
  • .path - str
  • .query - str
  • .raw_path - str
  • .fragment - str
  • .origin - Origin
  • .is_ssl - bool
  • .is_absolute_url - bool
  • .is_relative_url - bool
  • def .copy_with([scheme], [authority], [path], [query], [fragment]) - URL

Origin

An immutable, hashable set of normalized scheme, host, and effective port information.

>>> URL('https://example.org').origin == URL('HTTPS://EXAMPLE.ORG:443').origin
True
>>> URL('wss://[::1]/socket').origin
Origin(scheme='wss', host='::1', port=443)
  • def __init__(url)
  • .scheme - str
  • .host - str
  • .port - int or None

Headers

A case-insensitive multi-dict.

>>> headers = Headers({'Content-Type': 'application/json'})
>>> headers['content-type']
'application/json'
  • def __init__(self, headers, encoding=None)
  • def copy() - Headers

Cookies

A dict-like cookie store.

>>> cookies = Cookies()
>>> cookies.set("name", "value", domain="example.org")
  • def __init__(cookies: [dict, Cookies, CookieJar])
  • .jar - CookieJar
  • def extract_cookies(response)
  • def set_cookie_header(request)
  • def set(name, value, [domain], [path])
  • def get(name, [domain], [path])
  • def delete(name, [domain], [path])
  • def clear([domain], [path])
  • Standard mutable mapping interface

Proxy

A configuration of the proxy server.

>>> proxy = Proxy("http://proxy.example.com:8030")
>>> client = Client(proxy=proxy)
  • def __init__(url, [ssl_context], [auth], [headers])
  • .url - URL
  • .auth - tuple[str, str]
  • .headers - Headers
  • .ssl_context - SSLContext

EventSource

EventSource

Attributes

response

Type: Response

ServerSentEvent

ServerSentEvent

Attributes

event

Type: str Default: 'message'

data

Type: str Default: ''

id

Type: str Default: ''

retry

Type: int | None Default: None

Methods

json 
def json() -> object
Returns

object

WebSocketSession

WebSocketSession

Sync context manager representing an opened WebSocket session.

Attributes

subprotocol

Optional protocol that has been accepted by the server.

Type: typing.Optional[str]

response

The webSocket handshake response.

Type: Response | None

Methods

send_text 
def send_text(data: str) -> None

Send a text message.

Returns

None

Parameters

data : str

The text to send.

Raises
  • WebSocketNetworkError — A network error occured.
send_bytes 
def send_bytes(data: bytes) -> None

Send a bytes message.

Returns

None

Parameters

data : bytes

The data to send.

Raises
  • WebSocketNetworkError — A network error occured.
send_json 
def send_json(data: typing.Any, mode: JSONMode = 'text') -> None

Send JSON data.

Returns

None

Parameters

data : typing.Any

The data to send. Must be serializable by json.dumps.

mode : JSONMode Default: 'text'

The sending mode. Should either be 'text' or 'bytes'.

Raises
  • WebSocketNetworkError — A network error occured.
receive_text 
def receive_text(timeout: float | None = None) -> str

Receive text from the server.

Returns

str — Text data.

Parameters

timeout : float | None Default: None

Number of seconds to wait for an event. If None, will block until an event is available.

Raises
  • TimeoutError — No event was received before the timeout delay.
  • WebSocketDisconnect — The server closed the websocket.
  • WebSocketNetworkError — A network error occured.
  • WebSocketInvalidTypeReceived — The received event was not a text message.
receive_bytes 
def receive_bytes(timeout: float | None = None) -> bytes

Receive bytes from the server.

Returns

bytes — Bytes data.

Parameters

timeout : float | None Default: None

Number of seconds to wait for an event. If None, will block until an event is available.

Raises
  • TimeoutError — No event was received before the timeout delay.
  • WebSocketDisconnect — The server closed the websocket.
  • WebSocketNetworkError — A network error occured.
  • WebSocketInvalidTypeReceived — The received event was not a bytes message.
receive_json 
def receive_json(timeout: float | None = None, mode: JSONMode = 'text') -> typing.Any

Receive JSON data from the server.

The received data should be parseable by json.loads.

Returns

typing.Any — Parsed JSON data.

Parameters

timeout : float | None Default: None

Number of seconds to wait for an event. If None, will block until an event is available.

mode : JSONMode Default: 'text'

Receive mode. Should either be 'text' or 'bytes'.

Raises
  • TimeoutError — No event was received before the timeout delay.
  • WebSocketDisconnect — The server closed the websocket.
  • WebSocketNetworkError — A network error occured.
  • WebSocketInvalidTypeReceived — The received event didn’t correspond to the specified mode.
ping 
def ping(payload: bytes = b'') -> threading.Event

Send a Ping message.

Returns

threading.Event — An event that can be used to wait for the corresponding Pong response.

Parameters

payload : bytes Default: b''

Payload to attach to the Ping event. Internally, it’s used to track this specific event. If left empty, a random one will be generated.

close 
def close(code: int = 1000, reason: str | None = None) -> None

Close the WebSocket session.

Internally, it’ll send the CloseConnection event.

This method is automatically called when exiting the context manager.

Returns

None

Parameters

code : int Default: 1000

The integer close code to indicate why the connection has closed.

reason : str | None Default: None

Additional reasoning for why the connection has closed.

AsyncWebSocketSession

AsyncWebSocketSession

Bases: AsyncContextManagerMixin

Async context manager representing an opened WebSocket session.

Internally, this session uses an anyio task group to manage background tasks. As a result, exceptions that are not caught inside the context manager and propagate out of the async with block will be wrapped in an ExceptionGroup.

To handle them, use the except* syntax:

async with AsyncWebSocketSession(stream) as ws: try: data = await ws.receive_text() except WebSocketDisconnect:

Caught inside the context manager: plain exception.

print(“Connection closed”)

If not caught inside:

try: async with AsyncWebSocketSession(stream) as ws: data = await ws.receive_text() except* WebSocketDisconnect:

Propagated out of the context manager: wrapped in ExceptionGroup.

print(“Connection closed”)

Attributes

subprotocol

Optional protocol that has been accepted by the server.

Type: typing.Optional[str]

response

The webSocket handshake response.

Type: Response | None

Methods

send_text

@async

def send_text(data: str) -> None

Send a text message.

Returns

None

Parameters

data : str

The text to send.

Raises
  • WebSocketNetworkError — A network error occured.
send_bytes

@async

def send_bytes(data: bytes) -> None

Send a bytes message.

Returns

None

Parameters

data : bytes

The data to send.

Raises
  • WebSocketNetworkError — A network error occured.
send_json

@async

def send_json(data: typing.Any, mode: JSONMode = 'text') -> None

Send JSON data.

Returns

None

Parameters

data : typing.Any

The data to send. Must be serializable by json.dumps.

mode : JSONMode Default: 'text'

The sending mode. Should either be 'text' or 'bytes'.

Raises
  • WebSocketNetworkError — A network error occured.
receive_text

@async

def receive_text(timeout: float | None = None) -> str

Receive text from the server.

Returns

str — Text data.

Parameters

timeout : float | None Default: None

Number of seconds to wait for an event. If None, will block until an event is available.

Raises
  • TimeoutError — No event was received before the timeout delay.
  • WebSocketDisconnect — The server closed the websocket.
  • WebSocketNetworkError — A network error occured.
  • WebSocketInvalidTypeReceived — The received event was not a text message.
receive_bytes

@async

def receive_bytes(timeout: float | None = None) -> bytes

Receive bytes from the server.

Returns

bytes — Bytes data.

Parameters

timeout : float | None Default: None

Number of seconds to wait for an event. If None, will block until an event is available.

Raises
  • TimeoutError — No event was received before the timeout delay.
  • WebSocketDisconnect — The server closed the websocket.
  • WebSocketNetworkError — A network error occured.
  • WebSocketInvalidTypeReceived — The received event was not a bytes message.
receive_json

@async

def receive_json(timeout: float | None = None, mode: JSONMode = 'text') -> typing.Any

Receive JSON data from the server.

The received data should be parseable by json.loads.

Returns

typing.Any — Parsed JSON data.

Parameters

timeout : float | None Default: None

Number of seconds to wait for an event. If None, will block until an event is available.

mode : JSONMode Default: 'text'

Receive mode. Should either be 'text' or 'bytes'.

Raises
  • TimeoutError — No event was received before the timeout delay.
  • WebSocketDisconnect — The server closed the websocket.
  • WebSocketNetworkError — A network error occured.
  • WebSocketInvalidTypeReceived — The received event didn’t correspond to the specified mode.
ping

@async

def ping(payload: bytes = b'') -> anyio.Event

Send a Ping message.

Returns

anyio.Event — An event that can be used to wait for the corresponding Pong response.

Parameters

payload : bytes Default: b''

Payload to attach to the Ping event. Internally, it’s used to track this specific event. If left empty, a random one will be generated.

close

@async

def close(code: int = 1000, reason: str | None = None) -> None

Close the WebSocket session.

Internally, it’ll send the CloseConnection event.

This method is automatically called when exiting the context manager.

Returns

None

Parameters

code : int Default: 1000

The integer close code to indicate why the connection has closed.

reason : str | None Default: None

Additional reasoning for why the connection has closed.

alias_httpx

alias_httpx 

def alias_httpx() -> None

Make import httpx resolve to httpx2, and import httpcore to httpcore2, process-wide.

Intended for applications migrating from httpx, so that dependencies still importing httpx or httpcore share the httpx2 classes. Libraries should never call this.

Must be called before anything imports httpx or httpcore. Calling it again is a no-op.

Returns

None