qfa.domain.models#

Domain models for the feedback analysis backend.

All models are immutable (frozen) Pydantic models per ADR-001.

Classes

`AggregateSummaryResultModel`(*, ids, title, ...)	The result of summarizing multiple feedback records as a single aggregate.
`AnalysisRequestModel`(*, feedback_records, ...)	A request to analyze one or more feedback records.
`AnalysisResultModel`(*, result)	The result of a feedback analysis.
`AssignedCodeModel`(*, code_id, code_label, ...)	A single leaf code assigned to a feedback record.
`CallContext`(*, tenant_id, operation)	Per-call context propagated via ContextVar from orchestrator to tracker.
`CallStatus`(*values)	Outcome of a single LLM call attempt.
`CodedFeedbackRecordModel`(*, ...)	Coding output for one feedback record.
`CodingAssignmentRequestModel`(*, ...[, ...])	A request to assign hierarchical codes to feedback records.
`CodingAssignmentResultModel`(*, ...)	The result of assigning codes to multiple feedback records.
`DistributionStats`(*, avg, min, max, p5, p95)	Statistical distribution summary.
`FeedbackRecordModel`(*, id, text, ...)	A single feedback record submitted for analysis.
`FeedbackRecordSummaryModel`(*, id, title, ...)	Summary output for a single feedback record.
`LLMCallRecord`(*, tenant_id, operation, ...)	A single recorded LLM call attempt for usage and cost tracking.
`LLMResponse`(*, structured, model, ...)	Raw response from an LLM provider.
`Operation`(*values)	Orchestrator operations that produce LLM calls.
`SummaryRequestModel`(*, feedback_records[, ...])	A request to summarize one or more feedback records individually.
`SummaryResultModel`(*, feedback_record_summaries)	The result of summarizing multiple feedback records individually.
`TenantApiKey`(*, key_id, name, key, tenant_id)	An API key associated with a tenant.
`TokenStats`(*, avg, min, max, p5, p95, total)	Token distribution summary with a total count.
`UsageStats`(*[, tenant_id, failed_calls, ...])	Aggregated usage statistics for a tenant or grand total.

class qfa.domain.models.FeedbackRecordModel(*, id: str, text: ~typing.Annotated[str, ~annotated_types.MinLen(min_length=1), ~annotated_types.MaxLen(max_length=100000)], metadata: dict[str, str | int | float | bool] = <factory>)[source]#

Bases: BaseModel

A single feedback record submitted for analysis.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config = {'frozen': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

id: str#

text: str#

metadata: dict[str, str | int | float | bool]#

class qfa.domain.models.AnalysisRequestModel(*, feedback_records: Annotated[tuple[FeedbackRecordModel, ...], MinLen(min_length=1)], prompt: Annotated[str, MinLen(min_length=1), MaxLen(max_length=4000)], tenant_id: str)[source]#

Bases: BaseModel

A request to analyze one or more feedback records.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config = {'frozen': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

feedback_records: tuple[FeedbackRecordModel, ...]#

prompt: str#

tenant_id: str#

class qfa.domain.models.AnalysisResultModel(*, result: str)[source]#

Bases: BaseModel

The result of a feedback analysis.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config = {'frozen': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

result: str#

class qfa.domain.models.SummaryRequestModel(*, feedback_records: Annotated[tuple[FeedbackRecordModel, ...], MinLen(min_length=1)], output_language: str | None = None, prompt: Annotated[str | None, MaxLen(max_length=4000)] = None, tenant_id: str)[source]#

Bases: BaseModel

A request to summarize one or more feedback records individually.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config = {'frozen': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

feedback_records: tuple[FeedbackRecordModel, ...]#

output_language: str | None#

prompt: str | None#

tenant_id: str#

class qfa.domain.models.FeedbackRecordSummaryModel(*, id: str, title: str, summary: str, quality_score: float)[source]#

Bases: BaseModel

Summary output for a single feedback record.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config = {'frozen': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

id: str#

title: str#

summary: str#

quality_score: float#

class qfa.domain.models.SummaryResultModel(*, feedback_record_summaries: tuple[FeedbackRecordSummaryModel, ...])[source]#

Bases: BaseModel

The result of summarizing multiple feedback records individually.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config = {'frozen': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

feedback_record_summaries: tuple[FeedbackRecordSummaryModel, ...]#

class qfa.domain.models.AggregateSummaryResultModel(*, ids: tuple[str, ...], title: str, summary: str, quality_score: float)[source]#

Bases: BaseModel

The result of summarizing multiple feedback records as a single aggregate.

# TODO come up with nice solution for non-mutable quality-score, so this can be a frozen class.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

ids: tuple[str, ...]#

title: str#

summary: str#

quality_score: float#

model_config = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class qfa.domain.models.CodingAssignmentRequestModel(*, feedback_records: Annotated[tuple[FeedbackRecordModel, ...], MinLen(min_length=1)], coding_framework: dict[str, Any], max_codes: Annotated[int, Ge(ge=1), Le(le=50)], confidence_threshold: Annotated[float | None, Ge(ge=0.0), Le(le=1.0)] = None, tenant_id: str)[source]#

Bases: BaseModel

A request to assign hierarchical codes to feedback records.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config = {'frozen': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

feedback_records: tuple[FeedbackRecordModel, ...]#

coding_framework: dict[str, Any]#

max_codes: int#

confidence_threshold: float | None#

tenant_id: str#

class qfa.domain.models.AssignedCodeModel(*, code_id: str, code_label: str, confidence_type: float, confidence_category: float, confidence_code: float, confidence_aggregate: float, explanation: str)[source]#

Bases: BaseModel

A single leaf code assigned to a feedback record.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config = {'frozen': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

code_id: str#

code_label: str#

confidence_type: float#

confidence_category: float#

confidence_code: float#

confidence_aggregate: float#

explanation: str#

class qfa.domain.models.CodedFeedbackRecordModel(*, feedback_record_id: str, assigned_codes: tuple[AssignedCodeModel, ...])[source]#

Bases: BaseModel

Coding output for one feedback record.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config = {'frozen': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

feedback_record_id: str#

assigned_codes: tuple[AssignedCodeModel, ...]#

class qfa.domain.models.CodingAssignmentResultModel(*, coded_feedback_records: tuple[CodedFeedbackRecordModel, ...])[source]#

Bases: BaseModel

The result of assigning codes to multiple feedback records.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config = {'frozen': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

coded_feedback_records: tuple[CodedFeedbackRecordModel, ...]#

class qfa.domain.models.LLMResponse(*, structured: T_Response, model: str, prompt_tokens: int, completion_tokens: int, cost: float)[source]#

Bases: BaseModel, Generic[T_Response]

Raw response from an LLM provider.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config = {'frozen': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

structured: T_Response#

model: str#

prompt_tokens: int#

completion_tokens: int#

cost: float#

class qfa.domain.models.TenantApiKey(*, key_id: str, name: str, key: SecretStr, tenant_id: str, is_superuser: bool = False)[source]#

Bases: BaseModel

An API key associated with a tenant.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config = {'frozen': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

key_id: str#

name: str#

key: SecretStr#

tenant_id: str#

is_superuser: bool#

class qfa.domain.models.Operation(*values)[source]#

Bases: StrEnum

Orchestrator operations that produce LLM calls.

Stored as plain strings in the database; new members can be added without a DB migration. UNKNOWN is a sentinel for backfilled rows from before per-operation tracking was introduced and must never be removed (removal would orphan historical rows).

ANALYZE = 'analyze'#

SUMMARIZE = 'summarize'#

SUMMARIZE_AGGREGATE = 'summarize_aggregate'#

ASSIGN_CODES = 'assign_codes'#

UNKNOWN = 'unknown'#

class qfa.domain.models.CallStatus(*values)[source]#

Bases: StrEnum

Outcome of a single LLM call attempt.

OK = 'ok'#

ERROR = 'error'#

class qfa.domain.models.CallContext(*, tenant_id: str, operation: Operation)[source]#

Bases: BaseModel

Per-call context propagated via ContextVar from orchestrator to tracker.

Variables:

tenant_id (str) – Tenant making the call.
operation (Operation) – Public orchestrator operation that issued the call.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config = {'frozen': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

tenant_id: str#

operation: Operation#

class qfa.domain.models.LLMCallRecord(*, tenant_id: str, operation: Operation, timestamp: datetime, call_duration_ms: int, model: str, input_tokens: int = 0, output_tokens: int = 0, cost_usd: Decimal = Decimal('0'), status: CallStatus, error_class: str | None = None)[source]#

Bases: BaseModel

A single recorded LLM call attempt for usage and cost tracking.

Recorded once per LLM-call attempt — success or failure. cost_usd and token counts are populated only for successful attempts; failures record zeros plus error_class.

Variables:

tenant_id (str) – Tenant that made the call.
operation (Operation) – Public orchestrator operation that issued the call.
timestamp (datetime) – UTC wall-clock when the call started.
call_duration_ms (int) – Wall-clock duration of the call in milliseconds.
model (str) – The LLM model used.
input_tokens (int) – Number of input (prompt) tokens; 0 on failure.
output_tokens (int) – Number of output (completion) tokens; 0 on failure.
cost_usd (Decimal) – Estimated cost in USD; 0 on failure.
status (CallStatus) – Outcome of the attempt.
error_class (str | None) – type(exc).__name__ when status == CallStatus.ERROR; None otherwise. Enforced by model_validator.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config = {'frozen': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

tenant_id: str#

operation: Operation#

timestamp: datetime#

call_duration_ms: int#

model: str#

input_tokens: int#

output_tokens: int#

cost_usd: Decimal#

status: CallStatus#

error_class: str | None#

class qfa.domain.models.DistributionStats(*, avg: float, min: float, max: float, p5: float, p95: float)[source]#

Bases: BaseModel

Statistical distribution summary.

Variables:

avg (float) – Mean value.
min (float) – Minimum value.
max (float) – Maximum value.
p5 (float) – 5th percentile.
p95 (float) – 95th percentile.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config = {'frozen': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

avg: float#

min: float#

max: float#

p5: float#

p95: float#

class qfa.domain.models.TokenStats(*, avg: float, min: float, max: float, p5: float, p95: float, total: int)[source]#

Bases: DistributionStats

Token distribution summary with a total count.

Variables:: total (int) – Total number of tokens.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

total: int#

model_config = {'frozen': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class qfa.domain.models.UsageStats(*, tenant_id: str | None = None, total_calls: int, failed_calls: int = 0, total_cost_usd: Decimal = Decimal('0'), call_duration: DistributionStats, input_tokens: TokenStats, output_tokens: TokenStats)[source]#

Bases: BaseModel

Aggregated usage statistics for a tenant or grand total.

The token and duration distributions and total_cost_usd are scoped to status='ok' rows. total_calls and failed_calls count all attempts including failures (policy “alpha”).

Variables:

tenant_id (str | None) – Tenant identifier, or None for grand total.
total_calls (int) – Total attempts (successful + failed).
failed_calls (int) – Attempts with status='error'.
total_cost_usd (Decimal) – Sum of cost over successful attempts only.
call_duration (DistributionStats) – Call duration distribution in milliseconds (successful attempts only).
input_tokens (TokenStats) – Input token distribution (successful attempts only).
output_tokens (TokenStats) – Output token distribution (successful attempts only).

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config = {'frozen': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

tenant_id: str | None#

total_calls: int#

failed_calls: int#

total_cost_usd: Decimal#

call_duration: DistributionStats#

input_tokens: TokenStats#

output_tokens: TokenStats#