adafruit_httpserver

Socket based HTTP Server for CircuitPython

  • Author(s): Dan Halbert, Michał Pokusa

Implementation Notes

Software and Dependencies:

adafruit_httpserver.server

  • Author(s): Dan Halbert, Michał Pokusa

class adafruit_httpserver.server.Server(socket_source: _ISocketPool, root_path: str = None, *, debug: bool = False)

A basic socket-based HTTP server.

Create a server, and get it ready to run.

Parameters:

  • socket – An object that is a source of sockets. This could be a socketpool in CircuitPython or the socket module in CPython.

  • root_path (str) – Root directory to serve files from

  • debug (bool) – Enables debug messages useful during development

add_routes(routes: List[Route]) → None

Add multiple routes at once.

Parameters:

routes (List[Route]) – List of routes to add to the server

Example:

from separate_file import external_route1, external_route2

...

server.add_routes([
    Route("/example", GET, route_func1, append_slash=True),
    Route("/example/<my_parameter>", GET, route_func2),
    Route("/example/..../something", [GET, POST], route_func3),
    external_route1,
    external_route2,
]}
property headers: Headers

Headers to be sent with every response, without the need to specify them in each handler.

If a header is specified in both the handler and the server, the handler’s header will be used.

Example:

server = Server(pool, "/static")
server.headers = {
    "X-Server": "Adafruit CircuitPython HTTP Server",
    "Access-Control-Allow-Origin": "*",
}
host: str

Host name or IP address the server is listening on. None if server is stopped.

poll() → str

Call this method inside your main loop to get the server to check for new incoming client requests. When a request comes in, it will be handled by the handler function.

Returns str representing the result of the poll e.g. NO_REQUEST or REQUEST_HANDLED_RESPONSE_SENT.

port: int

Port the server is listening on. None if server is stopped.

property request_buffer_size: int

The maximum size of the incoming request buffer. If the default size isn’t adequate to handle your incoming data you can set this after creating the server instance.

Default size is 1024 bytes.

Example:

server = Server(pool, "/static")
server.request_buffer_size = 2048

server.serve_forever(str(wifi.radio.ipv4_address))
require_authentication(auths: List[Basic | Token | Bearer]) → None

Requires authentication for all routes and files in root_path. Any non-authenticated request will be rejected with a 401 status code.

Example:

server = Server(pool, "/static")
server.require_authentication([Basic("username", "password")])
root_path: str

Root directory to serve files from. None if serving files is disabled.

route(path: str, methods: str | Iterable[str] = GET, *, append_slash: bool = False) → Callable

Decorator used to add a route.

If request matches multiple routes, the first matched one added will be used.

Parameters:

  • path (str) – URL path

  • methods (str) – HTTP method(s): "GET", "POST", ["GET", "POST"] etc.

  • append_slash (bool) – If True, the route will be accessible with and without a trailing slash

Example:

# Default method is GET
@server.route("/example")
def route_func(request):
    ...

# It is necessary to specify other methods like POST, PUT, etc.
@server.route("/example", POST)
def route_func(request):
    ...

# If you want to access URL with and without trailing slash, use append_slash=True
@server.route("/example-with-slash", append_slash=True)
# which is equivalent to
@server.route("/example-with-slash")
@server.route("/example-with-slash/")
def route_func(request):
    ...

# Multiple methods can be specified
@server.route("/example", [GET, POST])
def route_func(request):
    ...

# URL parameters can be specified
@server.route("/example/<my_parameter>", GET) e.g. /example/123
def route_func(request, my_parameter):
    ...

# It is possible to use wildcard that can match any number of path segments
@server.route("/example/.../something", GET) # e.g. /example/123/something
@server.route("/example/..../something", GET) # e.g. /example/123/456/something
def route_func(request):
    ...
serve_forever(host: str = '0.0.0.0', port: int = 5000, *, poll_interval: float = 0.1) → None

Wait for HTTP requests at the given host and port. Does not return. Ignores any exceptions raised by the handler function and continues to serve. Returns only when the server is stopped by calling .stop().

Parameters:

  • host (str) – host name or IP address

  • port (int) – port

  • poll_interval (float) – interval between polls in seconds

property socket_timeout: int

Timeout after which the socket will stop waiting for more incoming data.

Must be set to positive integer or float. Default is 1 second.

When exceeded, raises OSError with errno.ETIMEDOUT.

Example:

server = Server(pool, "/static")
server.socket_timeout = 3

server.serve_forever(str(wifi.radio.ipv4_address))
start(host: str = '0.0.0.0', port: int = 5000) → None

Start the HTTP server at the given host and port. Requires calling .poll() in a while loop to handle incoming requests.

Parameters:

  • host (str) – host name or IP address

  • port (int) – port

stop() → None

Stops the server from listening for new connections and closes the socket. Current requests will be processed. Server can be started again by calling .start() or .serve_forever().

adafruit_httpserver.request

  • Author(s): Dan Halbert, Michał Pokusa

class adafruit_httpserver.request.File(filename: str, content_type: str, content: str | bytes)

Class representing a file uploaded via POST.

Examples:

file = request.form_data.files.get("uploaded_file")
# File(filename="foo.txt", content_type="text/plain", size=14)

file.content
# "Hello, world!\n"
content: str | bytes

Content of the file.

property content_bytes: bytes

Content of the file as bytes. It is recommended to use this instead of content as it will always return bytes.

Example:

file = request.form_data.files.get("uploaded_file")

with open(file.filename, "wb") as f:
    f.write(file.content_bytes)
content_type: str

Content type of the file.

filename: str

Filename of the file.

property size: int

Length of the file content.

class adafruit_httpserver.request.Files

Class for files uploaded via POST.

property fields

Returns a list of field names.

get(field_name: str, default: Any = None) → File | Any | None

Get the value of a field.

get_list(field_name: str) → List[File]

Get the list of values of a field.

items()

Returns a list of (name, value) tuples.

keys()

Returns a list of header names.

values()

Returns a list of header values.

class adafruit_httpserver.request.FormData(data: bytes, headers: Headers, *, debug: bool = False)

Class for parsing and storing form data from POST requests.

Supports application/x-www-form-urlencoded, multipart/form-data and text/plain content types.

Examples:

form_data = FormData(b"foo=bar&baz=qux&baz=quuz", "application/x-www-form-urlencoded")
# or
form_data = FormData(b"foo=bar\r\nbaz=qux\r\nbaz=quux", "text/plain")
# FormData({"foo": ["bar"], "baz": ["qux", "quux"]})

form_data.get("foo") # "bar"
form_data["foo"] # "bar"
form_data.get("non-existent-key") # None
form_data.get_list("baz") # ["qux", "quux"]
"unknown-key" in form_data # False
form_data.fields # ["foo", "baz"]
property fields

Returns a list of field names.

get(field_name: str, default: str | bytes = None, *, safe=True) → str | bytes | None

Get the value of a field.

get_list(field_name: str, *, safe=True) → List[str | bytes]

Get the list of values of a field.

items()

Returns a list of (name, value) tuples.

keys()

Returns a list of header names.

values()

Returns a list of header values.

class adafruit_httpserver.request.QueryParams(query_string: str)

Class for parsing and storing GET query parameters requests.

Examples:

query_params = QueryParams("foo=bar&baz=qux&baz=quux")
# QueryParams({"foo": ["bar"], "baz": ["qux", "quux"]})

query_params.get("foo") # "bar"
query_params["foo"] # "bar"
query_params.get("non-existent-key") # None
query_params.get_list("baz") # ["qux", "quux"]
"unknown-key" in query_params # False
query_params.fields # ["foo", "baz"]
property fields

Returns a list of field names.

get(field_name: str, default: str = None, *, safe=True) → str | None

Get the value of a field.

get_list(field_name: str, *, safe=True) → List[str]

Get the list of values of a field.

items()

Returns a list of (name, value) tuples.

keys()

Returns a list of header names.

values()

Returns a list of header values.

class adafruit_httpserver.request.Request(server: Server, connection: _ISocket, client_address: Tuple[str, int], raw_request: bytes = None)

Incoming request, constructed from raw incoming bytes. It is passed as first argument to all route handlers.

property body: bytes

Body of the request, as bytes.

client_address: Tuple[str, int]

Address and port bound to the socket on the other end of the connection.

Example:

request.client_address  # ('192.168.137.1', 40684)
connection: _ISocket

Socket object used to send and receive data on the connection.

property cookies: Dict[str, str]

Cookies sent with the request.

Example:

request.headers["Cookie"]
# "foo=bar; baz=qux; foo=quux"

request.cookies
# {"foo": "quux", "baz": "qux"}
property form_data: FormData | None

POST data of the request.

Example:

# application/x-www-form-urlencoded
request = Request(...,
    raw_request=b"""...
    foo=bar&baz=qux"""
)

# or

# multipart/form-data
request = Request(...,
    raw_request=b"""...
    --boundary
    Content-Disposition: form-data; name="foo"

    bar
    --boundary
    Content-Disposition: form-data; name="baz"

    qux
    --boundary--"""
)

# or

# text/plain
request = Request(...,
    raw_request=b"""...
    foo=bar
    baz=qux
    """
)

request.form_data                  # FormData({'foo': ['bar'], 'baz': ['qux']})
request.form_data["foo"]           # "bar"
request.form_data.get_list("baz")  # ["qux"]
headers: Headers

Headers from the request.

http_version: str

HTTP version, e.g. "HTTP/1.1".

json() → dict | None

Body of the request, as a JSON-decoded dictionary. Only available for POST, PUT, PATCH and DELETE requests.

method: str

Request method e.g. “GET” or “POST”.

path: str

Path of the request, e.g. "/foo/bar".

query_params: QueryParams

Query/GET parameters in the request.

Example:

request  = Request(..., raw_request=b"GET /?foo=bar&baz=qux HTTP/1.1...")

request.query_params                  # QueryParams({"foo": "bar"})
request.query_params["foo"]           # "bar"
request.query_params.get_list("baz")  # ["qux"]
raw_request: bytes

Raw bytes that were received from the client.

Should not be modified directly.

server: Server

Server object that received the request.

adafruit_httpserver.response

  • Author(s): Dan Halbert, Michał Pokusa

class adafruit_httpserver.response.ChunkedResponse(request: Request, body: Generator[str | bytes, Any, Any], *, status: Status | Tuple[int, str] = OK_200, headers: Headers | Dict[str, str] = None, cookies: Dict[str, str] = None, content_type: str = None)

Specialized version of Response class for sending data using chunked transfer encoding.

Instead of requiring the whole content to be passed to the constructor, it expects a generator that yields chunks of data.

Example:

@server.route(path, method)
def route_func(request: Request):

    def body():
        yield "Some ch"
        yield "unked co"
        yield "ntent"

    return ChunkedResponse(request, body, content_type="text/plain")
Parameters:

  • request (Request) – Request object

  • body (Generator) – Generator that yields chunks of data.

  • status (Status) – Status object or tuple with code and message.

  • headers (Headers) – Headers to be sent with the response.

  • cookies (Dict[str, str]) – Cookies to be sent with the response.

  • content_type (str) – Content type of the response.

class adafruit_httpserver.response.FileResponse(request: Request, filename: str = 'index.html', root_path: str = None, *, status: Status | Tuple[int, str] = OK_200, headers: Headers | Dict[str, str] = None, cookies: Dict[str, str] = None, content_type: str = None, as_attachment: bool = False, download_filename: str = None, buffer_size: int = 1024, head_only: bool = False, safe: bool = True)

Specialized version of Response class for sending files.

Instead of body it takes filename and root_path arguments. It is also possible to send only headers with head_only argument or modify buffer_size.

If browsers should download the file instead of displaying it, use as_attachment and download_filename arguments.

Example:

@server.route(path, method)
def route_func(request: Request):

    return FileResponse(request, filename='index.html', root_path='/www')
Parameters:

  • request (Request) – Request that this is a response to.

  • filename (str) – Name of the file to send.

  • root_path (str) – Path to the root directory from which to serve files. Defaults to server’s root_path.

  • status (Status) – Status code and text. Defaults to 200 OK.

  • headers (Headers) – Headers to include in response.

  • cookies (Dict[str, str]) – Cookies to be sent with the response.

  • content_type (str) – Content type of response.

  • as_attachment (bool) – If True, the file will be sent as an attachment.

  • download_filename (str) – Name of the file to send as an attachment.

  • buffer_size (int) – Size of the buffer used to send the file. Defaults to 1024.

  • head_only (bool) – If True, only headers will be sent. Defaults to False.

  • safe (bool) – If True, checks if filename is valid. Defaults to True.

class adafruit_httpserver.response.JSONResponse(request: Request, data: Dict[Any, Any], *, headers: Headers | Dict[str, str] = None, cookies: Dict[str, str] = None, status: Status | Tuple[int, str] = OK_200)

Specialized version of Response class for sending JSON data.

Instead of requiring body to be passed to the constructor, it expects data to be passed instead.

Example:

@server.route(path, method)
def route_func(request: Request):

    return JSONResponse(request, {"key": "value"})
Parameters:

  • request (Request) – Request that this is a response to.

  • data (dict) – Data to be sent as JSON.

  • headers (Headers) – Headers to include in response.

  • cookies (Dict[str, str]) – Cookies to be sent with the response.

  • status (Status) – Status code and text. Defaults to 200 OK.

class adafruit_httpserver.response.Redirect(request: Request, url: str, *, permanent: bool = False, preserve_method: bool = False, status: Status | Tuple[int, str] = None, headers: Headers | Dict[str, str] = None, cookies: Dict[str, str] = None)

Specialized version of Response class for redirecting to another URL.

Instead of requiring the body to be passed to the constructor, it expects a URL to redirect to.

Example:

@server.route(path, method)
def route_func(request: Request):

    return Redirect(request, "https://www.example.com")

By default uses permament and preserve_method to determine the status code to use, but if you prefer you can specify it directly.

Note that 301 Moved Permanently and 302 Found can change the method to GET while 307 Temporary Redirect and 308 Permanent Redirect preserve the method.

More information: https://developer.mozilla.org/en-US/docs/Web/HTTP/Status#redirection_messages

Parameters:

  • request (Request) – Request that this is a response to.

  • url (str) – URL to redirect to.

  • permanent (bool) – Whether to use a permanent redirect or a temporary one.

  • preserve_method (bool) – Whether to preserve the method of the request.

  • status (Status) – Status object or tuple with code and message.

  • headers (Headers) – Headers to include in response.

  • cookies (Dict[str, str]) – Cookies to be sent with the response.

class adafruit_httpserver.response.Response(request: Request, body: str | bytes = '', *, status: Status | Tuple[int, str] = OK_200, headers: Headers | Dict[str, str] = None, cookies: Dict[str, str] = None, content_type: str = None)

Response to a given Request. Use in Server.route handler functions.

Base class for all other response classes.

Example:

@server.route(path, method)
def route_func(request: Request):

    return Response(request, body='Some content', content_type="text/plain")
Parameters:

  • request (Request) – Request that this is a response to.

  • body (str) – Body of response. Defaults to empty string.

  • status (Status) – Status code and text. Defaults to 200 OK.

  • headers (Headers) – Headers to include in response. Defaults to empty dict.

  • cookies (Dict[str, str]) – Cookies to be sent with the response.

  • content_type (str) – Content type of response. Defaults to None.

class adafruit_httpserver.response.SSEResponse(request: Request, headers: Headers | Dict[str, str] = None, cookies: Dict[str, str] = None)

Specialized version of Response class for sending Server-Sent Events.

Allows one way communication with the client using a persistent connection.

Keep in mind, that in order to send events, the socket must be kept open. This means that you have to store the response object somewhere, so you can send events to it and close it later.

It is very important to close the connection manually, it will not be done automatically.

Example:

sse = None

@server.route(path, method)
def route_func(request: Request):

    # Store the response object somewhere in global scope
    global sse
    sse = SSEResponse(request)

    return sse

...

# Later, when you want to send an event
sse.send_event("Simple message")
sse.send_event("Message", event="event_name", id=1, retry=5000)

# Close the connection
sse.close()
Parameters:

  • request (Request) – Request object

  • headers (Headers) – Headers to be sent with the response.

  • cookies (Dict[str, str]) – Cookies to be sent with the response.

close()

Close the connection.

Always call this method when you are done sending events.

send_event(data: str, event: str = None, id: int = None, retry: int = None, custom_fields: Dict[str, str] = None) → None

Send event to the client.

Parameters:

  • data (str) – The data to be sent.

  • event (str) – (Optional) The name of the event.

  • id (int) – (Optional) The event ID.

  • retry (int) – (Optional) The time (in milliseconds) to wait before retrying the event.

  • custom_fields (Dict[str, str]) – (Optional) Custom fields to be sent with the event.

class adafruit_httpserver.response.Websocket(request: Request, headers: Headers | Dict[str, str] = None, cookies: Dict[str, str] = None, buffer_size: int = 1024)

Specialized version of Response class for creating a websocket connection.

Allows two way communication between the client and the server.

Keep in mind, that in order to send and receive messages, the socket must be kept open. This means that you have to store the response object somewhere, so you can send events to it and close it later.

It is very important to close the connection manually, it will not be done automatically.

Example:

ws = None

@server.route(path, method)
def route_func(request: Request):

    # Store the response object somewhere in global scope
    global ws
    ws = Websocket(request)

    return ws

...

# Receive message from client
message = ws.receive()

# Later, when you want to send an event
ws.send_message("Simple message")

# Close the connection
ws.close()
Parameters:

  • request (Request) – Request object

  • headers (Headers) – Headers to be sent with the response.

  • cookies (Dict[str, str]) – Cookies to be sent with the response.

  • buffer_size (int) – Size of the buffer used to send and receive messages.

close()

Close the connection.

Always call this method when you are done sending events.

receive(fail_silently: bool = False) → str | bytes | None

Receive a message from the client.

Parameters:

fail_silently (bool) – If True, no error will be raised if the connection is closed.

send_message(message: str | bytes, opcode: int = None, fail_silently: bool = False)

Send a message to the client.

Parameters:

  • message (str) – Message to be sent.

  • opcode (int) – Opcode of the message. Defaults to TEXT if message is a string and BINARY for bytes.

  • fail_silently (bool) – If True, no error will be raised if the connection is closed.

adafruit_httpserver.headers

  • Author(s): Michał Pokusa

class adafruit_httpserver.headers.Headers(headers: str | Dict[str, str] = None)

A dict-like class for storing HTTP headers.

Allows access to headers using case insensitive names.

Does not implement all dict methods.

Examples:

headers = Headers("Content-Type: text/html\r\nContent-Length: 1024\r\n")
# or
headers = Headers({"Content-Type": "text/html", "Content-Length": "1024"})

len(headers)
# 2

headers.setdefault("Access-Control-Allow-Origin", "*")
headers["Access-Control-Allow-Origin"]
# '*'

headers["Content-Length"]
# '1024'

headers["content-type"]
# 'text/html'

headers["User-Agent"]
# KeyError: User-Agent

"CONTENT-TYPE" in headers
# True
add(field_name: str, value: str)

Adds a header with the given field name and value. Allows adding multiple headers with the same name.

copy()

Returns a copy of the headers.

property fields

Returns a list of field names.

get(field_name: str, default: str = None) → str | None

Returns the value for the given header name, or default if not found.

get_directive(name: str, default: str = None) → str | None

Returns the main value (directive) for the given header name, or default if not found.

Example:

headers = Headers({"Content-Type": "text/html; charset=utf-8"})
headers.get_directive("Content-Type")
# 'text/html'
get_list(field_name: str) → List[str]

Get the list of values of a field.

get_parameter(name: str, parameter: str, default: str = None) → str | None

Returns the value of the given parameter for the given header name, or default if not found.

Example:

headers = Headers({"Content-Type": "text/html; charset=utf-8"})
headers.get_parameter("Content-Type", "charset")
# 'utf-8'
items()

Returns a list of (name, value) tuples.

keys()

Returns a list of header names.

set(name: str, value: str)

Sets the value for the given header name.

setdefault(name: str, default: str = None)

Sets the value for the given header name if it does not exist.

update(headers: Dict[str, str])

Updates the headers with the given dict.

values()

Returns a list of header values.

adafruit_httpserver.status

  • Author(s): Dan Halbert, Michał Pokusa

class adafruit_httpserver.status.Status(code: int, text: str)

HTTP status code.

Define a status code.

Parameters:

  • code (int) – Numeric value: 200, 404, etc.

  • text (str) – Short phrase: “OK”, “Not Found’, etc.

adafruit_httpserver.mime_types

  • Author(s): Michał Pokusa

class adafruit_httpserver.mime_types.MIMETypes

Contains MIME types for common file extensions. Allows to set default type for unknown files, unregister unused types and register new ones using the MIMETypes.configure().

DEFAULT = 'text/plain'

Default MIME type for unknown files. Can be changed using MIMETypes.configure(default_to=...).

classmethod configure(default_to: str = None, keep_for: List[str] = None, register: Dict[str, str] = None) → None

Allows to globally configure the MIME types.

It is recommended to always call this method before starting the Server. Unregistering unused MIME types will decrease overall memory usage.

Parameters:

  • default_to (str) – The MIME type to use for unknown files.

  • keep_for (List[str]) – File extensions to keep. All other will be unregistered.

  • register (Dict[str, str]) – A dictionary mapping file extensions to MIME types.

Example:

MIMETypes.configure(
    default_to="text/plain",
    keep_for=[".jpg", ".mp4", ".txt"],
    register={".foo": "text/foo", ".bar": "text/bar", ".baz": "text/baz"},
)
classmethod get_for_filename(filename: str, default: str = None) → str

Return the MIME type for the given file name. If the file extension is not registered, default is returned.

Parameters:

  • filename (str) – The file name to look up.

  • default (str) – Default MIME type to return if the file extension is not registered.

adafruit_httpserver.exceptions

  • Author(s): Michał Pokusa

exception adafruit_httpserver.exceptions.AuthenticationError

Raised by require_authentication when the Request is not authorized.

exception adafruit_httpserver.exceptions.BackslashInPathError(path: str)

Backslash \ in path.

Creates a new BackslashInPathError for the path.

exception adafruit_httpserver.exceptions.FileNotExistsError(path: str)

Raised when a file does not exist.

Creates a new FileNotExistsError for the file at path.

exception adafruit_httpserver.exceptions.InvalidPathError

Parent class for all path related errors.

exception adafruit_httpserver.exceptions.ParentDirectoryReferenceError(path: str)

Path contains .., a reference to the parent directory.

Creates a new ParentDirectoryReferenceError for the path.

exception adafruit_httpserver.exceptions.ServerStoppedError

Raised when .poll is called on a stopped Server.

exception adafruit_httpserver.exceptions.ServingFilesDisabledError

Raised when root_path is not set and there is no handler for request.