Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -0,0 +1,155 @@
"""Fold assessment_run into assessment

Revision ID: 086
Revises: 085
Create Date: 2026-10-06 00:00:00.000000

The child was 1:1 with its parent everywhere, so its config + runtime columns move
onto `assessment` and the table is dropped. `error_message` lands in
`execution.error`; keys the code no longer reads are stripped. `input` goes: nothing
writes it since the console pipeline was retired. Downgrade recreates `assessment_run`
empty: take a `pg_dump -t assessment_run` before upgrading, the rows come back from it.
An assessment with no run fails the `SET NOT NULL` on purpose rather than migrating half.
"""

import sqlalchemy as sa
from alembic import op
from sqlalchemy.dialects import postgresql

revision = "086"
down_revision = "085"
branch_labels = None
depends_on = None

CONFIG_INDEX = "idx_assessment_config"
CONFIG_FK = "fk_assessment_config_id_config"
RUN_TABLE = "assessment_run"
# Derivable from config_version or already in the stage dumps.
DROPPED_EXECUTION_KEYS = (
"stage_output_urls",
"provider",
"model",
"input_schema",
"verdicts",
"gate_passed",
)


def upgrade() -> None:
op.add_column(
"assessment",
sa.Column("config_id", postgresql.UUID(as_uuid=True), nullable=True),
)
op.add_column(
"assessment", sa.Column("config_version", sa.Integer(), nullable=True)
)
op.add_column(
"assessment",
sa.Column(
"total_items",
sa.Integer(),
nullable=True,
comment="Submitted row count; sizes the result without reading the rows",
),
)
op.add_column(
"assessment",
sa.Column(
"execution",
postgresql.JSONB(astext_type=sa.Text()),
nullable=True,
comment="Staged-batch runtime (AssessmentExecution); NULL for RESPONSE",
),
)

strip = " ".join(f"- '{key}'" for key in DROPPED_EXECUTION_KEYS)
op.execute(
f"""
UPDATE assessment AS a
SET config_id = r.config_id,
config_version = r.config_version,
total_items = r.total_items,
execution = CASE
WHEN r.error_message IS NULL THEN r.execution {strip}
ELSE (COALESCE(r.execution, '{{}}'::jsonb) {strip})
|| jsonb_build_object('error', r.error_message)
END
FROM {RUN_TABLE} AS r
WHERE r.assessment_id = a.id
AND a.config_id IS NULL
"""
)

op.alter_column("assessment", "config_id", nullable=False)
op.alter_column("assessment", "config_version", nullable=False)
op.alter_column("assessment", "total_items", nullable=False)
op.create_foreign_key(CONFIG_FK, "assessment", "config", ["config_id"], ["id"])
op.create_index(CONFIG_INDEX, "assessment", ["config_id", "config_version"])

op.drop_column("assessment", "input")
op.drop_table(RUN_TABLE)


def downgrade() -> None:
op.create_table(
RUN_TABLE,
sa.Column("id", sa.Integer(), primary_key=True),
sa.Column(
"assessment_id",
postgresql.UUID(as_uuid=True),
sa.ForeignKey(
"assessment.id",
name="fk_assessment_run_assessment_id",
ondelete="CASCADE",
),
nullable=False,
),
sa.Column(
"config_id",
postgresql.UUID(as_uuid=True),
sa.ForeignKey("config.id", name="fk_assessment_run_config_id"),
nullable=False,
),
sa.Column("config_version", sa.Integer(), nullable=False),
sa.Column(
"status",
postgresql.ENUM(name="assessment_status", create_type=False),
server_default="PENDING",
nullable=False,
),
sa.Column(
"batch_job_id",
sa.Integer(),
sa.ForeignKey(
"batch_job.id",
name="fk_assessment_run_batch_job_id",
ondelete="SET NULL",
),
nullable=True,
),
sa.Column("total_items", sa.Integer(), server_default="0", nullable=False),
sa.Column("execution", postgresql.JSONB(astext_type=sa.Text()), nullable=True),
sa.Column(
"post_processing_config",
postgresql.JSONB(astext_type=sa.Text()),
nullable=True,
),
sa.Column("error_message", sa.Text(), nullable=True),
sa.Column("inserted_at", sa.DateTime(), nullable=False),
sa.Column("updated_at", sa.DateTime(), nullable=False),
)
op.create_index("idx_assessment_run_assessment_id", RUN_TABLE, ["assessment_id"])
op.create_index(
"idx_assessment_run_config", RUN_TABLE, ["config_id", "config_version"]
)
op.create_index("idx_assessment_run_status", RUN_TABLE, ["status"])
op.add_column(
"assessment",
sa.Column("input", postgresql.JSONB(astext_type=sa.Text()), nullable=True),
)
op.drop_index(CONFIG_INDEX, table_name="assessment")
op.drop_constraint(CONFIG_FK, "assessment", type_="foreignkey")
op.drop_column("assessment", "execution")
op.drop_column("assessment", "total_items")
op.drop_column("assessment", "config_version")
op.drop_column("assessment", "config_id")
4 changes: 1 addition & 3 deletions backend/app/api/routes/assessment/__init__.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
from fastapi import APIRouter, Depends

from app.api.permissions import require_feature
from app.api.routes.assessment import assessments, datasets, runs
from app.api.routes.assessment import datasets
from app.core.feature_flags import FeatureFlag

router = APIRouter(
Expand All @@ -11,7 +11,5 @@
)

router.include_router(datasets.router)
router.include_router(assessments.router)
router.include_router(runs.router)

__all__ = ["router"]
6 changes: 3 additions & 3 deletions backend/app/api/routes/assessment/api.py
Original file line number Diff line number Diff line change
Expand Up @@ -85,7 +85,7 @@ def list_assessments(
offset: Annotated[int, Query(ge=0)] = 0,
) -> APIResponse[list[AssessmentSummary]]:
"""BATCH assessments newest-first, optionally narrowed to one config or version."""
rows = api_crud.list_assessments_with_execution(
rows = api_crud.list_assessments(
session=session,
organization_id=auth_context.organization_.id,
project_id=auth_context.project_.id,
Expand All @@ -96,8 +96,8 @@ def list_assessments(
)
return APIResponse.success_response(
data=[
results.build_summary(assessment, execution, submission_name)
for assessment, execution, submission_name in rows
results.build_summary(assessment, submission_name)
for assessment, submission_name in rows
]
)

Expand Down
7 changes: 4 additions & 3 deletions backend/app/celery/tasks/job_execution.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@
import logging
from collections.abc import Callable
from typing import TYPE_CHECKING, TypeVar
from uuid import UUID

from asgi_correlation_id import correlation_id
from celery import Task, current_task
Expand Down Expand Up @@ -370,7 +371,7 @@ def run_assessment_pipeline(
@gevent_timeout(settings.CELERY_TASK_SOFT_TIME_LIMIT, "run_assessment_api_batch")
def run_assessment_api_batch(
self,
execution_id: int,
assessment_id: str,
organization_id: int,
project_id: int,
trace_id: str,
Expand All @@ -389,15 +390,15 @@ def run_assessment_api_batch(
result = _run_with_otel_parent(
self,
lambda: run_batch_stage(
execution_id=execution_id,
assessment_id=UUID(assessment_id),
organization_id=organization_id,
project_id=project_id,
),
)
if result and result.get("requeue"):
self.apply_async(
kwargs={
"execution_id": execution_id,
"assessment_id": assessment_id,
"organization_id": organization_id,
"project_id": project_id,
"trace_id": trace_id,
Expand Down
Loading
Loading