"""
Pydantic models for API requests.
These define the contract between the API and external clients.
Following clean architecture principles, request models delegate validation
to domain model class methods and reuse field descriptions to avoid
duplication while maintaining single source of truth in the domain layer.
"""
from datetime import datetime, timezone
from typing import Any
from pydantic import BaseModel, Field, ValidationInfo, field_validator
from julee.domain.models import (
AssemblySpecification,
AssemblySpecificationStatus,
KnowledgeServiceQuery,
)
[docs]
class CreateAssemblySpecificationRequest(BaseModel):
"""Request model for creating an assembly specification.
This model defines what clients need to provide when creating a new
assembly specification. Validation logic is delegated to the domain
model to ensure consistency and avoid duplication.
Fields excluded from client control:
- assembly_specification_id: Always generated by the server
- status: Always set to DRAFT initially by the server
- created_at/updated_at: System-managed timestamps
"""
# Field definitions with descriptions reused from domain model
[docs]
name: str = Field(
description=AssemblySpecification.model_fields["name"].description
)
[docs]
applicability: str = Field(
description=AssemblySpecification.model_fields["applicability"].description
)
[docs]
jsonschema: dict[str, Any] = Field(
description=AssemblySpecification.model_fields["jsonschema"].description
)
[docs]
knowledge_service_queries: dict[str, str] = Field(
default_factory=dict,
description=AssemblySpecification.model_fields[
"knowledge_service_queries"
].description,
)
[docs]
version: str = Field(
default=AssemblySpecification.model_fields["version"].default,
description=AssemblySpecification.model_fields["version"].description,
)
# Delegate validation to domain model class methods
@field_validator("name")
@classmethod
[docs]
def validate_name(cls, v: str) -> str:
return AssemblySpecification.name_must_not_be_empty(v)
@field_validator("applicability")
@classmethod
[docs]
def validate_applicability(cls, v: str) -> str:
return AssemblySpecification.applicability_must_not_be_empty(v)
@field_validator("jsonschema")
@classmethod
[docs]
def validate_jsonschema(cls, v: dict[str, Any]) -> dict[str, Any]:
return AssemblySpecification.jsonschema_must_be_valid(v)
@field_validator("knowledge_service_queries")
@classmethod
[docs]
def validate_knowledge_service_queries(
cls, v: dict[str, str], info: ValidationInfo
) -> dict[str, str]:
return AssemblySpecification.knowledge_service_queries_must_be_valid(v, info)
@field_validator("version")
@classmethod
[docs]
def validate_version(cls, v: str) -> str:
return AssemblySpecification.version_must_not_be_empty(v)
[docs]
def to_domain_model(self, assembly_specification_id: str) -> AssemblySpecification:
"""Convert this request to a complete AssemblySpecification object.
Args:
assembly_specification_id: The ID to assign to the new spec
Returns:
AssemblySpecification: Complete domain object with system fields
"""
now = datetime.now(timezone.utc)
return AssemblySpecification(
assembly_specification_id=assembly_specification_id,
name=self.name,
applicability=self.applicability,
jsonschema=self.jsonschema,
knowledge_service_queries=self.knowledge_service_queries,
version=self.version,
status=AssemblySpecificationStatus.DRAFT,
created_at=now,
updated_at=now,
)
[docs]
class CreateKnowledgeServiceQueryRequest(BaseModel):
"""Request model for creating a knowledge service query.
This model defines what clients need to provide when creating a new
knowledge service query. Validation logic is delegated to the domain
model and descriptions are reused to avoid duplication while maintaining
single source of truth in the domain layer.
Fields excluded from client control:
- query_id: Always generated by the server
- created_at/updated_at: System-managed timestamps
"""
# Field definitions with descriptions reused from domain model
[docs]
name: str = Field(
description=KnowledgeServiceQuery.model_fields["name"].description
)
[docs]
knowledge_service_id: str = Field(
description=KnowledgeServiceQuery.model_fields[
"knowledge_service_id"
].description
)
[docs]
prompt: str = Field(
description=KnowledgeServiceQuery.model_fields["prompt"].description
)
[docs]
assistant_prompt: str | None = Field(
default=KnowledgeServiceQuery.model_fields["assistant_prompt"].default,
description=KnowledgeServiceQuery.model_fields["assistant_prompt"].description,
)
# Delegate validation to domain model class methods
@field_validator("name")
@classmethod
[docs]
def validate_name(cls, v: str) -> str:
return KnowledgeServiceQuery.name_must_not_be_empty(v)
@field_validator("knowledge_service_id")
@classmethod
[docs]
def validate_knowledge_service_id(cls, v: str) -> str:
return KnowledgeServiceQuery.knowledge_service_id_must_not_be_empty(v)
@field_validator("prompt")
@classmethod
[docs]
def validate_prompt(cls, v: str) -> str:
return KnowledgeServiceQuery.prompt_must_not_be_empty(v)
[docs]
def to_domain_model(self, query_id: str) -> KnowledgeServiceQuery:
"""Convert this request to a complete KnowledgeServiceQuery object.
Args:
query_id: The ID to assign to the new query
Returns:
KnowledgeServiceQuery: Complete domain object with system fields
"""
now = datetime.now(timezone.utc)
return KnowledgeServiceQuery(
query_id=query_id,
name=self.name,
knowledge_service_id=self.knowledge_service_id,
prompt=self.prompt,
query_metadata=self.query_metadata,
assistant_prompt=self.assistant_prompt,
created_at=now,
updated_at=now,
)