xpltdco/promptlooper
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

MAESTRO: Create backend/schemas.py with all Pydantic request/response schemas

Browse source 
Create/update/response schemas for Project, Experiment, Run, Endpoint,
Webhook, Score, Auth (setup/login/token), Export, and Health. All use
Pydantic v2 ConfigDict(from_attributes=True) for ORM compatibility.
RunDetailResponse nests StageResults and Scores. ExportRunRow provides
flat scorer_name→value dict for CSV/JSON export. 30 tests added.
This commit is contained in:
John Lightner 2026-04-07 01:54:02 -05:00
parent 0ec75ab617
commit 42668eeeb1
3 changed files with 639 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
Auto Run Docs/01-scaffold.md
View file

@ -26,7 +26,8 @@ Set up the PromptLooper repository, Docker infrastructure, and basic project ske
- [x] Set up Alembic: create alembic.ini and alembic/env.py configured to read DATABASE_URL from the config. Generate and apply the initial migration from the models.
> Created alembic.ini with logging config and script_location pointing to alembic/. env.py reads DATABASE_URL from backend.config.settings (with override support for tests). Added script.py.mako template. Generated initial migration (e1909678e89e) with all 8 tables, indexes, foreign keys, and enums. Migration applies cleanly on SQLite (render_as_batch=True for SQLite compatibility). 5 tests in tests/test_alembic.py covering upgrade/downgrade/columns/indexes/FKs. All 34 backend tests pass.
- [ ] Create backend/schemas.py with Pydantic request/response schemas for all API endpoints. Include create/update/response schemas for Project, Experiment, Run, Endpoint, and Webhook. Include the Score input schema and export format schemas.
- [x] Create backend/schemas.py with Pydantic request/response schemas for all API endpoints. Include create/update/response schemas for Project, Experiment, Run, Endpoint, and Webhook. Include the Score input schema and export format schemas.
> Created backend/schemas.py with all Pydantic v2 schemas using ConfigDict(from_attributes=True) for ORM compatibility. Includes: Project (create/update/response/list), Experiment (create/update/response/list), Run (response/list/detail with nested stages+scores), StageResult (response), Score (input/response), Endpoint (create/update/response/list), Webhook (create/update/response/list), Auth (setup/login/token/user), Export (run row with scores dict, export response), and Health. 30 tests in tests/test_schemas.py all passing. All 64 backend tests pass.
- [ ] Create backend/main.py with the FastAPI application. Set up CORS middleware, mount all routers (even if they're stubs), configure the WebSocket endpoint, add the /health endpoint that checks DB and Redis connectivity, and add startup/shutdown lifecycle hooks.

298
backend/schemas.py Normal file
View file

@ -0,0 +1,298 @@
"""PromptLooper Pydantic request/response schemas."""
import uuid
from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field
from models import ExperimentStatus, RunStatus
# ---------------------------------------------------------------------------
# Shared mixins
# ---------------------------------------------------------------------------
class _TimestampMixin(BaseModel):
created_at: datetime
updated_at: datetime
# ---------------------------------------------------------------------------
# Project
# ---------------------------------------------------------------------------
class ProjectCreate(BaseModel):
name: str = Field(..., min_length=1, max_length=255)
description: str | None = None
class ProjectUpdate(BaseModel):
name: str | None = Field(None, min_length=1, max_length=255)
description: str | None = None
class ProjectResponse(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: uuid.UUID
name: str
description: str | None
owner_id: uuid.UUID
created_at: datetime
updated_at: datetime
class ProjectListResponse(BaseModel):
items: list[ProjectResponse]
total: int
# ---------------------------------------------------------------------------
# Experiment
# ---------------------------------------------------------------------------
class ExperimentCreate(BaseModel):
name: str = Field(..., min_length=1, max_length=255)
description: str | None = None
sample_data: dict | None = None
pipeline_stages: dict | None = None
scoring_config: dict | None = None
parameter_space: dict | None = None
class ExperimentUpdate(BaseModel):
name: str | None = Field(None, min_length=1, max_length=255)
description: str | None = None
sample_data: dict | None = None
pipeline_stages: dict | None = None
scoring_config: dict | None = None
parameter_space: dict | None = None
status: ExperimentStatus | None = None
class ExperimentResponse(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: uuid.UUID
project_id: uuid.UUID
name: str
description: str | None
sample_data: dict | None
pipeline_stages: dict | None
scoring_config: dict | None
parameter_space: dict | None
status: ExperimentStatus
created_at: datetime
updated_at: datetime
class ExperimentListResponse(BaseModel):
items: list[ExperimentResponse]
total: int
# ---------------------------------------------------------------------------
# Run
# ---------------------------------------------------------------------------
class RunResponse(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: uuid.UUID
experiment_id: uuid.UUID
config_hash: str
config: dict
status: RunStatus
started_at: datetime | None
completed_at: datetime | None
duration_ms: int | None
tokens_in: int | None
tokens_out: int | None
cost_estimate: float | None
class RunListResponse(BaseModel):
items: list[RunResponse]
total: int
# ---------------------------------------------------------------------------
# StageResult (read-only, returned inside Run details)
# ---------------------------------------------------------------------------
class StageResultResponse(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: uuid.UUID
run_id: uuid.UUID
stage_index: int
prompt_sent: str
response_raw: str
model_used: str
parameters: dict | None
tokens_in: int | None
tokens_out: int | None
latency_ms: int | None
class RunDetailResponse(RunResponse):
"""Run with nested stage results and scores."""
stage_results: list[StageResultResponse] = []
scores: list["ScoreResponse"] = []
# ---------------------------------------------------------------------------
# Score
# ---------------------------------------------------------------------------
class ScoreInput(BaseModel):
scorer_name: str = Field(..., min_length=1, max_length=255)
value: float
metadata: dict | None = None
class ScoreResponse(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: uuid.UUID
run_id: uuid.UUID
scorer_name: str
value: float
scorer_metadata: dict | None
created_at: datetime
# ---------------------------------------------------------------------------
# Endpoint (LLM endpoint configuration)
# ---------------------------------------------------------------------------
class EndpointCreate(BaseModel):
name: str = Field(..., min_length=1, max_length=255)
url: str = Field(..., min_length=1, max_length=2048)
api_key: str | None = None
default_model: str | None = Field(None, max_length=255)
class EndpointUpdate(BaseModel):
name: str | None = Field(None, min_length=1, max_length=255)
url: str | None = Field(None, min_length=1, max_length=2048)
api_key: str | None = None
default_model: str | None = Field(None, max_length=255)
class EndpointResponse(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: uuid.UUID
name: str
url: str
default_model: str | None
class EndpointListResponse(BaseModel):
items: list[EndpointResponse]
total: int
# ---------------------------------------------------------------------------
# Webhook
# ---------------------------------------------------------------------------
class WebhookCreate(BaseModel):
event_type: str = Field(..., min_length=1, max_length=255)
url: str = Field(..., min_length=1, max_length=2048)
headers: dict | None = None
is_active: bool = True
class WebhookUpdate(BaseModel):
event_type: str | None = Field(None, min_length=1, max_length=255)
url: str | None = Field(None, min_length=1, max_length=2048)
headers: dict | None = None
is_active: bool | None = None
class WebhookResponse(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: uuid.UUID
event_type: str
url: str
headers: dict | None
is_active: bool
class WebhookListResponse(BaseModel):
items: list[WebhookResponse]
total: int
# ---------------------------------------------------------------------------
# Auth
# ---------------------------------------------------------------------------
class SetupRequest(BaseModel):
username: str = Field(..., min_length=1, max_length=255)
password: str = Field(..., min_length=8)
class LoginRequest(BaseModel):
username: str
password: str
class TokenResponse(BaseModel):
access_token: str
token_type: str = "bearer"
class UserResponse(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: uuid.UUID
username: str
is_admin: bool
created_at: datetime
# ---------------------------------------------------------------------------
# Export
# ---------------------------------------------------------------------------
class ExportRunRow(BaseModel):
"""Flat row for CSV/JSON export of run results."""
run_id: uuid.UUID
experiment_id: uuid.UUID
config_hash: str
config: dict
status: RunStatus
duration_ms: int | None = None
tokens_in: int | None = None
tokens_out: int | None = None
cost_estimate: float | None = None
scores: dict[str, float] = Field(
default_factory=dict,
description="Map of scorer_name → value",
)
class ExportResponse(BaseModel):
experiment_id: uuid.UUID
experiment_name: str
rows: list[ExportRunRow]
# ---------------------------------------------------------------------------
# Health
# ---------------------------------------------------------------------------
class HealthResponse(BaseModel):
status: str = "ok"
database: bool
redis: bool
# Rebuild forward refs for RunDetailResponse
RunDetailResponse.model_rebuild()

339
backend/tests/test_schemas.py Normal file
View file

@ -0,0 +1,339 @@
"""Tests for backend/schemas.py."""
import uuid
from datetime import datetime, timezone
import pytest
from pydantic import ValidationError
from models import ExperimentStatus, RunStatus
from schemas import (
EndpointCreate,
EndpointResponse,
EndpointUpdate,
ExperimentCreate,
ExperimentResponse,
ExperimentUpdate,
ExportResponse,
ExportRunRow,
HealthResponse,
LoginRequest,
ProjectCreate,
ProjectResponse,
ProjectUpdate,
RunDetailResponse,
RunResponse,
ScoreInput,
ScoreResponse,
SetupRequest,
StageResultResponse,
TokenResponse,
UserResponse,
WebhookCreate,
WebhookResponse,
WebhookUpdate,
)
NOW = datetime.now(timezone.utc)
UUID1 = uuid.uuid4()
UUID2 = uuid.uuid4()
# ---------------------------------------------------------------------------
# Project schemas
# ---------------------------------------------------------------------------
class TestProjectSchemas:
def test_create_valid(self) -> None:
p = ProjectCreate(name="My Project", description="desc")
assert p.name == "My Project"
assert p.description == "desc"
def test_create_name_required(self) -> None:
with pytest.raises(ValidationError):
ProjectCreate() # type: ignore[call-arg]
def test_create_empty_name_rejected(self) -> None:
with pytest.raises(ValidationError):
ProjectCreate(name="")
def test_update_partial(self) -> None:
p = ProjectUpdate(name="New Name")
assert p.name == "New Name"
assert p.description is None
def test_response_from_attributes(self) -> None:
class Fake:
id = UUID1
name = "Proj"
description = None
owner_id = UUID2
created_at = NOW
updated_at = NOW
r = ProjectResponse.model_validate(Fake())
assert r.id == UUID1
assert r.name == "Proj"
# ---------------------------------------------------------------------------
# Experiment schemas
# ---------------------------------------------------------------------------
class TestExperimentSchemas:
def test_create_minimal(self) -> None:
e = ExperimentCreate(name="Exp 1")
assert e.name == "Exp 1"
assert e.sample_data is None
def test_create_with_all_fields(self) -> None:
e = ExperimentCreate(
name="Full",
description="desc",
sample_data={"key": "value"},
pipeline_stages={"stages": []},
scoring_config={"scorer": "exact"},
parameter_space={"temp": [0.5, 1.0]},
)
assert e.parameter_space == {"temp": [0.5, 1.0]}
def test_update_status(self) -> None:
e = ExperimentUpdate(status=ExperimentStatus.running)
assert e.status == ExperimentStatus.running
def test_response_from_attributes(self) -> None:
class Fake:
id = UUID1
project_id = UUID2
name = "Exp"
description = None
sample_data = None
pipeline_stages = None
scoring_config = None
parameter_space = None
status = ExperimentStatus.draft
created_at = NOW
updated_at = NOW
r = ExperimentResponse.model_validate(Fake())
assert r.status == ExperimentStatus.draft
# ---------------------------------------------------------------------------
# Run schemas
# ---------------------------------------------------------------------------
class TestRunSchemas:
def test_response_from_attributes(self) -> None:
class Fake:
id = UUID1
experiment_id = UUID2
config_hash = "abc123"
config = {"model": "gpt-4"}
status = RunStatus.completed
started_at = NOW
completed_at = NOW
duration_ms = 1234
tokens_in = 100
tokens_out = 200
cost_estimate = 0.003
r = RunResponse.model_validate(Fake())
assert r.duration_ms == 1234
assert r.cost_estimate == 0.003
def test_detail_response_nested(self) -> None:
data = {
"id": UUID1,
"experiment_id": UUID2,
"config_hash": "abc",
"config": {},
"status": RunStatus.pending,
"started_at": None,
"completed_at": None,
"duration_ms": None,
"tokens_in": None,
"tokens_out": None,
"cost_estimate": None,
"stage_results": [],
"scores": [],
}
r = RunDetailResponse(**data)
assert r.stage_results == []
assert r.scores == []
# ---------------------------------------------------------------------------
# Score schemas
# ---------------------------------------------------------------------------
class TestScoreSchemas:
def test_input_valid(self) -> None:
s = ScoreInput(scorer_name="exact_match", value=0.95, metadata={"note": "ok"})
assert s.value == 0.95
assert s.metadata == {"note": "ok"}
def test_input_missing_name(self) -> None:
with pytest.raises(ValidationError):
ScoreInput(value=0.5) # type: ignore[call-arg]
def test_response_from_attributes(self) -> None:
class Fake:
id = UUID1
run_id = UUID2
scorer_name = "bleu"
value = 0.8
scorer_metadata = {"n": 4}
created_at = NOW
r = ScoreResponse.model_validate(Fake())
assert r.scorer_metadata == {"n": 4}
# ---------------------------------------------------------------------------
# Endpoint schemas
# ---------------------------------------------------------------------------
class TestEndpointSchemas:
def test_create_valid(self) -> None:
e = EndpointCreate(name="OpenAI", url="https://api.openai.com/v1")
assert e.api_key is None
def test_create_empty_name_rejected(self) -> None:
with pytest.raises(ValidationError):
EndpointCreate(name="", url="https://example.com")
def test_update_partial(self) -> None:
e = EndpointUpdate(url="https://new-url.com")
assert e.name is None
# ---------------------------------------------------------------------------
# Webhook schemas
# ---------------------------------------------------------------------------
class TestWebhookSchemas:
def test_create_valid(self) -> None:
w = WebhookCreate(
event_type="run.completed",
url="https://hooks.example.com/promptlooper",
headers={"Authorization": "Bearer xyz"},
)
assert w.is_active is True
def test_create_inactive(self) -> None:
w = WebhookCreate(
event_type="run.failed",
url="https://example.com",
is_active=False,
)
assert w.is_active is False
def test_update_partial(self) -> None:
w = WebhookUpdate(is_active=False)
assert w.event_type is None
assert w.is_active is False
def test_response_from_attributes(self) -> None:
class Fake:
id = UUID1
event_type = "run.completed"
url = "https://example.com"
headers = None
is_active = True
r = WebhookResponse.model_validate(Fake())
assert r.event_type == "run.completed"
# ---------------------------------------------------------------------------
# Auth schemas
# ---------------------------------------------------------------------------
class TestAuthSchemas:
def test_setup_password_min_length(self) -> None:
with pytest.raises(ValidationError):
SetupRequest(username="admin", password="short")
def test_setup_valid(self) -> None:
s = SetupRequest(username="admin", password="securepass123")
assert s.username == "admin"
def test_login_valid(self) -> None:
l = LoginRequest(username="user", password="pass")
assert l.username == "user"
def test_token_response(self) -> None:
t = TokenResponse(access_token="jwt.token.here")
assert t.token_type == "bearer"
def test_user_response_from_attributes(self) -> None:
class Fake:
id = UUID1
username = "admin"
is_admin = True
created_at = NOW
r = UserResponse.model_validate(Fake())
assert r.is_admin is True
# ---------------------------------------------------------------------------
# Export schemas
# ---------------------------------------------------------------------------
class TestExportSchemas:
def test_export_run_row(self) -> None:
row = ExportRunRow(
run_id=UUID1,
experiment_id=UUID2,
config_hash="abc",
config={"model": "gpt-4"},
status=RunStatus.completed,
duration_ms=500,
tokens_in=10,
tokens_out=20,
cost_estimate=0.001,
scores={"exact_match": 1.0, "bleu": 0.85},
)
assert row.scores["bleu"] == 0.85
def test_export_run_row_default_scores(self) -> None:
row = ExportRunRow(
run_id=UUID1,
experiment_id=UUID2,
config_hash="abc",
config={},
status=RunStatus.pending,
)
assert row.scores == {}
def test_export_response(self) -> None:
r = ExportResponse(
experiment_id=UUID1,
experiment_name="Test Exp",
rows=[],
)
assert r.rows == []
# ---------------------------------------------------------------------------
# Health schema
# ---------------------------------------------------------------------------
class TestHealthSchema:
def test_health_response(self) -> None:
h = HealthResponse(database=True, redis=False)
assert h.status == "ok"
assert h.database is True
assert h.redis is False