Skip to content
Module 2 · Data LayerBeginner6 min read · 1,163 words

Lesson 07: Pydantic Schemas

"Good validation prevents bad data from entering your system."

📚 Table of Contents

  1. Introduction
  2. What are Schemas?
  3. Request Schemas
  4. Response Schemas
  5. Validation Rules
  6. Config Classes
  7. All Schemas Overview
  8. Key Takeaways
  9. Interview Questions
  10. Exercises

Introduction

Pydantic schemas define the shape of data entering and leaving your API. They provide:

  • Automatic validation
  • Type safety
  • Serialization
  • Documentation

In this lesson, you'll understand every schema in schemas.py and why they exist.

What are Schemas?

Models vs Schemas

Models (SQLAlchemy) = Database tables Schemas (Pydantic) = API contracts

python
# Model - Database
class Agent(Base):
    __tablename__ = "agents"
    id = Column(Integer)
    name = Column(String)
 
# Schema - API Request
class AgentCreate(BaseModel):
    name: str
    system_prompt: str
 
# Schema - API Response
class AgentResponse(BaseModel):
    id: int
    name: str
    created_at: datetime

Request Schemas

AgentCreate

python
class AgentCreate(BaseModel):
    name: str
    type: str
    description: str | None = None
    system_prompt: str
    language: str = "English"
    tone: str = "Professional"
    enabled_tools: str = ""

Purpose: Validate agent creation requests

Usage:

python
@router.post("/agents/")
def create_agent(agent_data: AgentCreate):
    # agent_data is validated automatically

Validation:

  • ✅ name is required, must be string
  • ✅ description is optional (can be None)
  • ✅ language defaults to "English"

ChatRequest

python
class ChatRequest(BaseModel):
    message: str

Simple but powerful - ensures message exists and is a string.

EmailAnalyzeRequest

python
class EmailAnalyzeRequest(BaseModel):
    email_text: str

TaskStatusUpdate

python
class TaskStatusUpdate(BaseModel):
    status: str

Note: Validation happens in router (checks allowed values).

DocumentSearchRequest

python
class DocumentSearchRequest(BaseModel):
    query: str
    top_k: int = 3

top_k: Number of results to return, defaults to 3.

RAGAnswerRequest

python
class RAGAnswerRequest(BaseModel):
    question: str
    top_k: int = 3

MultiAgentEmailWorkflowRequest

python
class MultiAgentEmailWorkflowRequest(BaseModel):
    email_text: str
    create_tasks: bool = True

create_tasks: Controls whether tasks are auto-created.

Response Schemas

AgentResponse

python
class AgentResponse(BaseModel):
    id: int
    name: str
    type: str
    description: str | None = None
    system_prompt: str
    language: str
    tone: str
    created_at: datetime
    enabled_tools: str
 
    class Config:
        from_attributes = True

Purpose: Format agent data for API responses

Config.from_attributes = True: Allows creating from SQLAlchemy models

Usage:

python
@router.get("/agents/{id}", response_model=AgentResponse)
def get_agent(agent_id: int):
    agent = db.query(Agent).first()
    return agent  # Auto-converted to AgentResponse

ChatResponse

python
class ChatResponse(BaseModel):
    agent_id: int
    user_message: str
    assistant_response: str
    run_id: int
    latency_ms: int

Returns: Complete chat interaction with metrics.

MessageResponse

python
class MessageResponse(BaseModel):
    id: int
    agent_id: int
    role: str
    content: str
    created_at: datetime
 
    class Config:
        from_attributes = True

AgentRunResponse

python
class AgentRunResponse(BaseModel):
    id: int
    agent_id: int
    input_text: str
    output_text: str | None = None
    latency_ms: int | None = None
    status: str
    error_message: str | None = None
    created_at: datetime

Optional fields: output_text, latency_ms, error_message can be None.

EmailAnalysisResponse

python
class EmailAnalysisResponse(BaseModel):
    summary: str
    intent: str
    priority: Literal["low", "medium", "high", "urgent"]
    deadline: str | None = None
    action_items: list[str]
    suggested_reply: str
 
    class Config:
        from_attributes = True

Literal: Restricts values to specific options.

TaskResponse

python
class TaskResponse(BaseModel):
    id: int
    agent_id: int
    title: str
    description: str | None = None
    priority: str
    status: str
    due_date: str | None = None
    source_type: str
    created_at: datetime
 
    class Config:
        from_attributes = True

ToolCallResponse

python
class ToolCallResponse(BaseModel):
    id: int
    agent_id: int
    agent_run_id: int | None = None
    tool_name: str
    tool_input: str | None = None
    tool_output: str | None = None
    success: str
    error_message: str | None = None
    latency_ms: int | None = None
    created_at: datetime
    model_name: str | None = None
    estimated_tokens: int | None = None
    estimated_cost: str | None = None
 
    class Config:
        from_attributes = True

Complex Nested Schemas

RAGAnswerResponse

python
class RAGSource(BaseModel):
    filename: str
    document_id: int
    chunk_index: int
    score: float | None = None
 
class RetrievalEvaluation(BaseModel):
    has_answer: bool
    confidence: float
    reason: str
 
class RAGAnswerResponse(BaseModel):
    question: str
    answer: str
    sources: list[RAGSource]
    retrieval_evaluation: RetrievalEvaluation | None = None

Nested structure: Response contains list of sources and evaluation object.

WorkflowReviewResponse

python
class WorkflowReviewResponse(BaseModel):
    approved: bool
    quality_score: float
    issues: list[str]
    recommendation: str

MultiAgentEmailWorkflowResponse

python
class MultiAgentEmailWorkflowResponse(BaseModel):
    email_text: str
    analysis: dict
    tasks_created: list[TaskResponse]
    suggested_reply: str
    review: WorkflowReviewResponse

Complex: Contains analysis dict, list of tasks, and nested review object.

Dashboard Schemas

AgentDashboardResponse

python
class AgentDashboardResponse(BaseModel):
    agent_id: int
    agent_name: str
    
    total_tasks: int
    open_tasks: int
    in_progress_tasks: int
    completed_tasks: int
    
    total_tool_calls: int
    successful_tool_calls: int
    failed_tool_calls: int
    
    total_workflow_runs: int
    successful_workflow_runs: int
    failed_workflow_runs: int
    
    average_latency_ms: float | None = None
    estimated_total_tokens: int
    estimated_total_cost: str
    
    average_quality_score: float | None = None

Comprehensive metrics: Everything needed for dashboard visualization.

Validation Rules

Type Validation

python
class ChatRequest(BaseModel):
    message: str  # Must be string
 
# Valid
ChatRequest(message="Hello")
 
# Invalid - raises ValidationError
ChatRequest(message=123)

Optional Fields

python
class AgentCreate(BaseModel):
    name: str
    description: str | None = None
 
# Valid
AgentCreate(name="Agent1")
AgentCreate(name="Agent1", description="Helpful")

Default Values

python
class DocumentSearchRequest(BaseModel):
    query: str
    top_k: int = 3
 
# If top_k not provided, uses 3
request = DocumentSearchRequest(query="vacation policy")
# request.top_k == 3

Literal Types

python
from typing import Literal
 
class EmailAnalysisResponse(BaseModel):
    priority: Literal["low", "medium", "high", "urgent"]
 
# Valid
EmailAnalysisResponse(priority="high")
 
# Invalid - raises error
EmailAnalysisResponse(priority="extreme")

Lists

python
action_items: list[str]
 
# Valid
["task 1", "task 2", "task 3"]
 
# Invalid
["task 1", 123, "task 3"]  # 123 is not a string

Config Classes

from_attributes

python
class AgentResponse(BaseModel):
    id: int
    name: str
    
    class Config:
        from_attributes = True

What it does: Allows creating from SQLAlchemy models

Without it:

python
agent = db.query(Agent).first()
return AgentResponse(**agent.__dict__)  # Manual conversion

With it:

python
agent = db.query(Agent).first()
return agent  # Automatic conversion!

All Schemas Overview

Request Schemas (8)

  1. AgentCreate
  2. ChatRequest
  3. EmailAnalyzeRequest
  4. TaskStatusUpdate
  5. DocumentSearchRequest
  6. RAGAnswerRequest
  7. MultiAgentEmailWorkflowRequest

Response Schemas (15+)

  1. AgentResponse
  2. ChatResponse
  3. MessageResponse
  4. AgentRunResponse
  5. EmailAnalysisResponse
  6. TaskResponse
  7. ToolCallResponse
  8. DocumentResponse
  9. DocumentChunkResponse
  10. DocumentSearchResponse
  11. DocumentSearchResult
  12. RAGAnswerResponse
  13. RAGSource
  14. RetrievalEvaluation
  15. WorkflowReviewResponse
  16. MultiAgentEmailWorkflowResponse
  17. WorkflowStepResponse
  18. WorkflowRunResponse
  19. AgentDashboardResponse
  20. AgentStatsResponse

Key Takeaways

🎯 Schemas validate API data automatically with Pydantic

🎯 Request schemas define what clients must send

🎯 Response schemas define what API returns

🎯 Config.from_attributes enables SQLAlchemy model conversion

🎯 Optional fields use | None = None syntax

🎯 Literal types restrict values to specific options

🎯 Nested schemas enable complex response structures

Interview Questions

Junior

Q1: What's the difference between a model and a schema?
A: Models define database tables (SQLAlchemy). Schemas define API contracts (Pydantic).

Q2: What does str | None = None mean?
A: Field can be a string or None, defaults to None if not provided.

Mid-Level

Q3: Why use Config.from_attributes = True?
A: Allows creating Pydantic models directly from SQLAlchemy models without manual conversion.

Q4: How does Pydantic validation prevent bugs?
A: Rejects invalid data before code executes, preventing type errors and missing fields.

Senior

Q5: Design a schema versioning strategy for API evolution.
A: Use v1/v2 prefixes in schemas, maintain both versions, deprecate old versions gradually with warnings.

Exercises

Exercise 1

Create a UserCreate schema with email, password, and optional profile fields.

Exercise 2

Add validation to ensure top_k is between 1 and 10.

Exercise 3

Create a nested schema for blog posts with author information.

Next Steps

You now understand all schemas in VeyraOps AI.

Next: Services Layer - where business logic lives.

👉 Lesson 08: Services →

← Previous: Models | ↑ Index | Next: Services →

Module 2 · Data Layer

PreviousLesson 06 - Database ModelsNextLesson 08 - Service Layer Overview