High-level Twisted HTTP Client API for asynchronous HTTP requests in Python
Core HTTP request functions supporting all standard methods with comprehensive parameter options for headers, authentication, cookies, timeouts, and data handling. All functions return Twisted Deferreds that fire with response objects.
Makes HTTP GET requests for retrieving data from servers.
def get(url, headers=None, **kwargs):
"""
Make a GET request.
Parameters:
- url: str, bytes, EncodedURL, or DecodedURL - Target URL
- headers: Headers, dict, or None - HTTP headers to send
- params: dict, list of tuples, or None - Query string parameters
- auth: tuple of (username, password) or None - HTTP Basic Auth
- cookies: CookieJar, dict, or None - Cookies to send
- timeout: int or None - Request timeout in seconds
- allow_redirects: bool - Follow redirects (default: True)
- browser_like_redirects: bool - Use browser-like redirect behavior (default: False)
- unbuffered: bool - Disable response buffering (default: False)
- reactor: Twisted reactor or None - Custom reactor
- persistent: bool - Use persistent connections (default: True)
- agent: IAgent or None - Custom agent
Returns:
Deferred that fires with _Response object
"""Makes HTTP POST requests for sending data to servers.
def post(url, data=None, **kwargs):
"""
Make a POST request.
Parameters:
- url: str, bytes, EncodedURL, or DecodedURL - Target URL
- data: bytes, file-like, dict, list, or None - Request body data
- json: dict, list, or None - JSON data (mutually exclusive with data)
- files: dict, list, or None - Files to upload (creates multipart request)
- headers: Headers, dict, or None - HTTP headers to send
- params: dict, list of tuples, or None - Query string parameters
- auth: tuple of (username, password) or None - HTTP Basic Auth
- cookies: CookieJar, dict, or None - Cookies to send
- timeout: int or None - Request timeout in seconds
- allow_redirects: bool - Follow redirects (default: True)
- browser_like_redirects: bool - Use browser-like redirect behavior (default: False)
- unbuffered: bool - Disable response buffering (default: False)
- reactor: Twisted reactor or None - Custom reactor
- persistent: bool - Use persistent connections (default: True)
- agent: IAgent or None - Custom agent
Returns:
Deferred that fires with _Response object
"""Makes HTTP PUT requests for creating or updating resources.
def put(url, data=None, **kwargs):
"""
Make a PUT request.
Parameters:
- url: str, bytes, EncodedURL, or DecodedURL - Target URL
- data: bytes, file-like, dict, list, or None - Request body data
- json: dict, list, or None - JSON data (mutually exclusive with data)
- files: dict, list, or None - Files to upload (creates multipart request)
- headers: Headers, dict, or None - HTTP headers to send
- params: dict, list of tuples, or None - Query string parameters
- auth: tuple of (username, password) or None - HTTP Basic Auth
- cookies: CookieJar, dict, or None - Cookies to send
- timeout: int or None - Request timeout in seconds
- allow_redirects: bool - Follow redirects (default: True)
- browser_like_redirects: bool - Use browser-like redirect behavior (default: False)
- unbuffered: bool - Disable response buffering (default: False)
- reactor: Twisted reactor or None - Custom reactor
- persistent: bool - Use persistent connections (default: True)
- agent: IAgent or None - Custom agent
Returns:
Deferred that fires with _Response object
"""Makes HTTP PATCH requests for partial resource updates.
def patch(url, data=None, **kwargs):
"""
Make a PATCH request.
Parameters:
- url: str, bytes, EncodedURL, or DecodedURL - Target URL
- data: bytes, file-like, dict, list, or None - Request body data
- json: dict, list, or None - JSON data (mutually exclusive with data)
- files: dict, list, or None - Files to upload (creates multipart request)
- headers: Headers, dict, or None - HTTP headers to send
- params: dict, list of tuples, or None - Query string parameters
- auth: tuple of (username, password) or None - HTTP Basic Auth
- cookies: CookieJar, dict, or None - Cookies to send
- timeout: int or None - Request timeout in seconds
- allow_redirects: bool - Follow redirects (default: True)
- browser_like_redirects: bool - Use browser-like redirect behavior (default: False)
- unbuffered: bool - Disable response buffering (default: False)
- reactor: Twisted reactor or None - Custom reactor
- persistent: bool - Use persistent connections (default: True)
- agent: IAgent or None - Custom agent
Returns:
Deferred that fires with _Response object
"""Makes HTTP DELETE requests for removing resources.
def delete(url, **kwargs):
"""
Make a DELETE request.
Parameters:
- url: str, bytes, EncodedURL, or DecodedURL - Target URL
- headers: Headers, dict, or None - HTTP headers to send
- params: dict, list of tuples, or None - Query string parameters
- auth: tuple of (username, password) or None - HTTP Basic Auth
- cookies: CookieJar, dict, or None - Cookies to send
- timeout: int or None - Request timeout in seconds
- allow_redirects: bool - Follow redirects (default: True)
- browser_like_redirects: bool - Use browser-like redirect behavior (default: False)
- unbuffered: bool - Disable response buffering (default: False)
- reactor: Twisted reactor or None - Custom reactor
- persistent: bool - Use persistent connections (default: True)
- agent: IAgent or None - Custom agent
Returns:
Deferred that fires with _Response object
"""Makes HTTP HEAD requests for retrieving headers without response body.
def head(url, **kwargs):
"""
Make a HEAD request.
Parameters:
- url: str, bytes, EncodedURL, or DecodedURL - Target URL
- headers: Headers, dict, or None - HTTP headers to send
- params: dict, list of tuples, or None - Query string parameters
- auth: tuple of (username, password) or None - HTTP Basic Auth
- cookies: CookieJar, dict, or None - Cookies to send
- timeout: int or None - Request timeout in seconds
- allow_redirects: bool - Follow redirects (default: True)
- browser_like_redirects: bool - Use browser-like redirect behavior (default: False)
- unbuffered: bool - Disable response buffering (default: False)
- reactor: Twisted reactor or None - Custom reactor
- persistent: bool - Use persistent connections (default: True)
- agent: IAgent or None - Custom agent
Returns:
Deferred that fires with _Response object
"""Makes HTTP requests with any method.
def request(method, url, **kwargs):
"""
Make an HTTP request with any method.
Parameters:
- method: str - HTTP method ('GET', 'POST', 'PUT', 'PATCH', 'DELETE', 'HEAD', etc.)
- url: str, bytes, EncodedURL, or DecodedURL - Target URL
- headers: Headers, dict, or None - HTTP headers to send
- params: dict, list of tuples, or None - Query string parameters
- data: bytes, file-like, dict, list, or None - Request body data
- json: dict, list, or None - JSON data (mutually exclusive with data)
- files: dict, list, or None - Files to upload (creates multipart request)
- auth: tuple of (username, password) or None - HTTP Basic Auth
- cookies: CookieJar, dict, or None - Cookies to send
- timeout: int or None - Request timeout in seconds
- allow_redirects: bool - Follow redirects (default: True)
- browser_like_redirects: bool - Use browser-like redirect behavior (default: False)
- unbuffered: bool - Disable response buffering (default: False)
- reactor: Twisted reactor or None - Custom reactor
- persistent: bool - Use persistent connections (default: True)
- agent: IAgent or None - Custom agent
Returns:
Deferred that fires with _Response object
"""import treq
from twisted.internet import defer
@defer.inlineCallbacks
def make_requests():
# Simple GET
response = yield treq.get('https://httpbin.org/get')
# GET with query parameters
response = yield treq.get('https://httpbin.org/get', params={'key': 'value'})
# POST with form data
response = yield treq.post('https://httpbin.org/post', data={'field': 'value'})
# POST with JSON
response = yield treq.post('https://httpbin.org/post', json={'key': 'value'})@defer.inlineCallbacks
def upload_file():
# Upload a single file
with open('document.pdf', 'rb') as f:
response = yield treq.post(
'https://httpbin.org/post',
files={'file': f}
)
# Upload with additional form data
with open('image.jpg', 'rb') as f:
response = yield treq.post(
'https://httpbin.org/post',
data={'title': 'My Upload'},
files={'image': f}
)
# Upload with custom filename and content type
with open('data.txt', 'rb') as f:
response = yield treq.post(
'https://httpbin.org/post',
files={'data': ('custom_name.txt', 'text/plain', f)}
)@defer.inlineCallbacks
def authenticated_requests():
# Basic authentication
response = yield treq.get(
'https://httpbin.org/basic-auth/user/pass',
auth=('user', 'pass')
)
# Custom headers
response = yield treq.get(
'https://httpbin.org/get',
headers={'User-Agent': 'My App 1.0', 'Accept': 'application/json'}
)
# Cookies
response = yield treq.get(
'https://httpbin.org/cookies',
cookies={'session': 'abc123', 'preference': 'dark'}
)Request parameter types:
# URL parameter types
_URLType = Union[str, bytes, EncodedURL, DecodedURL]
# Query parameters
_ParamsType = Union[
Mapping[str, Union[str, Tuple[str, ...], List[str]]],
List[Tuple[str, str]]
]
# Request body data
_DataType = Union[
bytes, # Raw bytes
io.BytesIO, # Byte stream
io.BufferedReader, # File object
IBodyProducer, # Twisted body producer
Dict[str, str], # Form data
List[Tuple[str, str]] # Form tuples
]
# File upload types
_FileValue = Union[
str, # Simple string content
bytes, # Byte content
Tuple[str, str, IBodyProducer] # (filename, content_type, producer)
]
_FilesType = Union[
Mapping[str, _FileValue], # Dict mapping field names to files
Iterable[Tuple[str, _FileValue]] # List of (field_name, file) tuples
]
# JSON data
_JSONType = Any # Any JSON-serializable Python objectInstall with Tessl CLI
npx tessl i tessl/pypi-treq