Docs at https://httpy.readthedocs.io/
A Python lightweight socket-based library to create HTTP(s) and WebSocket connections.
- Cookies support
- Caching support
- Easy debugging
- HTTP Basic and Digest authentication
- Form support
- Keep-Alive and Sessions support
- JSON support
- Sessions support
- Runs in PyPy
- Independent of http.client
- HTTP/2 Support
- Async IO support
- Python>=3.6
git clone https://github.com/jenca-adam/httpycd httpypython3 setup.py install
The Python version check will be performed automatically
python3 -m pip install httpy
yay -S httpy
It's easy.
import httpy
resp = httpy.request("https://example.com/") # Do a request
resp.content #Access contentSet the http_version argument, but keep in mind the following
- You can't make an asynchronous request using HTTP/1.1
- HTTP/2 requests can't be performed over insecure (http scheme) connections.
If you don't set it, the HTTP version will be automatically detected using ALPN <https://datatracker.ietf.org/doc/html/rfc7301>.
Valid http_version values are "1.1" and "2".
import httpy
pending = httpy.request("https://example.com/", blocking = False)PendingRequest.response returns the result of the response. You can
check if the request is already done using PendingRequest.finished
The Dir class allows you to store httpy's data (cache and cookies)
on the path of your choice. By default, the data is stored in
~/.cache/httpy. If you want to store the data without using the Dir class, use the
enable_cookies or enable_cache argument of request.
import httpy
directory = httpy.Dir("your/path")
directory.request("https://example.com/") # ...If you want to reuse a connection, it is highly recommended to use a
Session class. It offers more control over connection closure than
the standard request
import httpy
session = httpy.Session()
session.request("https://example.com/")HTTPy sets Connection: close by default in non-Session requests. If
you want to keep the connection alive outside a session, you must
specify so in the headers argument.
You can perform async requests using the async_request method.
The simplest use case:
import httpy
async def my_function():
return await httpy.request("https://example.com/")If you want to perform multiple requests at once on the same connection
(i.e. with asyncio.gather), use the initiate_http2_connection
method of Session:
import httpy
import asyncio
async def my_function():
session = httpy.Session()
await session.initiate_http2_connection(host="example.com")
return await asyncio.gather(*(session.async_request("https://www.example.com/") for _ in range(69)))Session and Dir and everything with a request() method has
an async_request() equivalent.
If you want to receive the response as a stream, set the stream argument of request to True. A Stream or AsyncStream is returned. They both have the read() method. It returns the given number of bytes of the response body. If no arguments are given, the entire rest of the body is read and returned.
You can access the current stream state using stream.state. It contains some useful information about the stream. Status and headers are also available directly (stream.status, stream.headers).
Attributes: * bytes_read * body * connection * finished
Warning
The stream.state.bytes_read attribute represents the amount of bytes received from the server and is not representative of the actual number of bytes read from the stream. For this use stream.bytes_read instead. The same applies for stream.state.body
import httpy
stream = httpy.request("https://example.com/big-file", stream=True)
stream.read(1) # read 1 byte
stream.read(6) # read 6 bytes
stream.bytes_read # 7
stream.read() # read the rest
stream.state.finished #TrueThe Response class returned by request() has some useful
attributes:
The response content as bytes. Example:
import httpy
resp = httpy.request("https://www.google.com/")
print(resp.content)
#b'!<doctype html>\n<html>...The response status as a Status object. Example:
import httpy
resp = httpy.request("https://www.example.com/this_url_doesnt_exist")
print(resp.status)
# 404
print(resp.status.reason)
# NOT FOUND
print(resp.status.description)
# indicates that the origin server did not find a current representation for the target resource or is not willing to disclose that one exists.
print(resp.status>400)
# TrueStatus subclasses int.
All the redirects on the way to this response as list.
Example:
import httpy
resp = httpy.request("https://httpbin.org/redirect/1")
print(resp.history)
# [<Response GET [302 Found] (https://httpbin.org/redirect/1/)>, <Response GET [200 OK] (https://httpbin.org/get/)>]Response.history is ordered from oldest to newest
Indicates whether the response was loaded from cache (bool).
Example:
import httpy
resp = httpy.request("https://example.com/")
print(resp.fromcache)
# False
resp = httpy.request("https://example.com/")
print(resp.fromcache)
# TrueSome of the attributes of the request that produced this response, as a
Request object.
Request's attributes
Request.url- the URL requested (str)Request.headers- the requests' headers (Headers)Request.socket- the underlying connection (eithersocket.socketorhttpy.http2.connection.HTTP2Connection)Request.cache- the same asResponse.fromcache(bool)Request.http_version- the HTTP version used (str)Request.method- the HTTP method used (str)
Example:
import httpy
resp = httpy.request("https://example.com/")
print(resp.request.url)
# https://example.com/
print(resp.request.headers)
# {'Accept-Encoding': 'gzip, deflate, identity', 'Host': 'example.com', 'User-Agent': 'httpy/2.0.0', 'Connection': 'close', 'Accept': '*/*'}
print(resp.request.method)
# GETRaw content received from the server, not decoded with Content-Encoding
(bytes).
Example:
import httpy
resp = httpy.request("https://example.com/")
print(resp.original_content)
# b'\x1f\x8b\x08\x00\xc2 ...Time the request took, in seconds. Only the loading time of this
particular request, doesn't account for redirects. (float).
Example:
import httpy
resp = httpy.request("https://example.com/")
print(resp.time_elapsed)
# 0.2497The download speed for the response, in bytes per second. (float).
Might be different for HTTP/2 request. Example:
import httpy
resp = httpy.request("https://example.com/")
print(resp.speed)
# 2863.79The response's Content-Type header contents, with the charset
information stripped. If the headers lack Content-Type, it's
text/html by default.
import httpy
resp = httpy.request("https://example.com/")
print(resp.content_type)
# text/htmlGets the charset of the response (str or None):
- If a charset was specified in the response headers, return it
- If a charset was not specified, but
chardetis available, try to detect the charset (Note that this still returnsNoneifchardetfails) - If a charset was not specified, and
chardetis not available, returnNone
Example:
import httpy
resp = httpy.request("https://example.com/")
print(resp.charset)
# UTF-8Response.content, decoded using Response.charset (str)
Warning
Do not try to access Response.string, if Response.charset is
None, unless you are absolutely sure the response data is
decodable by the default locale encoding.
For ASCII responses this is probably harmless, but you have been warned!
Example:
import httpy
resp = httpy.request("https://example.com/")
print(resp.string)
#<!doctype html>
...If Response.content_type is application/json, try to parse
Response.string using JSON. Throw an error otherwise.
Warning
The same as above applies.
Example:
import httpy
resp = httpy.request("https://httpbin.org/get")
print(resp.json["url"])
# https://httpbin.org/getThe same as Response.request.method
Easy again...
>>> import httpy
>>> sock = httpy.WebSocket("wss://echo.websocket.events/")# create a websocket client(echo server example)
>>> sock.send("Hello, world!💥")# you can send also bytes
>>> sock.recv()
"Hello, world!💥"import httpy
resp = httpy.request("https://example.com/", method="POST", body = {"foo":"bar"})
# ...import httpy
resp = httpy.request("https://example.com/", method = "POST", body = { "foo" : "bar", "file" : httpy.File.open( "example.txt" ) })
# ...import httpy
resp = httpy.request("https://example.com/", method = "POST", body= b" Hello, World ! ")
# ...resp = httpy.request("https://example.com/", method = "POST", body = "I support Ãœnicode !")
# ...resp = httpy.request("https://example.com/", method = "POST", body = "{\"foo\" : \"bar\" }", content_type = "application/json")
# ...Just set debug to True :
>>> import httpy
>>> httpy.request("https://example.com/",debug=True)
[INFO][request](1266): request() called.
[INFO][_raw_request](1112): _raw_request() called.
[INFO][_raw_request](1113): Accessing cache.
[INFO][_raw_request](1120): No data in cache.
[INFO][_raw_request](1151): Establishing connection
[INFO]Connection[__init__](778): Created new Connection upon <socket.socket fd=3, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=6, laddr=('192.168.100.88', 58998), raddr=('93.184.216.34', 443)>
send:
GET / HTTP/1.1
Accept-Encoding: gzip, deflate, identity
Host: www.example.com
User-Agent: httpy/1.1.0
Connection: keep-alive
response:
HTTP/1.1 200 OK
Content-Encoding: gzip
Age: 438765
Cache-Control: max-age=604800
Content-Type: text/html; charset=UTF-8
Date: Wed, 13 Apr 2022 12:59:07 GMT
Etag: "3147526947 gzip"
Expires: Wed, 20 Apr 2022 12:59:07 GMT
Last-Modified: Thu, 17 Oct 2019 07:18:26 GMT
Server: ECS (dcb/7F37)
Vary: Accept-Encoding
X-Cache: HIT
Content-Length: 648
<Response [200 OK] (https://www.example.com/)>