Module: anynet.http

Provides HTTP-related classes, including a client and a server. Note that this implementation is by no means complete. Only basic HTTP features are supported.

class HTTPError(Exception)
General exception for errors related to HTTP.

class HTTPResponseError(HTTPError)
May be raised when the status code of an HTTP response indicates an error.

class HTTPMessage
Base class for HTTP messages. This class should not be instantiated directly. Instead, one of its subclasses should be used.

class HTTPRequest(HTTPMessage)
An HTTP request object.

class HTTPResponse(HTTPMessage)
An HTTP response object.

class HTTPClient
A reusable HTTP client.

class HTTPRouter
Routes incoming HTTP requests based on the request path.

async def head(url: str, **kwargs) -> HTTPResponse
async def get(url: str, **kwargs) -> HTTPResponse
async def post(url: str, **kwargs) -> HTTPResponse
async def put(url: str, **kwargs) -> HTTPResponse
async def patch(url: str, **kwargs) -> HTTPResponse
async def delete(url: str, **kwargs) -> HTTPResponse
Performs a HEAD, GET, POST, PUT, PATCH or DELETE request. These methods are provided for convenience.

url must contain at least the hostname or IP address of the server, and the path of the HTTP request. Scheme and port are optional. Example: https://example.com:8080/test.html.

The following keyword arguments may be provided to initialize the HTTP request: headers, body, text, files, boundary, form, rawform, json, xml, params, and continue_threshold. If no Host header is given it is filled in automatically based on the given url.

Other keyword arguments are passed on to request().

async def request(url: str, req: HTTPRequest, context: TLSContext = None, **kwargs) -> HTTPResponse
Performs an HTTP request on a new connection.

url must contain at least the hostname or IP address of the server. Scheme and port are optional. Example: https://example.com:8080.

If no scheme is provided, the connection is secured with TLS precisely if a TLS context is provided. If the scheme is https but no TLS context is provided the connection is secured with the default TLS context.

The keyword arguments are forwarded to HTTPClient.request.

async with connect(url: str, context: TLSContext = None) -> HTTPClient
Creates a reusable connection with the server. Blocks until the connection is ready. The parameters url and context have the same meaning as in request().

HTTPClient is not task-safe. Do not try to perform multiple request on a single client concurrently.

async with serve(handler: Callable, host: str = "", port: int = 0, context: TLSContext = None) -> None
Creates an HTTP server at the given address. If host is empty, the local address of the default gateway is used. If port is 0, it is chosen by the operating system. If context is provided, the server is secured with TLS.

handler must be an async function that takes a TLSClient and an HTTPRequest and returns an HTTPResponse. If handler raises an exception or returns anything other than a HTTPResponse, the server sends an empty HTTP response with status code 500 to the client.

async with serve_router(host: str = "", port: int = 0, context: TLSContext = None) -> HTTPRouter
Creates an HTTP server at the given address. If host is empty, the local address of the default gateway is used. If port is 0, it is chosen by the operating system. If context is provided, the server is secured with TLS.

A HTTPRouter is returned that may be used to attach handlers to request paths.

def current_date() -> str
Returns the current date and time in the format of the Date header.

def format_date(date: datetime.datetime) -> str
Returns the given datetime in the format of the Date header.

def parse_date(text: str) -> datetime.datetime
Parses the given Date header text and returns a datetime object.

def urlencode(data: str) -> str
Applies url-encoding on the given string (i.e. replaces special characters by %XX).

def urldecode(data: str) -> str
Decodes the given url-encoded string.

def formencode(data: MultiDict[str, str], url: bool = True) -> str
Encodes data using form-encoding. If url is True, field names and values are url-encoded automatically.

def formdecode(data: str, url: bool = True) -> MultiDict[str, str]
Parses a form-encoded string. If url is True, field names and values are automatically url-decoded.

HTTPMessage

This is the base class of HTTPRequest and HTTPResponse. This class should not be instantiated directly. Instead, one of its subclasses should be used.

This class provides several attributes that define the body of the HTTP message. In general, only one of them should be used. When the HTTP message is encoded, the attributes are evaluated in the following order: rawform, form, json, xml, files, text and body. The first attribute that is not None defines the body of the HTTP request. The others are ignored.

Most headers are left unchanged when the HTTP message is encoded. However, if no Content-Type header is present, a default is chosen based on the attribute that defines the body, unless the body is empty. The Content-Length header is always overwritten, unless the Transfer-Encoding is chunked or the body is empty.

When an HTTP message is parsed, the body attribute is always filled in. The other attributes are only filled if they fit the Content-Type of the HTTP message.

version: str = "HTTP/1.1"
The version of the HTTP message. Only HTTP/1.1 is supported.

headers: CIMultiDict[str, str] = {}
The HTTP headers (case insensitive dictionary).

body: bytes = ""
The raw body of the HTTP message. The Content-Type defaults to application/octet-stream.

text: str = None
The decoded body of the HTTP message, if applicable. The Content-Type defaults to text/plain.

form: MultiDict[str, str] = None
The form parameters in the body. The parameters are url-encoded automatically. The Content-Type defaults to application/x-www-form-urlencoded.

rawform: MultiDict[str, str] = None
The form parameters in the body. The difference with the form attribute is that the names and values are not url-encoded automatically. The Content-Type defaults to application/x-www-form-urlencoded.

json: dict = None
The JSON body. The Content-Type defaults to application/json.

xml: XMLTree = None
An XMLTree that represents the body. The Content-Type defaults to application/xml.

files: MultiDict[str, bytes] = None
A list of binary files. The Content-Type defaults to multipart/form-data.

boundary: str = "--------BOUNDARY--------"
The boundary string that's used if the body is encoded from files.

json_options: dict = {"separators": (",", ":")}
Additional keyword arguments that are passed to json.dumps.

def encode() -> bytes
Encodes the HTTP message.

def encode_headers() -> bytes
Encodes only the headers of the HTTP message.

def encode_body() -> bytes
Encodes only the body of the HTTP message.

@classmethod
def parse(data: bytes, head: bool = False) -> HTTPMessage
Parses an HTTP message. This method should not be called on the HTTPMessage class. Instead, it should be called on one of its subclasses. Raises HTTPError if the given data does not contain a valid HTTP message. If head is True, only the headers of the HTTP message are parsed and data must not contain the body.

HTTPRequest

This class inherits HTTPMessage. During encoding, the Expect header is set to 100-continue if the size of the body exceeds the given threshold.

method: str = "GET"
The HTTP method.

path: str = "/"
The path of the HTTP request, without the parameters.

params: MultiDict[str, str] = None
The GET parameters behind the path. The parameters are url-encoded automatically.

continue_threshold: int = 1024
The size of the body after which the Expect header is set to 100-continue. If set to None, the Expect header is never modified, regardless of the body size.

def __init__()
Creates a new HTTP request.

@classmethod
def head(path: str) -> HTTPRequest
def get(path: str) -> HTTPRequest
def post(path: str) -> HTTPRequest
def put(path: str) -> HTTPRequest
def patch(path: str) -> HTTPRequest
def delete(path: str) -> HTTPRequest
Creates a HEAD, GET, POST, PUT, PATCH or DELETE request with the given path. If the given path contains form parameters, they are parsed and put into params.

HTTPResponse

This class inherits HTTPMessage.

status_code: int
The status code of the HTTP response.

status_name: str
The reason string of the HTTP response.

upgrade: Callable = None
If set, this async function is called after the HTTP response is sent to the client. Only relevant on the server side.

def __init__(status_code: int = 500)
Creates a new HTTP response with the given status code. status_name is derived from the given status code. If the given status code is not recognized, status_name is set to "Unknown"

def success() -> bool
Returns True if the status code indicates success, i.e. if it has the form 2xx.

def error() -> bool
This is the reverse of success. Returns True if the status code does not indicate success.

def raise_if_error() -> None
Raises HTTPResponseError if the status code does not indicate success.

HTTPClient

async def request(req: HTTPRequest, *, headerfunc: Callable = None, writefunc: Callable = None) -> HTTPResponse
Performs an HTTP request.

If headerfunc is provided, it must be an async function that takes a HTTPResponse object. It is called once after all HTTP headers are received.

If writefunc is provided, it must be an async function that takes a bytes object. It is called whenever a part of the body is received from the server.

async def close() -> None
Closes the connection.

def local_address() -> tuple[str, int]
Returns the local address of the client.

def remote_address() -> tuple[str, int]
Returns the remote address of the client.

def remote_certificate() -> TLSCertificate
Returns the certificate that was provided by the other side of the connection. Returns None if the connection is not secured with TLS.

HTTPRouter

def route(path: str, handler: Callable)
Attaches handler to the given path. Incoming requests are routed to handler only if the paths are exactly the same. Raises ValueError if the given path is already in use.

This method may be called in a with statement to remove the handler automatically.

def remove(path: str)
Removes the handler that is attached to the given path.

HTTPResponseError

This is a subclass of HTTPError.

response: HTTPResponse
The HTTP response that caused the error.

def __init__(response: HTTPResponse)
Creates a new HTTPResponseError for the given HTTP response.