| from __future__ import annotations |
|
|
| import typing |
| from collections.abc import Generator |
| from contextlib import contextmanager |
|
|
| from ._client import Client |
| from ._config import ( |
| DEFAULT_KEEPALIVE_PING_INTERVAL_SECONDS, |
| DEFAULT_KEEPALIVE_PING_TIMEOUT_SECONDS, |
| DEFAULT_MAX_MESSAGE_SIZE_BYTES, |
| DEFAULT_QUEUE_SIZE, |
| DEFAULT_TIMEOUT_CONFIG, |
| ) |
| from ._models import Response |
| from ._types import ( |
| AuthTypes, |
| CookieTypes, |
| HeaderTypes, |
| ProxyTypes, |
| QueryParamTypes, |
| RequestContent, |
| RequestData, |
| RequestFiles, |
| TimeoutTypes, |
| ) |
| from ._urls import URL |
|
|
| if typing.TYPE_CHECKING: |
| import ssl |
|
|
| from .websockets._api import WebSocketSession |
|
|
|
|
| __all__ = [ |
| "delete", |
| "get", |
| "head", |
| "options", |
| "patch", |
| "post", |
| "put", |
| "query", |
| "request", |
| "stream", |
| "websocket", |
| ] |
|
|
|
|
| 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. |
| |
| Parameters: |
| method: HTTP method for the new `Request` object: `GET`, `OPTIONS`, `HEAD`, `POST`, `PUT`, `PATCH`, `DELETE`, |
| or `QUERY`. |
| url: URL for the new `Request` object. |
| params: *(optional)* Query parameters to include in the URL, as a string, dictionary, or sequence of two-tuples. |
| content: *(optional)* Binary content to include in the body of the request, as bytes or a byte iterator. |
| data: *(optional)* Form data to include in the body of the request, as a dictionary. |
| files: *(optional)* A dictionary of upload files to include in the body of the request. |
| json: *(optional)* A JSON serializable object to include in the body of the request. |
| headers: *(optional)* Dictionary of HTTP headers to include in the request. |
| cookies: *(optional)* Dictionary of Cookie items to include in the request. |
| auth: *(optional)* An authentication class to use when sending the request. |
| proxy: *(optional)* A proxy URL where all the traffic should be routed. |
| timeout: *(optional)* The timeout configuration to use when sending the request. |
| follow_redirects: *(optional)* Enables or disables HTTP redirects. |
| 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. |
| trust_env: *(optional)* Enables or disables usage of environment variables for configuration. |
| |
| Returns: |
| The `Response` object. |
| |
| Usage: |
| |
| ``` |
| >>> import httpx2 |
| >>> response = httpx2.request('GET', 'https://httpbin.org/get') |
| >>> response |
| <Response [200 OK]> |
| ``` |
| """ |
| with Client( |
| cookies=cookies, |
| proxy=proxy, |
| verify=verify, |
| timeout=timeout, |
| trust_env=trust_env, |
| ) as client: |
| return client.request( |
| method=method, |
| url=url, |
| content=content, |
| data=data, |
| files=files, |
| json=json, |
| params=params, |
| headers=headers, |
| auth=auth, |
| follow_redirects=follow_redirects, |
| ) |
|
|
|
|
| @contextmanager |
| 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][0] |
| |
| [0]: quickstart.md#streaming-responses |
| """ |
| with Client( |
| cookies=cookies, |
| proxy=proxy, |
| verify=verify, |
| timeout=timeout, |
| trust_env=trust_env, |
| ) as client: |
| with client.stream( |
| method=method, |
| url=url, |
| content=content, |
| data=data, |
| files=files, |
| json=json, |
| params=params, |
| headers=headers, |
| auth=auth, |
| follow_redirects=follow_redirects, |
| ) as response: |
| yield response |
|
|
|
|
| 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. |
| """ |
| return request( |
| "GET", |
| url, |
| params=params, |
| headers=headers, |
| cookies=cookies, |
| auth=auth, |
| proxy=proxy, |
| follow_redirects=follow_redirects, |
| verify=verify, |
| timeout=timeout, |
| trust_env=trust_env, |
| ) |
|
|
|
|
| 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. |
| """ |
| return request( |
| "OPTIONS", |
| url, |
| params=params, |
| headers=headers, |
| cookies=cookies, |
| auth=auth, |
| proxy=proxy, |
| follow_redirects=follow_redirects, |
| verify=verify, |
| timeout=timeout, |
| trust_env=trust_env, |
| ) |
|
|
|
|
| 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. |
| """ |
| return request( |
| "HEAD", |
| url, |
| params=params, |
| headers=headers, |
| cookies=cookies, |
| auth=auth, |
| proxy=proxy, |
| follow_redirects=follow_redirects, |
| verify=verify, |
| timeout=timeout, |
| trust_env=trust_env, |
| ) |
|
|
|
|
| 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`. |
| """ |
| return request( |
| "POST", |
| url, |
| content=content, |
| data=data, |
| files=files, |
| json=json, |
| params=params, |
| headers=headers, |
| cookies=cookies, |
| auth=auth, |
| proxy=proxy, |
| follow_redirects=follow_redirects, |
| verify=verify, |
| timeout=timeout, |
| trust_env=trust_env, |
| ) |
|
|
|
|
| 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`. |
| """ |
| return request( |
| "PUT", |
| url, |
| content=content, |
| data=data, |
| files=files, |
| json=json, |
| params=params, |
| headers=headers, |
| cookies=cookies, |
| auth=auth, |
| proxy=proxy, |
| follow_redirects=follow_redirects, |
| verify=verify, |
| timeout=timeout, |
| trust_env=trust_env, |
| ) |
|
|
|
|
| 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`. |
| """ |
| return request( |
| "PATCH", |
| url, |
| content=content, |
| data=data, |
| files=files, |
| json=json, |
| params=params, |
| headers=headers, |
| cookies=cookies, |
| auth=auth, |
| proxy=proxy, |
| follow_redirects=follow_redirects, |
| verify=verify, |
| timeout=timeout, |
| trust_env=trust_env, |
| ) |
|
|
|
|
| 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. |
| """ |
| return request( |
| "DELETE", |
| url, |
| params=params, |
| headers=headers, |
| cookies=cookies, |
| auth=auth, |
| proxy=proxy, |
| follow_redirects=follow_redirects, |
| verify=verify, |
| timeout=timeout, |
| trust_env=trust_env, |
| ) |
|
|
|
|
| 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`. |
| """ |
| return request( |
| "QUERY", |
| url, |
| content=content, |
| data=data, |
| files=files, |
| json=json, |
| params=params, |
| headers=headers, |
| cookies=cookies, |
| auth=auth, |
| proxy=proxy, |
| follow_redirects=follow_redirects, |
| verify=verify, |
| timeout=timeout, |
| trust_env=trust_env, |
| ) |
|
|
|
|
| @contextmanager |
| 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. |
| |
| ```python |
| 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`. |
| """ |
| with Client( |
| cookies=cookies, |
| proxy=proxy, |
| verify=verify, |
| timeout=timeout, |
| trust_env=trust_env, |
| ) as client: |
| with client.websocket( |
| url, |
| params=params, |
| headers=headers, |
| auth=auth, |
| follow_redirects=follow_redirects, |
| subprotocols=subprotocols, |
| max_message_size_bytes=max_message_size_bytes, |
| queue_size=queue_size, |
| keepalive_ping_interval_seconds=keepalive_ping_interval_seconds, |
| keepalive_ping_timeout_seconds=keepalive_ping_timeout_seconds, |
| ) as session: |
| yield session |
|
|
