qfa.domain.ports#

Port interfaces (protocols) for the feedback analysis backend.

Driven ports declared here use typing.Protocol for structural subtyping per ADR-002. The orchestrator is exposed as the concrete Orchestrator class per ADR-011 (no driving port).

Classes

AnonymizationPort(*args, **kwargs)

Port for anonymising and de-anonymising user-supplied text.

LLMPort(*args, **kwargs)

Port for interacting with a large-language-model provider.

UsageRepositoryPort(*args, **kwargs)

Port for recording and querying LLM usage data.
class qfa.domain.ports.LLMPort(*args, **kwargs)[source]#

Bases: Protocol

Port for interacting with a large-language-model provider.

Implementations must translate provider-specific details into the domain LLMResponse model.

async complete(system_message: str, user_message: str, tenant_id: str, response_model: type[T_Response], timeout: float = 20.0) → LLMResponse[source]#

Send a completion request to the LLM provider.

Parameters:

  • system_message (str) – The system-level instruction for the model.

  • user_message (str) – The user-level message to complete.

  • tenant_id (str) – Tenant identifier for tracking and billing.

  • response_model (type[T_Response]) – The Pydantic model to parse the response into.

  • timeout (float) – Maximum time in seconds to wait for a response.

Returns:

The model’s response including token usage.

Return type:

LLMResponse

class qfa.domain.ports.UsageRepositoryPort(*args, **kwargs)[source]#

Bases: Protocol

Port for recording and querying LLM usage data.

async record_call(record: LLMCallRecord) → None[source]#

Record a single LLM call attempt.

Parameters:

record (LLMCallRecord) – The call record to persist.

async get_usage_stats(tenant_id: str, from_: datetime | None = None, to: datetime | None = None) → UsageStats[source]#

Get aggregated usage stats for a single tenant.

Parameters:

  • tenant_id (str) – The tenant to query.

  • from (datetime | None) – Inclusive lower bound (UTC tz-aware), or None.

  • to (datetime | None) – Exclusive upper bound (UTC tz-aware), or None.

Returns:

Stats for the tenant, or None if no calls in window.

Return type:

UsageStats | None

async get_all_usage_stats(from_: datetime | None = None, to: datetime | None = None) → list[UsageStats][source]#

Get per-tenant stats plus a grand total entry (tenant_id=None).

Parameters:

  • from (datetime | None) – Inclusive lower bound (UTC tz-aware), or None.

  • to (datetime | None) – Exclusive upper bound (UTC tz-aware), or None.

Returns:

Per-tenant stats followed by a grand total entry.

Return type:

list[UsageStats]

class qfa.domain.ports.AnonymizationPort(*args, **kwargs)[source]#

Bases: Protocol

Port for anonymising and de-anonymising user-supplied text.

Implementations replace named entities (people, locations, phone numbers, etc.) in text with stable placeholders, returning the redacted text together with a mapping that can be used to restore the original values via deanonymize.

Implementations must be deterministic for a given input within a single call (same entity replaced by the same placeholder).

anonymize(text: str) → tuple[str, dict[str, str]][source]#

Replace sensitive entities in text with placeholders.

Parameters:

text (str) – The text to anonymise.

Returns:

The anonymised text and a mapping from placeholder to original value, suitable for passing to deanonymize.

Return type:

tuple[str, dict[str, str]]

deanonymize(text: str, mapping: dict[str, str]) → str[source]#

Restore original values in text using mapping.

Parameters:

  • text (str) – The anonymised text, possibly containing placeholders.

  • mapping (dict[str, str]) – Placeholder-to-original mapping returned by anonymize.

Returns:

The text with placeholders replaced by original values.

Return type:

str