Netizen

Netizen is a minimalist HTTP client with a symmetrical interface between async and sync modes. It doesn't aim to be feature-complete like requests or httpx.

Netizen is just enough for poking API endpoints and performing basic HTTP operations or testing. It suits me, as I prefer working closely with sockets and don't need high-level abstraction.

Features

• Symmetrical interface, e.g. client.send() vs await client.send()
• The retries parameter makes it resilient and prevents flaky tests
• ~500 lines of code
• No dependencies other than the Python Standard Library

Installation

pip install git+https://github.com/nggit/netizen.git

timeout and retries parameters

These two are related, but what you probably want to know is how retries work.

Basically, here is the flow of an HTTP request:

1. Connect to the host and port
2. Send the request header and (optionally) the body
3. Receive the response header
4. Stream the body if necessary

Netizen only guarantees retries until the 3rd stage is successful.

Retries in stage 1 are useful for waiting for the server to be ready to accept connections. This is suitable for testing without having to worry about sequence or timing.

retries depends on timeout to determine the interval value:

interval = timeout / retries

Handling a JSON response body

import asyncio

from netizen import HTTPClient


# sync
with HTTPClient('ip-api.com', 80) as client:
    response = client.send(b'GET /json HTTP/1.1')

    print(response.json())

# async
async def main():
    async with HTTPClient('ip-api.com', 80) as client:
        response = await client.send(b'GET /json HTTP/1.1')

        print(await response.json())

asyncio.run(main())

Handling a raw response body with streaming

import asyncio

from netizen import HTTPClient

client = HTTPClient('example.com', 80)


# sync
with client:
    response = client.send(b'GET / HTTP/1.1')

    for data in response:
        print('Received:', len(data), 'Bytes')

# async
async def main():
    async with client:
        response = await client.send(b'GET / HTTP/1.1')

        async for data in response:
            print('Received:', len(data), 'Bytes')

asyncio.run(main())

Append request headers via *args also body parameter

with HTTPClient('example.com', 80) as client:
    response = client.send(
        b'POST / HTTP/1.1',
        b'Content-Type: application/json',
        b'Content-Length: 14',
        body=b'{"foo": "bar"}'
    )

    print('Status code:', response.status)  # 403
    print('Reason phrase:', response.message)  # b'Forbidden'

# out of context, close the connection without reading the entire response body

If you don't specify any headers, then Content-Length will be automatically inserted along with Content-Type: application/x-www-form-urlencoded.

with HTTPClient('example.com', 80) as client:
    response = client.send(b'POST / HTTP/1.1', body=b'foo=bar')

Send multiple requests within the same context/connection

with HTTPClient('ip-api.com', 80) as client:
    # first request
    response = client.send(b'GET /json HTTP/1.1')

    # the first response body must be consumed before sending another one
    print(response.json())

    # second request
    response = client.send(b'GET /json HTTP/1.1')

    print(response.body())

Handling URL redirects

from urllib.parse import urlparse


with HTTPClient('google.com', 443, ssl=True) as client:
    response = client.send(b'GET / HTTP/1.1')

    print('1. Status code:', response.status)  # 301
    print('1. Reason phrase:', response.message)  # b'Moved Permanently'
    print('1. Location:', response.url)  # b'http://www.google.com/'

    for data in response:
        pass

    if response.url:
        url = urlparse(response.url)

        if url.netloc:  # b'www.google.com' (different host)
            with HTTPClient(url.netloc.decode(), 443, ssl=True) as client:
                response = client.send(b'GET %s HTTP/1.1' % url.path)

                print('2. Status code:', response.status)  # 200
                print('2. Reason phrase:', response.message)  # b'OK'

                for data in response:
                    pass
        else:
            pass

Working directly with socket using client.sendall() and client.recv()

with HTTPClient('localhost', 8000, timeout=10, retries=10) as client:
    response = client.send(
        b'GET /chat HTTP/1.1',
        b'Upgrade: WebSocket',
        b'Connection: Upgrade',
        b'Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==',
        b'Sec-WebSocket-Version: 13'
    )

    if response.status == 101:
        client.sendall(b'\x81\x0dHello, World!\x88\x02\x03\xe8')
        print('Received:', client.recv(4096))

Manually craft a bad request with client.sendall() and client.end()

with HTTPClient('localhost', 8000) as client:
    client.sendall(
        b'POST /upload HTTP/1.1\r\n'
        b'Host: localhost\r\n'
        b'Content-Length: 5\r\n'
        b'Transfer-Encoding: chunked\r\n\r\n'
    )
    client.sendall(b'0\r\n\r\n')

    # we are not using the `client.send()`, but we need the response object?
    response = client.end()

    print(response.body())

License

MIT License