refactor(pipelines): address code review - shared get_current_user_id, aria-live, deduplicate user ID logic

- Extract _get_user_id into shared auth.get_current_user_id() used by both
  pipelines API and the assign-pipeline endpoint in files API
- Fix aria-live attribute: use two separate static containers (polite/assertive)
  instead of dynamic Alpine.js binding for correct screen reader announcements
- Fix migration comment to accurately describe batch-mode FK creation
- Remove redundant tags parameter from reorder endpoint decorator
- Rename _make_file test helper to _make_test_file_record for clarity
- Update docs/UserGuide.md and docs/API.md with full Pipelines reference

Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>

app/api/files.py

@@ -1590,15 +1590,11 @@ def assign_pipeline_to_file(
         HTTPException 404: If the file or pipeline does not exist / is not
             accessible to the current user.
     """
-    from app.auth import get_current_user
+    from app.auth import get_current_user, get_current_user_id
     from app.models import Pipeline
 
     user = get_current_user(request)
-    # Derive user identity the same way the pipelines API does (_get_user_id)
+    user_id: str = get_current_user_id(request)
-    if user:
-        user_id: str = user.get("preferred_username") or user.get("email") or user.get("id") or "anonymous"
-    else:
-        user_id = "anonymous"
 
     is_admin_user = bool(user and user.get("is_admin"))
 

app/api/pipelines.py

@@ -18,7 +18,7 @@ from fastapi import APIRouter, Body, Depends, HTTPException, Request, status
 from pydantic import BaseModel, Field
 from sqlalchemy.orm import Session
 
-from app.auth import get_current_user, require_login
+from app.auth import get_current_user, get_current_user_id, require_login
 from app.database import get_db
 from app.models import Pipeline, PipelineStep
 
@@ -90,11 +90,12 @@ MAX_NAME_LENGTH = 255
 
 
 def _get_user_id(request: Request) -> str:
-    """Return a stable user identifier from the session."""
+    """Return a stable user identifier from the session.
-    user = get_current_user(request)
+
-    if user:
+    Delegates to :func:`app.auth.get_current_user_id` so the same fallback
-        return user.get("preferred_username") or user.get("email") or user.get("id", "anonymous")
+    logic ("anonymous") is used consistently throughout the application.
-    return "anonymous"
+    """
+    return get_current_user_id(request)
 
 
 def _is_admin(request: Request) -> bool:
@@ -591,7 +592,7 @@ def add_step(pipeline_id: int, request: Request, db: DbSession, body: PipelineSt
     return _serialize_step(step)
 
 
-@router.put("/{pipeline_id}/steps/reorder", tags=["pipelines"])
+@router.put("/{pipeline_id}/steps/reorder")
 @require_login
 def reorder_steps(
     pipeline_id: int,

app/auth.py

@@ -43,6 +43,27 @@ def get_current_user(request: Request):
     return request.session.get("user")
 
 
+def get_current_user_id(request: Request) -> str:
+    """Return a stable string identifier for the authenticated user.
+
+    Falls back to ``"anonymous"`` when no user is in the session (e.g. when
+    AUTH_ENABLED=False in single-user mode).  The returned value is consistent
+    with how pipeline and file ownership is stored in the database.
+
+    Priority order: ``preferred_username`` → ``email`` → ``id`` → ``"anonymous"``.
+
+    Args:
+        request: The current FastAPI request.
+
+    Returns:
+        A non-empty string identifying the current user.
+    """
+    user = get_current_user(request)
+    if not user or not isinstance(user, dict):
+        return "anonymous"
+    return user.get("preferred_username") or user.get("email") or user.get("id") or "anonymous"
+
+
 def require_login(func):
     if not AUTH_ENABLED:
         return func  # no-op

docs/API.md

@@ -1441,6 +1441,176 @@ Execute a full data migration from source to target database.
 ```
 
 
+## Pipelines
+
+The pipeline API lets you build and manage custom document processing workflows. Each pipeline is owned by a single user (or by the system when `owner_id` is `null`).
+
+### Step-types catalogue
+
+```bash
+GET /api/pipelines/step-types
+```
+
+Returns the catalogue of built-in step types.
+
+**Response (200):**
+```json
+{
+  "convert_to_pdf": {
+    "label": "Convert to PDF",
+    "description": "Convert non-PDF documents to PDF format using Gotenberg.",
+    "config_schema": {}
+  },
+  "ocr": {
+    "label": "OCR Processing",
+    "description": "Extract text using Azure Document Intelligence or local Tesseract.",
+    "config_schema": {
+      "force_cloud_ocr": { "type": "boolean", "default": false }
+    }
+  }
+}
+```
+
+### List pipelines
+
+```bash
+GET /api/pipelines
+```
+
+Returns pipelines visible to the current user (own + system pipelines). Admins see all pipelines.
+
+### Create pipeline
+
+```bash
+POST /api/pipelines
+Content-Type: application/json
+
+{
+  "name": "My Workflow",
+  "description": "Converts, OCRs, and stores documents.",
+  "is_default": false,
+  "is_active": true
+}
+```
+
+**Response (201):**
+```json
+{
+  "id": 1,
+  "owner_id": "alice",
+  "name": "My Workflow",
+  "description": "Converts, OCRs, and stores documents.",
+  "is_default": false,
+  "is_active": true,
+  "created_at": "2026-03-07T10:00:00+00:00",
+  "updated_at": "2026-03-07T10:00:00+00:00"
+}
+```
+
+### Create system pipeline (admin only)
+
+```bash
+POST /api/pipelines/admin/system
+Content-Type: application/json
+
+{
+  "name": "Global Default",
+  "is_default": true
+}
+```
+
+### Get pipeline with steps
+
+```bash
+GET /api/pipelines/{pipeline_id}
+```
+
+**Response (200):**
+```json
+{
+  "id": 1,
+  "owner_id": "alice",
+  "name": "My Workflow",
+  "steps": [
+    { "id": 1, "position": 0, "step_type": "convert_to_pdf", "enabled": true, "config": {} },
+    { "id": 2, "position": 1, "step_type": "ocr", "enabled": true, "config": { "force_cloud_ocr": false } }
+  ]
+}
+```
+
+### Update pipeline
+
+```bash
+PUT /api/pipelines/{pipeline_id}
+Content-Type: application/json
+
+{ "name": "Renamed Workflow", "is_default": true }
+```
+
+### Delete pipeline
+
+```bash
+DELETE /api/pipelines/{pipeline_id}
+```
+
+Returns **204 No Content**.
+
+### Add step
+
+```bash
+POST /api/pipelines/{pipeline_id}/steps
+Content-Type: application/json
+
+{
+  "step_type": "ocr",
+  "label": "Cloud OCR",
+  "config": { "force_cloud_ocr": true },
+  "enabled": true
+}
+```
+
+### Update step
+
+```bash
+PUT /api/pipelines/{pipeline_id}/steps/{step_id}
+Content-Type: application/json
+
+{ "enabled": false }
+```
+
+### Delete step
+
+```bash
+DELETE /api/pipelines/{pipeline_id}/steps/{step_id}
+```
+
+Returns **204 No Content**.
+
+### Reorder steps
+
+```bash
+PUT /api/pipelines/{pipeline_id}/steps/reorder
+Content-Type: application/json
+
+[3, 1, 2]
+```
+
+Provide a complete ordered list of **all** step IDs. Their positions are reassigned 0, 1, 2, … in the given order.
+
+### Assign pipeline to a file
+
+```bash
+POST /api/files/{file_id}/assign-pipeline?pipeline_id=2
+```
+
+Pass no `pipeline_id` query parameter (or omit it) to clear the assignment.
+
+**Response (200):**
+```json
+{ "file_id": 42, "pipeline_id": 2 }
+```
+
+
 ## Further Assistance
 
 For additional help with the API, please contact our support team or refer to the [Development Guide](../CONTRIBUTING.md).

docs/UserGuide.md

@@ -334,6 +334,55 @@ PAPERLESS_CUSTOM_FIELDS_MAPPING='{"absender": "Sender", "empfaenger": "Recipient
 3. After successful upload, custom fields are automatically populated
 4. You can view the populated fields in your Paperless-ngx document details
 
+## Processing Pipelines
+
+Processing pipelines let you define exactly what happens to your documents when they are uploaded. Each pipeline is an ordered sequence of **steps** — for example: convert to PDF → OCR → extract metadata → send to storage.
+
+### Key concepts
+
+| Term | Meaning |
+|------|---------|
+| **Pipeline** | A named, ordered list of processing steps |
+| **Step** | A single processing action (e.g., OCR, metadata extraction) |
+| **System pipeline** | Created by an admin; visible to all users as a shared default |
+| **User pipeline** | Created by a regular user; private to that user |
+| **Default pipeline** | Marked `is_default=true`; used automatically for new uploads |
+
+### Managing your pipelines
+
+1. Navigate to **Pipelines** in the top navigation bar.
+2. Click **New Pipeline** to create a pipeline, give it a name and optional description.
+3. Expand the pipeline card and click **Add Step** to build the workflow.
+4. Use the ↑ / ↓ arrows to reorder steps, or click the edit icon to change step settings.
+5. Mark a pipeline as **Default** so new documents are automatically processed by it.
+
+### Available step types
+
+| Step Type | Description |
+|-----------|-------------|
+| `convert_to_pdf` | Convert non-PDF files to PDF using Gotenberg |
+| `check_duplicates` | Detect duplicate files by content hash |
+| `ocr` | Extract text with Azure Document Intelligence or local Tesseract |
+| `extract_metadata` | Extract structured metadata (type, sender, tags) with AI |
+| `embed_metadata` | Write extracted metadata into the PDF document properties |
+| `compute_embedding` | Compute semantic embeddings for similarity search |
+| `send_to_destinations` | Upload the processed document to all configured storage destinations |
+| `classify` | Classify the document type with AI |
+
+### Assigning a pipeline to a file
+
+You can assign (or change) the pipeline for an individual document via the file detail page or the API:
+
+```bash
+POST /api/files/{file_id}/assign-pipeline?pipeline_id=3
+```
+
+Pass no `pipeline_id` to clear the assignment and fall back to the system default.
+
+### Admin: system-wide pipelines
+
+Admins can create **system pipelines** that appear in every user's pipeline list. These can be set as the global default so all users benefit from a consistent processing baseline. Navigate to **Pipelines** and check the **System pipeline** box when creating a new one (admin only).
+
 ## API Access
 
 For programmatic access, DocuElevate provides a comprehensive REST API:

frontend/templates/pipelines.html

@@ -29,17 +29,23 @@
   </div>
 
   <!-- ── Alert banner ───────────────────────────────────────────────────────── -->
-  <div x-show="alert.show" x-transition class="mb-4" role="alert" :aria-live="alert.type === 'error' ? 'assertive' : 'polite'">
+  <!-- Two containers with static aria-live so screen readers register them on page load -->
-    <div
+  <div aria-live="polite" aria-atomic="true">
-      :class="alert.type === 'success'
+    <div x-show="alert.show && alert.type !== 'error'" x-transition class="mb-4" role="status">
-        ? 'bg-green-50 border-green-400 text-green-800'
+      <div class="bg-green-50 border-green-400 text-green-800 border-l-4 p-4 rounded dark:bg-opacity-10">
-        : 'bg-red-50 border-red-400 text-red-800'"
-      class="border-l-4 p-4 rounded dark:bg-opacity-10"
-    >
         <p class="font-semibold" x-text="alert.title"></p>
         <p class="text-sm" x-text="alert.message"></p>
       </div>
     </div>
+  </div>
+  <div aria-live="assertive" aria-atomic="true">
+    <div x-show="alert.show && alert.type === 'error'" x-transition class="mb-4" role="alert">
+      <div class="bg-red-50 border-red-400 text-red-800 border-l-4 p-4 rounded dark:bg-opacity-10">
+        <p class="font-semibold" x-text="alert.title"></p>
+        <p class="text-sm" x-text="alert.message"></p>
+      </div>
+    </div>
+  </div>
 
   <!-- ── Loading ────────────────────────────────────────────────────────────── -->
   <template x-if="loading">

migrations/versions/017_add_pipelines.py

@@ -44,9 +44,8 @@ def upgrade() -> None:
         sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now()),
     )
 
-    # Use batch mode for SQLite compatibility when altering the files table.
+    # Use batch mode (copy-and-move strategy) for SQLite compatibility.
-    # SQLite does not support adding FK constraints inline via ALTER TABLE, so we
+    # A named FK constraint is created so Alembic can reference it in the downgrade.
-    # add the plain integer column and define the FK reference at the model level.
     with op.batch_alter_table("files") as batch_op:
         batch_op.add_column(
             sa.Column("pipeline_id", sa.Integer(), nullable=True),

tests/test_api_pipelines.py

@@ -16,7 +16,7 @@ from app.models import FileRecord, Pipeline, PipelineStep
 # ---------------------------------------------------------------------------
 
 
-def _make_file(db_session, owner_id=None):
+def _make_test_file_record(db_session, owner_id=None):
     """Insert a minimal FileRecord and return it.
 
     Default owner_id=None so tests work without an authenticated session.
@@ -344,7 +344,7 @@ class TestAssignPipelineToFile:
     def test_assign_pipeline_to_file(self, client, db_session):
         """Assigning a pipeline to a file stores pipeline_id on the record."""
         # File with no owner so anonymous test session can access it
-        fr = _make_file(db_session, owner_id=None)
+        fr = _make_test_file_record(db_session, owner_id=None)
         pipeline = client.post("/api/pipelines", json={"name": "Assign Test"}).json()
 
         r = client.post(f"/api/files/{fr.id}/assign-pipeline?pipeline_id={pipeline['id']}")
@@ -358,7 +358,7 @@ class TestAssignPipelineToFile:
 
     def test_clear_pipeline_from_file(self, client, db_session):
         """Passing no pipeline_id clears the assignment."""
-        fr = _make_file(db_session, owner_id=None)
+        fr = _make_test_file_record(db_session, owner_id=None)
         pipeline = client.post("/api/pipelines", json={"name": "Clearable"}).json()
         client.post(f"/api/files/{fr.id}/assign-pipeline?pipeline_id={pipeline['id']}")
 
@@ -368,7 +368,7 @@ class TestAssignPipelineToFile:
 
     def test_assign_nonexistent_pipeline_returns_404(self, client, db_session):
         """Assigning a non-existent pipeline returns 404."""
-        fr = _make_file(db_session, owner_id=None)
+        fr = _make_test_file_record(db_session, owner_id=None)
         r = client.post(f"/api/files/{fr.id}/assign-pipeline?pipeline_id=99999")
         assert r.status_code == 404