"""
DocumentPolicyValidation domain models for the Capture, Extract, Assemble,
Publish workflow.
This module contains the DocumentPolicyValidation domain object that
represents
the result of validating a document against a policy configuration in the CEAP
workflow system.
A DocumentPolicyValidation captures the complete validation process including:
- The document being validated and the policy used
- Actual validation scores achieved against policy criteria
- Optional transformation results and post-transformation scores
- Status tracking throughout the validation lifecycle
All domain models use Pydantic BaseModel for validation, serialization,
and type safety, following the patterns established in the sample project.
"""
from datetime import datetime, timezone
from enum import Enum
from pydantic import BaseModel, Field, field_validator
[docs]
class DocumentPolicyValidationStatus(str, Enum):
"""Status of a document policy validation process."""
[docs]
IN_PROGRESS = "in_progress"
[docs]
VALIDATION_COMPLETE = "validation_complete"
[docs]
class DocumentPolicyValidation(BaseModel):
"""Represents the validation of a document against a policy configuration.
A DocumentPolicyValidation tracks the complete lifecycle of validating
a document against policy criteria. It includes:
1. Initial validation: Document is scored against policy validation
queries
2. Optional transformation: If policy includes transformation queries and
initial validation fails, transformations are applied
3. Re-validation: Transformed document is re-scored against policy
criteria
4. Final determination: Pass/fail based on final validation scores
The validation process supports both validation-only policies and policies
that include transformations for document quality improvement.
"""
# Core validation identification
[docs]
validation_id: str = Field(
description="Unique identifier for this validation instance"
)
[docs]
policy_id: str = Field(
description="ID of the policy configuration used for validation"
)
# Validation process status
[docs]
status: DocumentPolicyValidationStatus = DocumentPolicyValidationStatus.PENDING
# Initial validation results
[docs]
validation_scores: list[tuple[str, int]] = Field(
default_factory=list,
description="List of (knowledge_service_query_id, actual_score) "
"tuples representing the scores achieved during initial validation. "
"Scores are between 0 and 100",
)
# Transformation results (if applicable)
[docs]
post_transform_validation_scores: list[tuple[str, int]] | None = Field(
default=None,
description="List of (knowledge_service_query_id, actual_score) "
"tuples representing scores achieved after transformation. "
"Only present if transformations were applied and re-validation "
"occurred",
)
# Validation metadata
[docs]
started_at: datetime | None = Field(
default_factory=lambda: datetime.now(timezone.utc),
description="When the validation process was initiated",
)
[docs]
completed_at: datetime | None = Field(
default=None, description="When the validation process completed"
)
[docs]
error_message: str | None = Field(
default=None, description="Error message if validation process failed"
)
# Results summary
[docs]
passed: bool | None = Field(
default=None,
description="Whether the document passed policy validation. "
"None while validation is in progress, True/False when complete",
)
@field_validator("input_document_id")
@classmethod
@field_validator("policy_id")
@classmethod
[docs]
def policy_id_must_not_be_empty(cls, v: str) -> str:
if not v or not v.strip():
raise ValueError("Policy ID cannot be empty")
return v.strip()
@field_validator("validation_scores")
@classmethod
[docs]
def validation_scores_must_be_valid(
cls, v: list[tuple[str, int]]
) -> list[tuple[str, int]]:
if not isinstance(v, list):
raise ValueError("Validation scores must be a list")
# Empty list is valid for pending validations
if not v:
return v
return cls._validate_score_tuples(v, "validation_scores")
@field_validator("post_transform_validation_scores")
@classmethod
[docs]
def post_transform_scores_must_be_valid(
cls, v: list[tuple[str, int]] | None
) -> list[tuple[str, int]] | None:
if v is None:
return v
if not isinstance(v, list):
raise ValueError("Post-transform validation scores must be a list or None")
# Empty list is valid
if not v:
return v
return cls._validate_score_tuples(v, "post_transform_validation_scores")
@field_validator("error_message")
@classmethod
[docs]
def error_message_must_be_valid(cls, v: str | None) -> str | None:
if v is None:
return v
if not isinstance(v, str):
raise ValueError("Error message must be a string or None")
return v.strip() if v.strip() else None
@field_validator("transformed_document_id")
@classmethod
@classmethod
def _validate_score_tuples(
cls, scores: list[tuple[str, int]], field_name: str
) -> list[tuple[str, int]]:
"""Helper method to validate score tuple lists."""
validated_scores = []
query_ids_seen = set()
for item in scores:
if not isinstance(item, tuple) or len(item) != 2:
raise ValueError(
f"Each item in {field_name} must be a 2-tuple of "
f"(query_id, actual_score)"
)
query_id, actual_score = item
# Validate query ID
if not isinstance(query_id, str) or not query_id.strip():
raise ValueError(f"Query ID in {field_name} must be a non-empty string")
query_id = query_id.strip()
# Check for duplicate query IDs within this field
if query_id in query_ids_seen:
raise ValueError(f"Duplicate query ID '{query_id}' in {field_name}")
query_ids_seen.add(query_id)
# Validate actual score
if not isinstance(actual_score, int):
raise ValueError(
f"Actual score in {field_name} must be an integer "
f"between 0 and 100"
)
if actual_score < 0 or actual_score > 100:
raise ValueError(
f"Actual score {actual_score} in {field_name} must be "
f"between 0 and 100"
)
validated_scores.append((query_id, actual_score))
return validated_scores