Implementation:Ucbepic Docetl ServerModels

Knowledge Sources	Ucbepic_Docetl
Domains	Data_Processing, Web_API, Schema_Validation
Last Updated	2026-02-08 00:00 GMT

Overview

Concrete tool for Pydantic request and response schemas used by the DocETL server API provided by DocETL.

Description

The models module in the server application defines the Pydantic data models for API request and response payloads. It includes PipelineRequest for submitting YAML pipeline configs, PipelineConfigRequest for saving pipeline configurations with namespace/name metadata, TaskStatus enum for tracking async task states (pending, processing, completed, failed, cancelled), OptimizeResult and OptimizeRequest for optimization endpoints, and DecomposeRequest/DecomposeResult for operation decomposition endpoints. These models provide type safety and validation for all server API interactions.

Usage

Use these models when building or extending the DocETL server API endpoints. They define the contract between the frontend UI and the backend server.

Code Reference

Source Location

• Repository: Ucbepic_Docetl
• File: server/app/models.py
• Lines: 1-60

Signature

class PipelineRequest(BaseModel):
    yaml_config: str

class PipelineConfigRequest(BaseModel):
    namespace: str
    name: str
    config: str
    input_path: str
    output_path: str

class TaskStatus(str, Enum):
    PENDING = "pending"
    PROCESSING = "processing"
    COMPLETED = "completed"
    FAILED = "failed"
    CANCELLED = "cancelled"

class OptimizeResult(BaseModel):
    task_id: str
    status: TaskStatus
    should_optimize: str | None = None
    input_data: list[dict[str, Any]] | None = None
    output_data: list[dict[str, Any]] | None = None
    num_docs_analyzed: int | None = None
    cost: float | None = None
    error: str | None = None
    created_at: datetime
    completed_at: datetime | None = None

class OptimizeRequest(BaseModel):
    yaml_config: str
    step_name: str
    op_name: str

class DecomposeRequest(BaseModel):
    yaml_config: str
    step_name: str
    op_name: str

class DecomposeResult(BaseModel):
    task_id: str
    status: TaskStatus
    decomposed_operations: list[dict[str, Any]] | None = None
    winning_directive: str | None = None
    candidates_evaluated: int | None = None
    original_outputs: list[dict[str, Any]] | None = None
    decomposed_outputs: list[dict[str, Any]] | None = None
    comparison_rationale: str | None = None
    cost: float | None = None
    error: str | None = None
    created_at: datetime
    completed_at: datetime | None = None

Import

from server.app.models import (
    PipelineRequest,
    PipelineConfigRequest,
    TaskStatus,
    OptimizeResult,
    OptimizeRequest,
    DecomposeRequest,
    DecomposeResult,
)

I/O Contract

Inputs

Name	Type	Required	Description
yaml_config	str	Yes	YAML pipeline configuration string
namespace	str	Yes	User namespace for organizing pipelines
name	str	Yes	Pipeline name
step_name	str	Yes	Target step name (for optimize/decompose)
op_name	str	Yes	Target operation name (for optimize/decompose)

Outputs

Name	Type	Description
task_id	str	Unique identifier for the async task
status	TaskStatus	Current status of the task
cost	float or None	Total LLM cost incurred
error	str or None	Error message if task failed
decomposed_operations	list[dict] or None	List of decomposed operation configurations

Usage Examples

from server.app.models import OptimizeRequest, TaskStatus

# Create an optimization request
request = OptimizeRequest(
    yaml_config="default_model: gpt-4o\noperations: ...",
    step_name="process_step",
    op_name="extract_info",
)

# Check task status
status = TaskStatus.PROCESSING
print(f"Task is: {status.value}")  # "processing"

Related Pages

• Environment:Ucbepic_Docetl_Python_Runtime