From d318110bbe133e75a9cc3852d86295e6f682e2a4 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 7 Mar 2026 10:58:39 +0000 Subject: [PATCH] fix(pipelines): seed standard processing pipeline as system default on startup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The pipeline management UI showed an empty list after first boot because no default system pipeline was created. This adds seed_default_pipeline() which: - Creates a system-owned (owner_id=NULL), is_default=True pipeline named "Standard Processing Pipeline" at application startup - Steps mirror the current hardcoded Celery processing workflow: convert_to_pdf → check_duplicates → ocr → extract_metadata → embed_metadata → compute_embedding → send_to_destinations - Is idempotent: no-op if any system pipeline already exists - Handles missing pipelines table gracefully (during first migration run) Also wires the seeder into app/main.py lifespan startup using the same pattern as seed_default_plans. 9 new tests added covering creation, step order, idempotency, and API visibility. Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com> --- app/api/pipelines.py | 92 ++++++++++++++++++++++++++++ app/main.py | 14 +++++ tests/test_api_pipelines.py | 117 ++++++++++++++++++++++++++++++++++++ 3 files changed, 223 insertions(+) diff --git a/app/api/pipelines.py b/app/api/pipelines.py index d1ee6c23..fc7bf851 100644 --- a/app/api/pipelines.py +++ b/app/api/pipelines.py @@ -792,3 +792,95 @@ def _unset_default(db: Session, owner_id: str | None) -> None: db.query(Pipeline).filter(Pipeline.owner_id == owner_id, Pipeline.is_default.is_(True)).update( {"is_default": False} ) + + +# --------------------------------------------------------------------------- +# Default system pipeline seeding +# --------------------------------------------------------------------------- + +# The steps that make up the standard document-processing workflow. The order +# here mirrors what the existing Celery-based pipeline executes for every +# uploaded file. +_DEFAULT_PIPELINE_STEPS: list[tuple[str, str]] = [ + ("convert_to_pdf", "Convert to PDF"), + ("check_duplicates", "Check for Duplicates"), + ("ocr", "OCR Processing"), + ("extract_metadata", "Extract Metadata"), + ("embed_metadata", "Embed Metadata into PDF"), + ("compute_embedding", "Compute Text Embedding"), + ("send_to_destinations", "Send to Storage Destinations"), +] + +#: Human-readable name shown in the management UI for the auto-seeded pipeline. +DEFAULT_PIPELINE_NAME = "Standard Processing Pipeline" + + +def seed_default_pipeline(db: Session) -> int: + """Ensure a system-owned default pipeline exists in the database. + + This function is idempotent — it is a no-op when any system pipeline + (``owner_id IS NULL``) already exists. It is intended to be called once + at application startup (in ``app.main.lifespan``) so that the pipeline + management UI always shows the default workflow that mirrors the existing + Celery-based processing steps. + + The created pipeline: + + * ``owner_id = None`` — owned by the system, visible to all users + * ``is_default = True`` — selected automatically for new documents + * Steps (in order): convert_to_pdf → check_duplicates → ocr → + extract_metadata → embed_metadata → compute_embedding → + send_to_destinations + + Args: + db: An active SQLAlchemy session. + + Returns: + ``1`` if a new pipeline was created, ``0`` if one already existed. + """ + try: + if db.query(Pipeline).filter(Pipeline.owner_id.is_(None)).count() > 0: + return 0 + except Exception: + # Table may not exist yet during the very first migration run. + return 0 + + pipeline = Pipeline( + owner_id=None, + name=DEFAULT_PIPELINE_NAME, + description=( + "The standard document processing workflow: PDF conversion, " + "duplicate detection, OCR, metadata extraction and embedding, " + "semantic embeddings, and final distribution to storage destinations." + ), + is_default=True, + is_active=True, + ) + db.add(pipeline) + try: + db.flush() # Assign pipeline.id without committing yet + except Exception as exc: # pragma: no cover + db.rollback() + logger.error(f"Failed to create default pipeline: {exc}") + return 0 + + for pos, (step_type, label) in enumerate(_DEFAULT_PIPELINE_STEPS): + db.add( + PipelineStep( + pipeline_id=pipeline.id, + position=pos, + step_type=step_type, + label=label, + enabled=True, + ) + ) + + try: + db.commit() + logger.info("Seeded default system pipeline: '%s' (id=%d)", DEFAULT_PIPELINE_NAME, pipeline.id) + except Exception as exc: # pragma: no cover + db.rollback() + logger.error(f"Failed to seed default pipeline steps: {exc}") + return 0 + + return 1 diff --git a/app/main.py b/app/main.py index 5f39ae2c..7c46593f 100644 --- a/app/main.py +++ b/app/main.py @@ -112,6 +112,20 @@ async def lifespan(app: FastAPI): except Exception: logging.debug("Subscription plan seeding skipped — DB may not be ready yet") # noqa: S110 + # Seed the default system pipeline (mirrors the current hardcoded processing + # workflow) so it is immediately visible in the Pipelines management UI. + try: + from app.api.pipelines import seed_default_pipeline as _seed_pipeline + from app.database import SessionLocal as _SessionLocal # noqa: F811 (re-import for clarity) + + _db_pipeline = _SessionLocal() + try: + _seed_pipeline(_db_pipeline) + finally: + _db_pipeline.close() + except Exception: + logging.debug("Default pipeline seeding skipped — DB may not be ready yet") # noqa: S110 + # Application is now running yield diff --git a/tests/test_api_pipelines.py b/tests/test_api_pipelines.py index f5daf532..a0d7785f 100644 --- a/tests/test_api_pipelines.py +++ b/tests/test_api_pipelines.py @@ -465,3 +465,120 @@ class TestPipelineAPIHelpers: p = Pipeline(owner_id="user99", name="Private", is_default=False, is_active=True) assert _can_access_pipeline(p, "admin", admin=True) is True + + +# --------------------------------------------------------------------------- +# Unit + integration tests – seed_default_pipeline +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +class TestSeedDefaultPipeline: + """Tests for the seed_default_pipeline startup helper.""" + + def test_seed_creates_pipeline(self, db_session): + """seed_default_pipeline creates exactly one system pipeline.""" + from app.api.pipelines import seed_default_pipeline + + result = seed_default_pipeline(db_session) + + assert result == 1 + pipelines = db_session.query(Pipeline).filter(Pipeline.owner_id.is_(None)).all() + assert len(pipelines) == 1 + + def test_seeded_pipeline_is_default(self, db_session): + """The seeded pipeline has is_default=True and is_active=True.""" + from app.api.pipelines import seed_default_pipeline + + seed_default_pipeline(db_session) + p = db_session.query(Pipeline).filter(Pipeline.owner_id.is_(None)).first() + + assert p is not None + assert p.is_default is True + assert p.is_active is True + + def test_seeded_pipeline_has_correct_name(self, db_session): + """The seeded pipeline uses the canonical DEFAULT_PIPELINE_NAME.""" + from app.api.pipelines import DEFAULT_PIPELINE_NAME, seed_default_pipeline + + seed_default_pipeline(db_session) + p = db_session.query(Pipeline).filter(Pipeline.owner_id.is_(None)).first() + + assert p.name == DEFAULT_PIPELINE_NAME + + def test_seeded_pipeline_steps_count(self, db_session): + """The seeded pipeline has the correct number of steps.""" + from app.api.pipelines import _DEFAULT_PIPELINE_STEPS, seed_default_pipeline + + seed_default_pipeline(db_session) + p = db_session.query(Pipeline).filter(Pipeline.owner_id.is_(None)).first() + steps = db_session.query(PipelineStep).filter(PipelineStep.pipeline_id == p.id).all() + + assert len(steps) == len(_DEFAULT_PIPELINE_STEPS) + + def test_seeded_pipeline_step_types_and_order(self, db_session): + """Steps are in the correct order and match the expected step types.""" + from app.api.pipelines import _DEFAULT_PIPELINE_STEPS, seed_default_pipeline + + seed_default_pipeline(db_session) + p = db_session.query(Pipeline).filter(Pipeline.owner_id.is_(None)).first() + steps = ( + db_session.query(PipelineStep) + .filter(PipelineStep.pipeline_id == p.id) + .order_by(PipelineStep.position) + .all() + ) + + expected_types = [step_type for step_type, _ in _DEFAULT_PIPELINE_STEPS] + actual_types = [s.step_type for s in steps] + assert actual_types == expected_types + + def test_seeded_pipeline_all_steps_enabled(self, db_session): + """All seeded steps are enabled by default.""" + from app.api.pipelines import seed_default_pipeline + + seed_default_pipeline(db_session) + p = db_session.query(Pipeline).filter(Pipeline.owner_id.is_(None)).first() + steps = db_session.query(PipelineStep).filter(PipelineStep.pipeline_id == p.id).all() + + assert all(s.enabled for s in steps), "All seeded steps should be enabled" + + def test_seed_is_idempotent(self, db_session): + """Calling seed_default_pipeline twice does not create a duplicate.""" + from app.api.pipelines import seed_default_pipeline + + first = seed_default_pipeline(db_session) + second = seed_default_pipeline(db_session) + + assert first == 1 + assert second == 0 # No-op on second call + + count = db_session.query(Pipeline).filter(Pipeline.owner_id.is_(None)).count() + assert count == 1 + + def test_seeded_pipeline_visible_via_api(self, client): + """The default pipeline is visible in the GET /api/pipelines listing.""" + from app.api.pipelines import DEFAULT_PIPELINE_NAME, seed_default_pipeline + from app.database import get_db + + # Seed using the same DB session that the test client uses + db = next(client.app.dependency_overrides[get_db]()) + seed_default_pipeline(db) + + r = client.get("/api/pipelines") + assert r.status_code == 200 + names = [p["name"] for p in r.json()] + assert DEFAULT_PIPELINE_NAME in names + + def test_seeded_pipeline_default_flag_visible_via_api(self, client): + """The seeded pipeline is returned with is_default=True via the API.""" + from app.api.pipelines import DEFAULT_PIPELINE_NAME, seed_default_pipeline + from app.database import get_db + + db = next(client.app.dependency_overrides[get_db]()) + seed_default_pipeline(db) + + r = client.get("/api/pipelines") + default_pipelines = [p for p in r.json() if p["name"] == DEFAULT_PIPELINE_NAME] + assert len(default_pipelines) == 1 + assert default_pipelines[0]["is_default"] is True