RakeshRamini changes to contentflow v0.1.2

.env.template

@@ -15,8 +15,6 @@ AZURE_LOCATION=eastus2
 COSMOS_DB_ENDPOINT=
 COSMOS_DB_NAME=contentflow
 STORAGE_ACCOUNT_NAME=
-STORAGE_ACCOUNT_WORKER_QUEUE_URL=
-STORAGE_WORKER_QUEUE_NAME=contentflow-execution-requests
 BLOB_STORAGE_CONTAINER_NAME=content
 APP_CONFIG_ENDPOINT=
 APPLICATIONINSIGHTS_CONNECTION_STRING=
@@ -33,26 +31,6 @@ API_SERVER_PORT=8090
 DEBUG=false
 LOG_LEVEL=INFO
 
-# ==============================================
-# Worker Configuration
-# ==============================================
-NUM_PROCESSING_WORKERS=4
-NUM_SOURCE_WORKERS=2
-WORKER_NAME=contentflow-worker
-LOG_LEVEL=INFO
-
-# Queue Settings
-QUEUE_POLL_INTERVAL_SECONDS=5
-QUEUE_VISIBILITY_TIMEOUT_SECONDS=300
-QUEUE_MAX_MESSAGES=32
-DEFAULT_POLLING_INTERVAL_SECONDS=300
-SCHEDULER_SLEEP_INTERVAL_SECONDS=5
-LOCK_TTL_SECONDS=300
-
-# Processing Settings
-MAX_TASK_RETRIES=3
-TASK_TIMEOUT_SECONDS=600
-
 # ==============================================
 # Web Configuration
 # ==============================================

.gitignore

@@ -139,6 +139,7 @@ celerybeat.pid
 .venv
 env/
 venv/
+temp-venv/
 ENV/
 env.bak/
 venv.bak/

azure.yaml

@@ -22,18 +22,7 @@ services:
       path: ./Dockerfile
       context: ..
       target: ""
-      remoteBuild: true
-
-  # ContentFlow Worker - Container App
-  worker:
-    project: ./contentflow-worker
-    language: python
-    host: containerapp
-    docker:
-      path: ./Dockerfile
-      context: ..
-      target: ""
-      remoteBuild: true
+      remoteBuild: false
 
   # ContentFlow Web - Container App
   web:
@@ -44,23 +33,23 @@ services:
       path: ./Dockerfile
       context: ..
       target: ""
-      remoteBuild: true
+      remoteBuild: false
 
 # Global hooks
-# hooks:
-#   # Run before provisioning infrastructure
-#   preprovision:
-#     shell: sh
-#     run: ./infra/scripts/pre-provision.sh
-#     interactive: false
-#     continueOnError: false
+hooks:
+  # Import placeholder image to ACR before provisioning (needed for AILZ internal environments)
+  preprovision:
+    shell: sh
+    run: ./infra/scripts/pre-provision.sh
+    interactive: false
+    continueOnError: false
 
-#   # Run after infrastructure is provisioned
-#   # postprovision:
-#   #   shell: sh
-#   #   run: ./infra/scripts/post-provision.sh
-#   #   interactive: false
-#   #   continueOnError: false
+  # Run after infrastructure is provisioned
+  postprovision:
+    shell: sh
+    run: ./infra/scripts/post-provision.sh
+    interactive: false
+    continueOnError: false
 
 #   # # Run before deploying all services
 #   # predeploy:
@@ -69,9 +58,9 @@ services:
 #   #   interactive: false
 #   #   continueOnError: false
 
-#   # Run after all services are deployed
-#   postdeploy:
-#     shell: sh
-#     run: ./infra/scripts/post-deploy.sh
-#     interactive: true
-#     continueOnError: false
+  # Run after all services are deployed
+  postdeploy:
+    shell: sh
+    run: ./infra/scripts/post-deploy.sh
+    interactive: true
+    continueOnError: false

contentflow-api/app/dependencies.py

@@ -20,6 +20,7 @@
 
 
 EXECUTOR_CATALOG_FILE_PATH = f'{Path(__file__).parent.parent.parent}/contentflow-lib/executor_catalog.yaml'
+SEED_PIPELINES_DIR = f'{Path(__file__).parent.parent}/seed/pipelines'
 
 def get_config_provider(refresh: bool = False) -> ConfigurationProvider:
     """Get a singleton instance of ConfigurationProvider"""
@@ -54,10 +55,7 @@ def get_health_service():
                          cosmos_db_name=app_settings.COSMOS_DB_NAME,
                          cosmos_db_containers=app_settings.get_cosmos_db_containers(),
                          blob_storage_account=app_settings.BLOB_STORAGE_ACCOUNT_NAME,
-                         blob_storage_container=app_settings.BLOB_STORAGE_CONTAINER_NAME,
-                         storage_account_worker_queue_url=app_settings.STORAGE_ACCOUNT_WORKER_QUEUE_URL,
-                         storage_worker_queue_name=app_settings.STORAGE_WORKER_QUEUE_NAME,
-                         worker_engine_api_endpoint=app_settings.WORKER_ENGINE_API_ENDPOINT)
+                         blob_storage_container=app_settings.BLOB_STORAGE_CONTAINER_NAME)
 
 @ttl_cache(ttl=__cache_ttl)  # Cache for 10 minutes
 async def get_pipeline_service():
@@ -89,6 +87,20 @@ async def get_pipeline_execution_service():
     from app.services.pipeline_execution_service import PipelineExecutionService
     return PipelineExecutionService(await get_cosmos_client())
 
+
+async def get_ingest_service():
+    """Dependency to get IngestService (uses global blob storage singleton)"""
+    from app.settings import get_settings
+    from app.utils.blob_storage import get_blob_storage_service
+    from app.services.ingest_service import IngestService
+
+    app_settings = get_settings()
+    blob_service = await get_blob_storage_service(
+        account_name=app_settings.BLOB_STORAGE_ACCOUNT_NAME,
+        container_name=app_settings.BLOB_STORAGE_CONTAINER_NAME,
+    )
+    return IngestService(blob_service)
+
 async def initialize_executor_catalog():
     """Initialize executor catalog"""
     logger.info(f"{'>'*70}")
@@ -158,6 +170,116 @@ async def initialize_blob_storage():
         logger.exception(e)
         raise
 
+async def initialize_seed_pipelines():
+    """Seed pipelines from YAML files in seed/pipelines/ on startup.
+
+    For each YAML file, checks if a pipeline with the same name already exists.
+    If not, creates it. If it already exists, skips it.
+    """
+    import yaml
+    import glob
+
+    logger.info(f"{'>'*70}")
+    logger.info("contentflow.api: Seeding pipelines from seed/pipelines/...")
+    logger.info(f"{'-'*70}")
+    try:
+        pipeline_service = await get_pipeline_service()
+        seed_dir = str(SEED_PIPELINES_DIR)
+        yaml_files = glob.glob(f"{seed_dir}/*.yaml") + glob.glob(f"{seed_dir}/*.yml")
+
+        if not yaml_files:
+            logger.info("No pipeline YAML files found in seed/pipelines/. Skipping.")
+            logger.info(f"{'<'*70}")
+            return True
+
+        created = 0
+        skipped = 0
+        for yaml_path in yaml_files:
+            try:
+                with open(yaml_path, "r") as f:
+                    raw = yaml.safe_load(f)
+
+                pipeline_def = raw.get("pipeline", raw)
+                name = pipeline_def.get("name")
+                if not name:
+                    logger.warning(f"Skipping {yaml_path}: no 'name' field found.")
+                    continue
+
+                # Check if already exists
+                existing = await pipeline_service.get_pipeline_by_name(name)
+                if existing:
+                    logger.info(f"  Pipeline '{name}' already exists (id={existing.id}). Skipping.")
+                    skipped += 1
+                    continue
+
+                # Build the full YAML string for storage
+                with open(yaml_path, "r") as f:
+                    yaml_str = f.read()
+
+                # Parse executors from YAML into nodes for canvas visualization
+                executors = pipeline_def.get("executors", [])
+                nodes = []
+                for executor in executors:
+                    nodes.append({
+                        "id": executor.get("id", ""),
+                        "type": "executor",
+                        "position": executor.get("position", {"x": 0, "y": 0}),
+                        "data": {
+                            "label": executor.get("name", ""),
+                            "executor": {
+                                "id": executor.get("type", ""),
+                                "name": executor.get("name", ""),
+                                "category": "unknown",
+                                "description": executor.get("description", ""),
+                            },
+                            "config": {
+                                "name": executor.get("name", ""),
+                                "description": executor.get("description", ""),
+                                "settings": executor.get("settings", {}),
+                            },
+                        },
+                    })
+
+                # Parse edges from YAML for canvas connections
+                yaml_edges = pipeline_def.get("edges", [])
+                edges = []
+                for edge_def in yaml_edges:
+                    src = edge_def.get("from", "")
+                    tgt = edge_def.get("to", "")
+                    if isinstance(src, str) and isinstance(tgt, str):
+                        edges.append({
+                            "id": f"{src}-{tgt}-{len(edges)}",
+                            "source": src,
+                            "target": tgt,
+                            "type": "default",
+                            "animated": True,
+                        })
+
+                pipeline_data = {
+                    "name": name,
+                    "description": pipeline_def.get("description", ""),
+                    "yaml": yaml_str,
+                    "nodes": nodes,
+                    "edges": edges,
+                    "tags": pipeline_def.get("tags", []),
+                    "enabled": True,
+                }
+
+                result = await pipeline_service.create_pipeline(pipeline_data)
+                logger.info(f"  Created pipeline '{name}' with id={result.id}")
+                created += 1
+
+            except Exception as e:
+                logger.warning(f"  Failed to seed pipeline from {yaml_path}: {e}")
+
+        logger.info(f"Pipeline seeding complete: {created} created, {skipped} skipped.")
+        logger.info(f"{'<'*70}")
+        return True
+    except Exception as e:
+        logger.error(f"Error during pipeline seeding: {str(e)}")
+        logger.exception(e)
+        raise
+
 async def close_all():
     """Close all dependencies"""
     global __cosmos_client

contentflow-api/app/models/__init__.py

@@ -17,6 +17,7 @@
     VaultExecution,
     VaultCrawlCheckpoint,
 )
+from ._ingest import IngestPayload, IngestResponse, IngestResultsResponse
 
 try:
     __version__ = importlib.metadata.version(__name__)
@@ -39,4 +40,7 @@
             "VaultUpdateRequest",
             "VaultExecution",
             "VaultCrawlCheckpoint",
+            "IngestPayload",
+            "IngestResponse",
+            "IngestResultsResponse",
           ]

contentflow-api/app/models/_ingest.py

@@ -0,0 +1,33 @@
+"""
+Ingest API models for multipart document submission.
+"""
+from typing import Optional
+from pydantic import BaseModel
+
+
+class IngestPayload(BaseModel):
+    """Form fields submitted by the client that become ProvidedDetails.json"""
+    caseId: str
+    firstName: str
+    lastName: str
+    mailingAddress: str
+    dateOfBirth: str
+
+
+class IngestResponse(BaseModel):
+    """Response returned to client on successful ingest (202 Accepted)"""
+    execution_id: str
+    case_id: str
+    status: str
+    files_uploaded: int
+    blob_prefix: str
+    pipeline_id: str
+    pipeline_name: str
+
+
+class IngestResultsResponse(BaseModel):
+    """Response for GET /ingest/{execution_id}/results"""
+    execution_id: str
+    status: str
+    results: Optional[dict] = None
+    error: Optional[str] = None

contentflow-api/app/routers/__init__.py

@@ -4,6 +4,7 @@
 from .pipelines import router as pipelines_router
 from .executors import router as executors_router
 from .vaults import router as vaults_router
+from .ingest import router as ingest_router
 
 try:
     __version__ = importlib.metadata.version(__name__)
@@ -15,4 +16,5 @@
             "pipelines_router",
             "executors_router",
             "vaults_router",
+            "ingest_router",
           ]