httk.serve.http

Public lightweight HTTP application helpers.

Submodules

• httk.serve.http.accept
• httk.serve.http.api
• httk.serve.http.apptypes
• httk.serve.http.fields
• httk.serve.http.identifiers
• httk.serve.http.openapi
• httk.serve.http.webhook

Attributes

JsonDocument
JsonDocumentFactory
ResponseHook	A zero-argument coroutine callback run once, after the response has been sent.
ServeApp	The httk-serve serving application type (a Starlette app; branded so consumers need not import Starlette).
WebhookSender

Exceptions

WebhookTransportError

Represent an outbound webhook transport or URL-policy failure.

Classes

MediaRange	One valid media range parsed from an HTTP Accept header.
PinnedHttpsJsonPoster	Send one HTTPS JSON POST through a validated, pinned socket.

Functions

best_quality_ignoring_parameterised(ranges, major, minor)	Select the best range, discarding any range carrying a parameter.
best_quality_matching_parameters(ranges, major, minor, ...)	Select the best range that keeps and matches its parameters.
http_parameter_value(value)	Decode one token or quoted-string parameter value.
parse_accept(header)	Parse an Accept header into its valid media ranges.
parse_media_type(value)	Parse a JSON media type into its major, minor, and parameters.
split_http_list(value, delimiter)	Split an HTTP field outside quoted strings, rejecting broken quoting.
create_file_map_app(files, *[, debug])	Create an application exposing only explicitly mapped files.
json_get_app(document, *[, path, media_type, profile, ...])	Create a lightweight application serving one live JSON document.
jsonld_get_app(document, *[, path, media_type, ...])	Create a lightweight application serving one live JSON-LD document.
is_field_name(value)	Report whether a value is a valid RFC 9110 field name.
is_field_value(value)	Report whether a value is an acceptable HTTP field value.
validated_headers(pairs)	Validate response header name-value pairs, rejecting duplicate names.
is_json_encodable_text(value)	Report whether a string is free of lone UTF-16 surrogate code points.
urn_uuid(value)	Normalise a UUID factory result to the required urn:uuid: URN form.
xsd_utc_timestamp(value)	Render a timezone-aware datetime as an XSD/RFC 3339 UTC timestamp.
deliver_with_retries(sender, url, document, *[, attempts])	Deliver a document through a sender at most attempts times, requiring a 2xx status.
join_url_path(base_url, path)	Append a fixed absolute path without depending on a trailing slash.

Package Contents

class httk.serve.http.MediaRange

One valid media range parsed from an HTTP Accept header.

Parameters:

• major – Lowercase major type, or "*" for a wildcard.

• minor – Lowercase minor type, or "*" for a wildcard.

• parameters – Non-q parameters as immutable lowercase name-value pairs.

• quality – The q weight, defaulting to 1.0 when absent.

major: str

minor: str

parameters: tuple[tuple[str, str], Ellipsis]

quality: float

httk.serve.http.best_quality_ignoring_parameterised(ranges, major, minor)

Select the best range, discarding any range carrying a parameter.

Any range that carries a parameter is discarded outright. The specificity tiebreak favours a more specific major/minor match, with a constant second component.

Parameters:

• ranges (tuple[MediaRange, Ellipsis]) – The parsed media ranges to select among.

• major (str) – The response major type to match.

• minor (str) – The response minor type to match.

Returns:

The winning range’s quality, or None when nothing matches.

Return type:

float | None

httk.serve.http.best_quality_matching_parameters(ranges, major, minor, parameters)

Select the best range that keeps and matches its parameters.

A parameterised range is kept only when every one of its parameters is present and equal in parameters. The specificity tiebreak favours a more specific major/minor match and then a range carrying more parameters.

Parameters:

• ranges (tuple[MediaRange, Ellipsis]) – The parsed media ranges to select among.

• major (str) – The response major type to match.

• minor (str) – The response minor type to match.

• parameters (collections.abc.Mapping[str, str]) – The response media-type parameters to match against.

Returns:

The winning range’s quality, or None when nothing matches.

Return type:

float | None

httk.serve.http.http_parameter_value(value)

Decode one token or quoted-string parameter value.

Parameters:

value (str) – The raw parameter value, possibly a quoted string.

Returns:

The decoded value, or None when the value is malformed.

Return type:

str | None

httk.serve.http.parse_accept(header)

Parse an Accept header into its valid media ranges.

Parameters:

header (str) – The raw Accept header value.

Returns:

The valid parsed media ranges, or () when the header cannot be split.

Return type:

tuple[MediaRange, Ellipsis]

httk.serve.http.parse_media_type(value)

Parse a JSON media type into its major, minor, and parameters.

Parameters:

value (object) – The declared media type; only application/json and application/*+json types with unique parameters are accepted.

Returns:

The lowercase major type, lowercase minor type, and parameter mapping.

Return type:

tuple[str, str, dict[str, str]]

httk.serve.http.split_http_list(value, delimiter)

Split an HTTP field outside quoted strings, rejecting broken quoting.

Parameters:

• value (str) – The raw HTTP field value to split.

• delimiter (str) – The single-character separator to split on outside quotes.

Returns:

The delimited parts, or None when a quoted string is unterminated.

Return type:

tuple[str, Ellipsis] | None

type httk.serve.http.JsonDocument = Mapping[str, object]

type httk.serve.http.JsonDocumentFactory = Callable[[], JsonDocument | Awaitable[JsonDocument]]

httk.serve.http.create_file_map_app(files, *, debug=False)

Create an application exposing only explicitly mapped files.

Parameters:

• files (collections.abc.Mapping[str, str | pathlib.Path]) – Root-relative URL paths mapped to filesystem paths.

• debug (bool) – Whether Starlette debug responses are enabled.

Returns:

A mountable application serving the declared files.

Return type:

httk.serve.http.apptypes.ServeApp

httk.serve.http.json_get_app(document, *, path='/', media_type='application/json', profile=None, cache_control='public, max-age=60', cors_allow_origin='*', debug=False)

Create a lightweight application serving one live JSON document.

Parameters:

• document (JsonDocument | JsonDocumentFactory) – JSON mapping or live zero-argument document factory.

• path (str) – Canonical root-relative route, normally / for mounting.

• media_type (str) – JSON response media type, including optional parameters.

• profile (str | None) – Optional profile IRI emitted as an RFC 6906 Link header.

• cache_control (str | None) – Optional Cache-Control response value.

• cors_allow_origin (str | None) – Optional Access-Control-Allow-Origin value.

• debug (bool) – Whether Starlette debug responses are enabled.

Returns:

A mountable application serving the declared JSON resource.

Return type:

httk.serve.http.apptypes.ServeApp

httk.serve.http.jsonld_get_app(document, *, path='/', media_type='application/ld+json', profile=None, cache_control='public, max-age=60', cors_allow_origin='*', debug=False)

Create a lightweight application serving one live JSON-LD document.

Parameters:

• document (JsonDocument | JsonDocumentFactory) – JSON-LD mapping or live zero-argument document factory.

• path (str) – Canonical root-relative route, normally / for mounting.

• media_type (str) – JSON-LD response media type, including optional parameters.

• profile (str | None) – Optional profile IRI emitted as an RFC 6906 Link header.

• cache_control (str | None) – Optional Cache-Control response value.

• cors_allow_origin (str | None) – Optional Access-Control-Allow-Origin value.

• debug (bool) – Whether Starlette debug responses are enabled.

Returns:

A mountable application serving the declared JSON-LD resource.

Return type:

httk.serve.http.apptypes.ServeApp

type httk.serve.http.ResponseHook = Callable[[], Awaitable[None]]

A zero-argument coroutine callback run once, after the response has been sent.

type httk.serve.http.ServeApp = Starlette

The httk-serve serving application type (a Starlette app; branded so consumers need not import Starlette).

Annotation-only: this is a type alias, not the Starlette class, so use Starlette itself for isinstance/subclass checks.

httk.serve.http.is_field_name(value)

Report whether a value is a valid RFC 9110 field name.

Parameters:

value (str) – The candidate field name.

Returns:

True when the value is a string of RFC 9110 tchar characters.

Return type:

bool

httk.serve.http.is_field_value(value)

Report whether a value is an acceptable HTTP field value.

Parameters:

value (str) – The candidate field value.

Returns:

True when the value is a non-empty string free of CR and LF.

Return type:

bool

httk.serve.http.validated_headers(pairs)

Validate response header name-value pairs, rejecting duplicate names.

Names must be RFC 9110 field names and values must be non-empty and free of CR and LF; header names are compared case-insensitively for uniqueness.

Parameters:

pairs (collections.abc.Iterable[tuple[str, str]]) – The header name-value pairs to validate.

Returns:

The validated pairs as an immutable tuple.

Return type:

tuple[tuple[str, str], Ellipsis]

httk.serve.http.is_json_encodable_text(value)

Report whether a string is free of lone UTF-16 surrogate code points.

Parameters:

value (str) – String to test for lone surrogate code points.

Returns:

False when the string contains a lone surrogate, otherwise True.

Return type:

bool

httk.serve.http.urn_uuid(value)

Normalise a UUID factory result to the required urn:uuid: URN form.

Parameters:

value (object) – UUID factory output rendered to text before normalisation.

Returns:

The identifier in urn:uuid: form.

Raises:

RuntimeError – If the rendered value is blank or contains a lone surrogate.

Return type:

str

httk.serve.http.xsd_utc_timestamp(value)

Render a timezone-aware datetime as an XSD/RFC 3339 UTC timestamp.

Parameters:

value (datetime.datetime) – Timezone-aware datetime to render as a UTC XML Schema date-time.

Returns:

The value in UTC with a Z suffix.

Raises:

• TypeError – If value is not a datetime.

• ValueError – If value is naive or lacks a UTC offset.

Return type:

str

class httk.serve.http.PinnedHttpsJsonPoster(*, connect_timeout=5.0, read_timeout=10.0, dns_timeout=5.0, total_timeout=20.0, response_body_limit=65536, max_concurrency=8, allow_private_addresses=False, resolver=None, connector=None)

Send one HTTPS JSON POST through a validated, pinned socket.

DNS resolution occurs inside the concurrency cap and produces the exact numeric socket address dialed for the request. TLS SNI and the HTTP Host header retain the original DNS hostname. The stdlib transport does not consult proxy or environment settings; redirects are not implemented. Both DNS and the complete operation have finite deadlines.

Parameters:

• connect_timeout (float) – Maximum seconds for TCP connect and TLS handshake.

• read_timeout (float) – Maximum seconds for reading a complete response body.

• dns_timeout (float) – Maximum seconds for DNS resolution.

• total_timeout (float) – Maximum seconds from the start of resolution through response completion.

• response_body_limit (int) – Maximum response bytes consumed before rejection.

• max_concurrency (int) – Maximum simultaneous resolution and send operations.

• allow_private_addresses (bool) – Skip global-address filtering when True; the default rejects private targets.

• resolver (_Resolver | None) – Optional test seam that resolves public pinned addresses.

• connector (_Connector | None) – Optional test seam that dials a supplied pinned address.

type httk.serve.http.WebhookSender = Callable[[str, dict[str, JsonValue]], Awaitable[int]]

exception httk.serve.http.WebhookTransportError(detail)

Bases: RuntimeError

Represent an outbound webhook transport or URL-policy failure.

Parameters:

detail (str) – Safe detail suitable for local process-delivery state.

detail

async httk.serve.http.deliver_with_retries(sender, url, document, *, attempts=2)

Deliver a document through a sender at most attempts times, requiring a 2xx status.

Parameters:

• sender (WebhookSender) – Awaitable sender returning a peer HTTP status code.

• url (str) – Complete webhook URL passed to the sender.

• document (dict[str, httk.serve.jsondata.JsonValue]) – Plain JSON message document passed to the sender.

• attempts (int) – Maximum number of delivery attempts before failing.

Raises:

WebhookTransportError – If no attempt yields a 2xx acknowledgement.

httk.serve.http.join_url_path(base_url, path)

Append a fixed absolute path without depending on a trailing slash.

Parameters:

• base_url (str) – Base URL from an inbound request.

• path (str) – Absolute path beginning with /.

Returns:

Complete URL.

Raises:

ValueError – If the base or fixed path is malformed.

Return type:

str