from typing import Optional, Any from fastapi import APIRouter, HTTPException, status, Query, Depends from app.schemas.pipeline import ( DeployRequest, DraftRequest, PipelineListResponse, PublishVersionRequest, RevertVersionRequest, VersionListResponse ) from app.services.pipeline_parser import ( list_all_pipelines, get_pipeline_by_id, save_pipeline_draft, deploy_pipeline_dag, duplicate_pipeline, delete_pipeline_files, publish_pipeline_version, list_pipeline_versions, revert_pipeline_version ) from app.core.auth.dependencies import get_optional_current_user from app.core.auth.models import UserSession router = APIRouter(tags=["Pipelines"]) def _resolve_user(current_user: Any) -> Optional[UserSession]: """Safely return UserSession or None when endpoint is called directly as a Python function.""" return current_user if isinstance(current_user, UserSession) else None @router.get("/pipelines", response_model=PipelineListResponse) def get_pipelines( scope: Optional[str] = Query(None, description="'my' for user's pipelines, 'all' for all pipelines (Admin only)"), current_user: Optional[UserSession] = Depends(get_optional_current_user) ): """ List pipelines filtered by owner_id or return all if Admin requested scope='all'. Enforces per-user workspace isolation according to design-spec. """ all_pipelines = list_all_pipelines() user = _resolve_user(current_user) if not user: # Fallback for unauthenticated test callers / backward compatibility return {"pipelines": all_pipelines} is_admin = user.is_admin or user.capability == "admin" if scope == "all" and is_admin: return {"pipelines": all_pipelines} # Filter to current user's pipelines (Overview workspace) filtered = [] for p in all_pipelines: p_owner = p.get("owner_id") if p_owner == user.id: filtered.append(p) elif not p_owner and is_admin: # Legacy pipelines without owner claim are visible to Admin filtered.append(p) return {"pipelines": filtered} @router.get("/pipelines/{pipeline_id}") def get_single_pipeline( pipeline_id: str, current_user: Optional[UserSession] = Depends(get_optional_current_user) ): """ Get a single pipeline by ID. Enforces that non-admin users cannot open pipelines owned by another user (HTTP 403). """ pipeline = get_pipeline_by_id(pipeline_id) if not pipeline: raise HTTPException(status_code=404, detail=f"Pipeline '{pipeline_id}' not found.") user = _resolve_user(current_user) if user: is_admin = user.is_admin or user.capability == "admin" p_owner = pipeline.get("owner_id") if p_owner and p_owner != user.id and not is_admin: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Akses ditolak: Anda tidak memiliki izin untuk melihat atau mengedit pipeline milik user lain." ) return pipeline @router.post("/pipelines/draft", status_code=status.HTTP_201_CREATED) def create_draft( req: DraftRequest, current_user: Optional[UserSession] = Depends(get_optional_current_user) ): """Save an in-progress pipeline design as draft.""" user = _resolve_user(current_user) existing = get_pipeline_by_id(req.pipeline_id) if existing and user: is_admin = user.is_admin or user.capability == "admin" p_owner = existing.get("owner_id") if p_owner and p_owner != user.id and not is_admin: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Akses ditolak: Anda tidak memiliki izin untuk mengubah pipeline milik user lain." ) owner_id = user.id if user else None owner_name = (user.name or user.username) if user else None return save_pipeline_draft( pipeline_id=req.pipeline_id, description=req.description, schedule=req.schedule, nodes=req.nodes, edges=req.edges, notes=req.notes, owner_id=owner_id, owner_name=owner_name ) @router.post("/pipelines/deploy", status_code=status.HTTP_201_CREATED) def deploy_pipeline( req: DeployRequest, current_user: Optional[UserSession] = Depends(get_optional_current_user) ): """Deploy pipeline design to JSON metadata and Airflow YAML configuration.""" user = _resolve_user(current_user) if user: # Enforce deploy permission (Per Decision #3: Data Engineer & Admin can deploy) if not user.can_deploy and not user.is_admin: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Akses ditolak: Peran Anda tidak memiliki izin untuk melakukan Deploy DAG." ) existing = get_pipeline_by_id(req.pipeline_id) if existing: is_admin = user.is_admin or user.capability == "admin" p_owner = existing.get("owner_id") if p_owner and p_owner != user.id and not is_admin: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Akses ditolak: Anda tidak memiliki izin untuk men-deploy pipeline milik user lain." ) owner_id = user.id if user else None owner_name = (user.name or user.username) if user else None return deploy_pipeline_dag( pipeline_id=req.pipeline_id, description=req.description, schedule=req.schedule, nodes=req.nodes, edges=req.edges, spec_yaml=req.spec_yaml, notes=req.notes, owner_id=owner_id, owner_name=owner_name ) @router.post("/pipelines/{pipeline_id}/duplicate", status_code=status.HTTP_201_CREATED) def duplicate_pipeline_endpoint( pipeline_id: str, current_user: Optional[UserSession] = Depends(get_optional_current_user) ): """ Duplicate a pipeline. Per Decision #4: The cloner becomes the owner, and the status is set to draft. """ source = get_pipeline_by_id(pipeline_id) if not source: raise HTTPException(status_code=404, detail=f"Pipeline '{pipeline_id}' not found.") user = _resolve_user(current_user) if user: is_admin = user.is_admin or user.capability == "admin" p_owner = source.get("owner_id") if p_owner and p_owner != user.id and not is_admin: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Akses ditolak: Anda tidak memiliki izin untuk menduplikasi pipeline ini." ) owner_id = user.id if user else "anonymous" owner_name = (user.name or user.username) if user else "Anonymous" new_id = f"{pipeline_id}_copy" return duplicate_pipeline( pipeline_id=pipeline_id, new_pipeline_id=new_id, owner_id=owner_id, owner_name=owner_name ) @router.post("/pipelines/publish", status_code=status.HTTP_201_CREATED) def publish_version(req: PublishVersionRequest): """ Publish a permanent semantic version tag for a pipeline. Creates version snapshot and updates DAG Bundle configuration. """ return publish_pipeline_version( pipeline_id=req.pipeline_id, version_tag=req.version_tag, changelog=req.changelog or "", bundle_name=req.bundle_name or "dlh-core-ingestion", nodes=req.nodes, edges=req.edges ) @router.get("/pipelines/{pipeline_id}/versions", response_model=VersionListResponse) def get_pipeline_versions(pipeline_id: str): """List historical version snapshots of a pipeline (for UI Time-Machine).""" return list_pipeline_versions(pipeline_id) @router.post("/pipelines/{pipeline_id}/revert") def revert_version(pipeline_id: str, req: RevertVersionRequest): """Revert the active canvas pipeline to a specific past historical version.""" result = revert_pipeline_version(pipeline_id, req.version_tag) if result.get("status") == "error": raise HTTPException(status_code=404, detail=result.get("message")) return result @router.delete("/pipelines/{pipeline_id}") def delete_pipeline( pipeline_id: str, current_user: Optional[UserSession] = Depends(get_optional_current_user) ): """Delete a pipeline and its associated configurations.""" pipeline = get_pipeline_by_id(pipeline_id) user = _resolve_user(current_user) if pipeline and user: is_admin = user.is_admin or user.capability == "admin" p_owner = pipeline.get("owner_id") if p_owner and p_owner != user.id and not is_admin: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Akses ditolak: Anda tidak memiliki izin untuk menghapus pipeline milik user lain." ) deleted = delete_pipeline_files(pipeline_id) if deleted: return {"status": "deleted", "pipeline_id": pipeline_id} raise HTTPException(status_code=404, detail="Pipeline configuration file not found")