qfa.domain.ports#

Port interfaces (protocols) for the feedback analysis backend.

Driven ports declared here use typing.Protocol for structural subtyping per ADR-002. Each application service in qfa.services is exposed as its own concrete class per ADR-011 and ADR-017 (no driving port).

Classes

AnonymizationPort(*args, **kwargs)

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

AuthLookupPort(*args, **kwargs)

Port for authenticating users of the application.

AuthManagementPort(*args, **kwargs)

Port for adding/ removing keys and tenants from the application.

EmbeddingPort(*args, **kwargs)

Port for a multilingual text-embedding model.

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.EmbeddingPort(*args, **kwargs)[source]#

Bases: Protocol

Port for a multilingual text-embedding model.

Implementations MUST be multilingual: community feedback is multilingual, and a monolingual model would cluster by language instead of theme. embed returns one dense vector per input text, in input order.

Synchronous by design: encoding is CPU-bound local computation, not an I/O call (contrast LLMPort.complete()). If a future externalised adapter makes embedding I/O-bound, an async variant can be added then.

embed(texts: tuple[str, ...]) → tuple[tuple[float, ...], ...][source]#

Return one dense embedding vector per input text, in order.

Parameters:

texts (tuple[str, ...]) – The texts to embed. May be empty.

Returns:

One vector (tuple of floats) per input text, same length and order as texts. Every vector has the same dimensionality.

Return type:

tuple[tuple[float, …], …]

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_for_one_tenant(tenant_id: str, from_: datetime | None = None, to: datetime | None = None) → TenantUsageStats[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. When no calls match the window, a zero-valued TenantUsageStats is returned (never None).

Return type:

TenantUsageStats

async get_all_usage_by_tenant(from_: datetime | None = None, to: datetime | None = None) → list[TenantUsageStats][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[TenantUsageStats]

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

Get per-operation stats with nested per-tenant breakdown plus grand total.

Inverse hierarchy of get_all_usage_by_tenant(): top-level aggregation is by orchestrator operation; each operation block carries a tuple of per-tenant blocks. The grand-total entry (operation=None) is always emitted last, matching the convention used by get_all_usage_by_tenant().

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-operation stats followed by a grand total entry.

Return type:

list[OperationUsageStats]

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

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

Bases: Protocol

Port for authenticating users of the application.

async validate_api_key(provided_key: str) → TenantApiKey | None[source]#

Validate if a key exists in the implemented adapter.

Parameters:

provided_key (str) – The API key value supplied by the caller.

Returns:

The matching tenant API key, or None if no match was found.

Return type:

TenantApiKey | None

async get_auth_keys(tenant_id: str | None = None) → list[AuthKeyInfo][source]#

Get all API keys for a tenant, or all keys if tenant_id is None.

Parameters:

tenant_id (str | None) – The tenant to query, or None to get keys for all tenants.

Returns:

A list of AuthKeyMetadata objects

Return type:

list[AuthKeyMetadata]

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

Bases: Protocol

Port for adding/ removing keys and tenants from the application.

async add_tenant(tenant_name: str, allows_superusers: bool = False) → str[source]#

Add a new tenant to the implemented adapter and return its unique identifier.

Parameters:

  • tenant_name (str) – The name of the tenant to create.

  • allows_superusers (bool) – Whether this tenant allows creation of superuser keys (default False).

Returns:

The unique identifier of the created tenant.

Return type:

str

async delete_tenant(tenant_id: str) → None[source]#

Delete an existing tenant from the implemented adapter.

Parameters:

tenant_id (str) – The unique identifier of the tenant to delete.

Raises:

TenantNotFoundError: – If no tenant with this tenant_id exists

async add_key(key_name: str, tenant_id: str, is_superuser: bool = False) → KeyCreationResponse[source]#

Generate and persist a new API key in the implemented adapter.

Parameters:

  • key_name (str) – A human-friendly name for the key.

  • tenant_id (str) – The tenant this key belongs to.

  • is_superuser (bool) – Whether this key should have superuser privileges (default False).

Returns:

The created key identifier and plaintext API key as (key_id, api_key). The plaintext key is returned only once at creation time.

Return type:

tuple[str, str]

Raises:

  • KeyAlreadyExistsError: – If key with this key_id already exists

  • TenantDoesNotAllowSuperUsersError: – If the tenant does not allow superuser keys and is_superuser is True

async delete_key(key_id: str) → None[source]#

Delete an existing API key from the implemented adapter.

Parameters:

key_id (str) – The unique identifier of the API key record to remove.

async get_tenants() → list[TenantInfo][source]#

Return metadata for all tenants in the implemented adapter.

Returns:

A list of TenantInfo objects with tenant metadata.

Return type:

list[TenantInfo]