httk.serve.dsp.provider¶

In-memory business implementation of the constrained DSP provider.

Attributes¶

`UuidFactory`
`UtcClock`

Classes¶

DspProvider

Serve a live dataset catalogue and manage non-durable DSP processes.

Module Contents¶

type httk.serve.dsp.provider.UuidFactory = Callable[[], UUID | str][source]¶

type httk.serve.dsp.provider.UtcClock = Callable[[], datetime][source]¶

class httk.serve.dsp.provider.DspProvider(config, *, store=None, publications=None, catalogue_policy=None, callback_sender=None, uuid_factory=uuid4, utc_clock=None)[source]¶

Serve a live dataset catalogue and manage non-durable DSP processes.

Business methods accept and return only ordinary JSON dictionaries; a thin HTTP adapter is responsible for route and response-code presentation. All process state is in memory and is lost on restart. Callback transitions are never committed until a peer acknowledges a 2xx response.

Parameters:

config (httk.serve.dsp.config.DspProviderConfig) – Validated fixed provider configuration.
store (httk.store.EntryStore | None) – Caller-owned entry store containing the DSP publication family. Exactly one of store and publications is required.
publications (collections.abc.Iterable[httk.serve.dsp.config.DspPublicationRecord] | None) – Inline dataset and service publication envelopes. Exactly one of publications and store is required.
catalogue_policy (httk.serve.dsp.catalogue.DspCataloguePolicy | None) – Optional replaceable catalogue requirements and serialization policy. The built-in minimal policy is used by default.
callback_sender (httk.serve.dsp.callbacks.CallbackSender | None) – Optional asynchronous callback transport. Supplying one bypasses default network policy and is useful for deterministic tests.
uuid_factory (UuidFactory) – Optional source for provider and agreement identifiers.
utc_clock (UtcClock | None) – Optional UTC clock used for agreement timestamps.

config[source]¶

catalogue_policy[source]¶

property profile: httk.serve.dsp.models.CatalogueProfile[source]¶

Return a freshly validated catalogue snapshot.

automatic_batch()[source]¶

Create a response-local holder for automatic callback actions.

The HTTP adapter uses the returned private holder to release callbacks only from that response’s background hook. Ordinary callers do not need this seam: their automatic callbacks are managed immediately after the business method returns.

Returns:: An empty response-local automatic callback holder.
Return type:: _AutomaticBatch

has_automatic_actions(batch)[source]¶

Report whether a response-local holder has callbacks to release.

Parameters:: batch (_AutomaticBatch) – Holder returned by automatic_batch().
Returns:: Whether the holder contains at least one action.
Return type:: bool

async release_automatic(batch)[source]¶

Start one response’s automatic callbacks after its body was sent.

Parameters:: batch (_AutomaticBatch) – Holder returned by automatic_batch().

async drain_automatic()[source]¶

Wait until all provider-managed automatic callback tasks settle.

async cancel_automatic()[source]¶

Cancel and drain provider-managed automatic callbacks at shutdown.

version_document()[source]¶

Return the DSP 2025-1 HTTPS version-discovery document.

Returns:: Plain DSP protocol-version document.
Return type:: dict[str, httk.serve.dsp.models.JsonValue]

dsp_catalogue(request)[source]¶

Return the DSP catalogue snapshot for an empty catalogue filter.

Parameters:: request (dict[str, object]) – Catalog request message JSON.
Returns:: Plain DSP catalogue document.
Raises:: httk.serve.dsp.models.DspProtocolError – If the request is malformed or filters are unsupported.
Return type:: dict[str, httk.serve.dsp.models.JsonValue]

catalogue(request, representation)[source]¶

Return a catalogue in a representation selected by the policy.

Parameters:

request (dict[str, object]) – Catalog request message JSON.
representation (httk.serve.dsp.catalogue.DspCatalogueRepresentation) – Value returned by select_catalogue_representation().

Returns:

Plain catalogue document.

Return type:

dict[str, httk.serve.dsp.models.JsonValue]

select_catalogue_representation(accept)[source]¶

Select a catalogue response representation through the active policy.

Parameters:: accept (str | None) – Raw HTTP Accept field, or None when absent.
Returns:: Selected representation metadata.
Return type:: httk.serve.dsp.catalogue.DspCatalogueRepresentation

validate_catalogue_request(request)[source]¶

Validate the one unfiltered catalogue request supported by DSP minimal.

dsp_dataset(dataset_id)[source]¶

Return one DSP dataset only when its ID exactly matches.

Parameters:: dataset_id (str) – Requested dataset identifier.
Returns:: Plain DSP dataset document.
Raises:: httk.serve.dsp.models.DspProtocolError – If the identifier is absent or unknown.
Return type:: dict[str, httk.serve.dsp.models.JsonValue]

dcat_catalogue()[source]¶

Return the separate strict owned-context DCAT-AP projection.

Returns:: Plain DCAT-AP-compatible JSON-LD catalogue document.
Return type:: dict[str, httk.serve.dsp.models.JsonValue]

async get_negotiation(provider_pid)[source]¶

Return one acknowledged negotiation process.

Parameters:: provider_pid (str) – Provider negotiation process identifier.
Returns:: Plain DSP negotiation document.
Raises:: httk.serve.dsp.models.DspProtocolError – If the process is unknown.
Return type:: dict[str, httk.serve.dsp.models.JsonValue]

async request_negotiation(message, *, _automatic_batch=None)[source]¶

Accept an initial consumer contract request.

The initial request must omit providerPid, identify the configured offer and dataset exactly, and provide an HTTPS callback. With automatic progression enabled, the agreement callback is scheduled only after the returned process snapshot has been acknowledged to the caller.

Parameters:

message (dict[str, object]) – Contract request message JSON.
_automatic_batch (_AutomaticBatch | None) – Optional response-local holder used by the HTTP adapter.

Returns:

Newly created DSP negotiation document.

Raises:

httk.serve.dsp.models.DspProtocolError – If message validation fails.

Return type:

dict[str, httk.serve.dsp.models.JsonValue]

async counter_request(provider_pid, message)[source]¶

Receive a consumer counter-request after a provider offer.

Parameters:

provider_pid (str) – Provider negotiation process identifier from the route.
message (dict[str, object]) – Consumer contract request message JSON.

Raises:

httk.serve.dsp.models.DspProtocolError – If PIDs, policy, or the transition are invalid.

async negotiation_event(provider_pid, message)[source]¶

Receive the only permitted consumer negotiation event, ACCEPTED.

Parameters:

provider_pid (str) – Provider negotiation process identifier from the route.
message (dict[str, object]) – Contract-negotiation event message JSON.

Raises:

httk.serve.dsp.models.DspProtocolError – If PIDs, event, or transition are invalid.

async verify_agreement(provider_pid, message, *, _automatic_batch=None)[source]¶

Receive consumer verification of an acknowledged agreement.

Parameters:

provider_pid (str) – Provider negotiation process identifier from the route.
message (dict[str, object]) – Agreement-verification message JSON.
_automatic_batch (_AutomaticBatch | None) – Optional response-local holder used by the HTTP adapter.

Raises:

httk.serve.dsp.models.DspProtocolError – If PIDs, state, or finalization delivery are invalid.

async receive_negotiation_termination(provider_pid, message)[source]¶

Receive consumer termination of a nonterminal negotiation.

Parameters:

provider_pid (str) – Provider negotiation process identifier from the route.
message (dict[str, object]) – Negotiation-termination message JSON.

Raises:

httk.serve.dsp.models.DspProtocolError – If PIDs or the transition are invalid.

async get_transfer(provider_pid)[source]¶

Return one acknowledged transfer process.

Parameters:: provider_pid (str) – Provider transfer-process identifier.
Returns:: Plain DSP transfer-process document.
Raises:: httk.serve.dsp.models.DspProtocolError – If the process is unknown.
Return type:: dict[str, httk.serve.dsp.models.JsonValue]

async request_transfer(message, *, _automatic_batch=None)[source]¶

Accept a consumer pull transfer request under a finalized agreement.

Identical repeated consumer process IDs return the original transfer. A reuse with different agreement, callback, or format is rejected.

Parameters:

message (dict[str, object]) – Transfer request message JSON.
_automatic_batch (_AutomaticBatch | None) – Optional response-local holder used by the HTTP adapter.

Returns:

Newly created or idempotently recovered transfer-process document.

Raises:

httk.serve.dsp.models.DspProtocolError – If request validation or callback delivery fails.

Return type:

dict[str, httk.serve.dsp.models.JsonValue]

async resume_transfer(provider_pid, message)[source]¶

Receive a consumer start message that resumes a suspended transfer.

Parameters:

provider_pid (str) – Provider transfer-process identifier from the route.
message (dict[str, object]) – Transfer-start message JSON.

Raises:

httk.serve.dsp.models.DspProtocolError – If PIDs or the transition are invalid.

async receive_transfer_suspension(provider_pid, message)[source]¶

Receive consumer suspension of a started transfer.

Parameters:

provider_pid (str) – Provider transfer-process identifier from the route.
message (dict[str, object]) – Transfer-suspension message JSON.

Raises:

httk.serve.dsp.models.DspProtocolError – If PIDs or the transition are invalid.

async receive_transfer_completion(provider_pid, message)[source]¶

Receive consumer completion of a started transfer.

Parameters:

provider_pid (str) – Provider transfer-process identifier from the route.
message (dict[str, object]) – Transfer-completion message JSON.

Raises:

httk.serve.dsp.models.DspProtocolError – If PIDs or the transition are invalid.

async receive_transfer_termination(provider_pid, message)[source]¶

Receive consumer termination of a nonterminal transfer.

Parameters:

provider_pid (str) – Provider transfer-process identifier from the route.
message (dict[str, object]) – Transfer-termination message JSON.

Raises:

httk.serve.dsp.models.DspProtocolError – If PIDs or the transition are invalid.

async send_offer(provider_pid)[source]¶

Send a provider contract offer and acknowledge REQUESTED to OFFERED.

Parameters:

provider_pid (str) – Provider negotiation process identifier.

Raises:

httk.serve.dsp.models.DspProtocolError – If state or callback delivery is invalid.
httk.serve.dsp.models.DspTransitionSuperseded – If a concurrent transition won the callback commit.

async send_agreement(provider_pid)[source]¶

Send a provider agreement from REQUESTED or ACCEPTED.

Parameters:

provider_pid (str) – Provider negotiation process identifier.

Raises:

httk.serve.dsp.models.DspProtocolError – If state or callback delivery is invalid.
httk.serve.dsp.models.DspTransitionSuperseded – If a concurrent transition won the callback commit.

async finalize_negotiation(provider_pid)[source]¶

Send provider finalization after consumer agreement verification.

Parameters:

provider_pid (str) – Provider negotiation process identifier.

Raises:

httk.serve.dsp.models.DspProtocolError – If state or callback delivery is invalid.
httk.serve.dsp.models.DspTransitionSuperseded – If a concurrent transition won the callback commit.

async terminate_negotiation(provider_pid, *, code='terminated', reason='negotiation terminated by provider')[source]¶

Send provider termination for any nonterminal negotiation state.

Parameters:

provider_pid (str) – Provider negotiation process identifier.
code (str) – Machine-readable termination code.
reason (str) – Human-readable termination reason.

Raises:

httk.serve.dsp.models.DspProtocolError – If state or callback delivery is invalid.
httk.serve.dsp.models.DspTransitionSuperseded – If a concurrent transition won the callback commit.

async start_transfer(provider_pid)[source]¶

Send provider transfer start with the configured pull data address.

Parameters:

provider_pid (str) – Provider transfer-process identifier.

Raises:

httk.serve.dsp.models.DspProtocolError – If state or callback delivery is invalid.
httk.serve.dsp.models.DspTransitionSuperseded – If a concurrent transition won the callback commit.

async suspend_transfer(provider_pid, *, code='suspended', reason='transfer suspended by provider')[source]¶

Send provider suspension for a started transfer.

Parameters:

provider_pid (str) – Provider transfer-process identifier.
code (str) – Machine-readable suspension code.
reason (str) – Human-readable suspension reason.

Raises:

httk.serve.dsp.models.DspProtocolError – If state or callback delivery is invalid.
httk.serve.dsp.models.DspTransitionSuperseded – If a concurrent transition won the callback commit.

async complete_transfer(provider_pid)[source]¶

Send provider completion for a started transfer.

Parameters:

provider_pid (str) – Provider transfer-process identifier.

Raises:

httk.serve.dsp.models.DspProtocolError – If state or callback delivery is invalid.
httk.serve.dsp.models.DspTransitionSuperseded – If a concurrent transition won the callback commit.

async terminate_transfer(provider_pid, *, code='terminated', reason='transfer terminated by provider')[source]¶

Send provider termination for any nonterminal transfer state.

Parameters:

provider_pid (str) – Provider transfer-process identifier.
code (str) – Machine-readable termination code.
reason (str) – Human-readable termination reason.

Raises:

httk.serve.dsp.models.DspProtocolError – If state or callback delivery is invalid.
httk.serve.dsp.models.DspTransitionSuperseded – If a concurrent transition won the callback commit.