New pipeline deployment system

.gitignore

@@ -159,3 +159,7 @@ cython_debug/
 #  and can be added to the global gitignore or merged into this file.  For a more nuclear
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 #.idea/
+
+
+# Pipelines default folder
+/pipelines

src/hayhooks/cli/__init__.py

@@ -3,6 +3,7 @@
 # SPDX-License-Identifier: Apache-2.0
 import click
 
+from hayhooks.cli.deploy_files import deploy_files
 from hayhooks.cli.run import run
 from hayhooks.cli.deploy import deploy
 from hayhooks.cli.status import status
@@ -22,3 +23,4 @@ def hayhooks(ctx, server, disable_ssl):
 hayhooks.add_command(deploy)
 hayhooks.add_command(status)
 hayhooks.add_command(undeploy)
+hayhooks.add_command(deploy_files)

src/hayhooks/cli/deploy_files/__init__.py

@@ -0,0 +1,34 @@
+import click
+import requests
+from pathlib import Path
+from urllib.parse import urljoin
+from hayhooks.server.utils.deploy_utils import read_pipeline_files_from_folder
+
+
+@click.command()
+@click.pass_obj
+@click.option('-n', '--name', required=True, help="Name of the pipeline to deploy")
+@click.argument('folder', type=click.Path(exists=True, file_okay=False, dir_okay=True, path_type=Path))
+def deploy_files(server_conf, name, folder):
+    """Deploy all pipeline files from a folder to the Hayhooks server."""
+    server, disable_ssl = server_conf
+
+    files_dict = {}
+    try:
+        files_dict = read_pipeline_files_from_folder(folder)
+
+        if not files_dict:
+            click.echo("Error: No valid files found in the specified folder")
+            return
+
+        resp = requests.post(
+            urljoin(server, "deploy_files"), json={"name": name, "files": files_dict}, verify=not disable_ssl
+        )
+
+        if resp.status_code >= 400:
+            click.echo(f"Error deploying pipeline: {resp.json().get('detail')}")
+        else:
+            click.echo(f"Pipeline successfully deployed with name: {resp.json().get('name')}")
+
+    except Exception as e:
+        click.echo(f"Error processing folder: {str(e)}")

src/hayhooks/server/app.py

@@ -1,16 +1,72 @@
-import logging
-import os
-import glob
 from fastapi import FastAPI
 from pathlib import Path
-from hayhooks.server.utils.deploy_utils import deploy_pipeline_def, PipelineDefinition
+from hayhooks.server.utils.deploy_utils import (
+    deploy_pipeline_def,
+    PipelineDefinition,
+    deploy_pipeline_files,
+    read_pipeline_files_from_folder,
+)
 from hayhooks.server.routers import status_router, draw_router, deploy_router, undeploy_router
 from hayhooks.settings import settings
+from hayhooks.server.logger import log
 
-logger = logging.getLogger("uvicorn.info")
+
+def deploy_yaml_pipeline(app: FastAPI, pipeline_file_path: Path) -> dict:
+    """
+    Deploy a pipeline from a YAML file.
+
+    Args:
+        app: FastAPI application instance
+        pipeline_file_path: Path to the YAML pipeline definition
+
+    Returns:
+        dict: Deployment result containing pipeline name
+    """
+    name = pipeline_file_path.stem
+    with open(pipeline_file_path, "r") as pipeline_file:
+        source_code = pipeline_file.read()
+
+    pipeline_definition = PipelineDefinition(name=name, source_code=source_code)
+    deployed_pipeline = deploy_pipeline_def(app, pipeline_definition)
+    log.info(f"Deployed pipeline from yaml: {deployed_pipeline['name']}")
+    return deployed_pipeline
+
+
+def deploy_files_pipeline(app: FastAPI, pipeline_dir: Path) -> dict:
+    """
+    Deploy a pipeline from a directory containing multiple files.
+
+    Args:
+        app: FastAPI application instance
+        pipeline_dir: Path to the pipeline directory
+
+    Returns:
+        dict: Deployment result containing pipeline name
+    """
+    name = pipeline_dir.name
+    files = read_pipeline_files_from_folder(pipeline_dir)
+
+    if files:
+        deployed_pipeline = deploy_pipeline_files(app, name, files)
+        log.info(f"Deployed pipeline from directory: {deployed_pipeline['name']}")
+        return deployed_pipeline
+    return {"name": name}
 
 
 def create_app() -> FastAPI:
+    """
+    Create and configure a FastAPI application.
+
+    This function initializes a FastAPI application with the following features:
+    - Configures root path from settings if provided
+    - Includes all router endpoints (status, draw, deploy, undeploy)
+    - Auto-deploys pipelines from the configured pipelines directory:
+        - YAML pipeline definitions (*.yml, *.yaml)
+        - Pipeline folders containing multiple files
+
+    Returns:
+        FastAPI: Configured FastAPI application instance
+    """
     if root_path := settings.root_path:
         app = FastAPI(root_path=root_path)
     else:
@@ -26,15 +82,30 @@ def create_app() -> FastAPI:
     pipelines_dir = settings.pipelines_dir
 
     if pipelines_dir:
-        logger.info(f"Pipelines dir set to: {pipelines_dir}")
-        for pipeline_file_path in glob.glob(f"{pipelines_dir}/*.y*ml"):
-            name = Path(pipeline_file_path).stem
-            with open(pipeline_file_path, "r") as pipeline_file:
-                source_code = pipeline_file.read()
-
-            pipeline_defintion = PipelineDefinition(name=name, source_code=source_code)
-            deployed_pipeline = deploy_pipeline_def(app, pipeline_defintion)
-            logger.info(f"Deployed pipeline: {deployed_pipeline['name']}")
+        log.info(f"Pipelines dir set to: {pipelines_dir}")
+        pipelines_path = Path(pipelines_dir)
+
+        yaml_files = list(pipelines_path.glob("*.y*ml"))
+        pipeline_dirs = [d for d in pipelines_path.iterdir() if d.is_dir()]
+
+        if yaml_files:
+            log.info(f"Deploying {len(yaml_files)} pipeline(s) from YAML files")
+            for pipeline_file_path in yaml_files:
+                try:
+                    deploy_yaml_pipeline(app, pipeline_file_path)
+                except Exception as e:
+                    log.warning(f"Skipping pipeline file {pipeline_file_path}: {str(e)}")
+                    continue
+
+        if pipeline_dirs:
+            log.info(f"Deploying {len(pipeline_dirs)} pipeline(s) from folders")
+            for pipeline_dir in pipeline_dirs:
+                try:
+                    deploy_files_pipeline(app, pipeline_dir)
+                except Exception as e:
+                    log.warning(f"Skipping pipeline folder {pipeline_dir}: {str(e)}")
+                    continue
+
     return app
 
 

src/hayhooks/server/exceptions.py

@@ -0,0 +1,20 @@
+class PipelineFilesError(Exception):
+    """Exception for errors saving pipeline files."""
+
+    pass
+
+
+class PipelineWrapperError(Exception):
+    """Exception for errors loading pipeline wrapper."""
+
+    pass
+
+
+class PipelineModuleLoadError(Exception):
+    """Exception for errors loading pipeline module."""
+
+
+class PipelineAlreadyExistsError(Exception):
+    """Exception for errors when a pipeline already exists."""
+
+    pass

src/hayhooks/server/logger.py

@@ -1,8 +1,18 @@
-# logger.py
-
 import os
 import sys
 from loguru import logger as log
 
+
+def formatter(record):
+    if record["extra"]:
+        return "<green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> | <level>{message}</level> - <magenta>{extra}</magenta>\n"
+
+    return "<green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> | <level>{message}</level>\n"
+
+
 log.remove()
-log.add(sys.stderr, level=os.getenv("LOG", "INFO").upper())
+log.add(
+    sys.stderr,
+    level=os.getenv("LOG", "INFO").upper(),
+    format=formatter
+)

src/hayhooks/server/pipelines/registry.py

@@ -1,31 +1,38 @@
-from typing import Optional
+from typing import Optional, Union
 
 from haystack import Pipeline
 from haystack.core.errors import PipelineError
+from hayhooks.server.utils.base_pipeline_wrapper import BasePipelineWrapper
+
+PipelineType = Union[Pipeline, BasePipelineWrapper]
 
 
 class _PipelineRegistry:
     def __init__(self) -> None:
-        self._pipelines: dict[str, Pipeline] = {}
+        self._pipelines: dict[str, PipelineType] = {}
 
-    def add(self, name: str, source: str) -> Pipeline:
+    def add(self, name: str, source_or_pipeline: Union[str, PipelineType]) -> PipelineType:
         if name in self._pipelines:
             msg = f"A pipeline with name {name} is already in the registry."
             raise ValueError(msg)
 
-        try:
-            self._pipelines[name] = Pipeline.loads(source)
-        except PipelineError as e:
-            msg = f"Unable to parse Haystack Pipeline {name}: {e}"
-            raise ValueError(msg) from e
+        if isinstance(source_or_pipeline, (Pipeline, BasePipelineWrapper)):
+            pipeline = source_or_pipeline
+        else:
+            try:
+                pipeline = Pipeline.loads(source_or_pipeline)
+            except PipelineError as e:
+                msg = f"Unable to parse Haystack Pipeline {name}: {e}"
+                raise ValueError(msg) from e
 
-        return self._pipelines[name]
+        self._pipelines[name] = pipeline
+        return pipeline
 
     def remove(self, name: str):
         if name in self._pipelines:
             del self._pipelines[name]
 
-    def get(self, name: str) -> Optional[Pipeline]:
+    def get(self, name: str) -> Optional[PipelineType]:
         return self._pipelines.get(name)
 
     def get_names(self) -> list[str]:

src/hayhooks/server/routers/deploy.py

@@ -1,9 +1,40 @@
-from fastapi import APIRouter, Request
-from hayhooks.server.utils.deploy_utils import deploy_pipeline_def, PipelineDefinition
+from fastapi import APIRouter, Request, HTTPException
+from pydantic import BaseModel
+from typing import Dict
+from hayhooks.server.exceptions import PipelineAlreadyExistsError
+from hayhooks.server.utils.deploy_utils import (
+    deploy_pipeline_def,
+    PipelineDefinition,
+    deploy_pipeline_files,
+    PipelineFilesError,
+    PipelineModuleLoadError,
+    PipelineWrapperError,
+)
 
 router = APIRouter()
 
 
+class PipelineFilesRequest(BaseModel):
+    name: str
+    files: Dict[str, str]
+
+
 @router.post("/deploy", tags=["config"])
 async def deploy(pipeline_def: PipelineDefinition, request: Request):
     return deploy_pipeline_def(request.app, pipeline_def)
+
+
+@router.post("/deploy_files", tags=["config"])
+async def deploy_files(pipeline_files: PipelineFilesRequest, request: Request):
+    try:
+        return deploy_pipeline_files(request.app, pipeline_files.name, pipeline_files.files)
+    except PipelineFilesError as e:
+        raise HTTPException(status_code=500, detail=str(e))
+    except PipelineModuleLoadError as e:
+        raise HTTPException(status_code=422, detail=str(e))
+    except PipelineWrapperError as e:
+        raise HTTPException(status_code=422, detail=str(e))
+    except PipelineAlreadyExistsError as e:
+        raise HTTPException(status_code=409, detail=str(e))
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Unexpected error deploying pipeline: {str(e)}")

src/hayhooks/server/routers/draw.py

@@ -3,13 +3,18 @@
 from fastapi import APIRouter, HTTPException
 from fastapi.responses import FileResponse
 from hayhooks.server.pipelines import registry
+from hayhooks.server.utils.base_pipeline_wrapper import BasePipelineWrapper
 
 router = APIRouter()
 
 
 @router.get("/draw/{pipeline_name}", tags=["config"])
 async def draw(pipeline_name):
     pipeline = registry.get(pipeline_name)
+
+    if isinstance(pipeline, BasePipelineWrapper):
+        pipeline = pipeline.pipeline
+
     if not pipeline:
         raise HTTPException(status_code=404)
 

src/hayhooks/server/utils/base_pipeline_wrapper.py

@@ -0,0 +1,46 @@
+from abc import ABC, abstractmethod
+from typing import List
+
+
+class BasePipelineWrapper(ABC):
+    def __init__(self):
+        self.pipeline = None
+
+    @abstractmethod
+    def setup(self) -> None:
+        """
+        Setup the pipeline.
+
+        This method should be called before using the pipeline.
+        """
+        pass
+
+    @abstractmethod
+    def run_api(self, urls: List[str], question: str) -> dict:
+        """
+        Run the pipeline in API mode.
+
+        Args:
+            urls: List of URLs to fetch content from
+            question: Question to be answered
+
+        Returns:
+            dict: Pipeline execution results
+        """
+        pass
+
+    @abstractmethod
+    def run_chat(self, user_message: str, model_id: str, messages: List[dict], body: dict) -> dict:
+        """
+        Run the pipeline in chat mode.
+
+        Args:
+            user_message: Message from the user
+            model_id: ID of the model to use
+            messages: List of previous messages
+            body: Additional request body parameters
+
+        Returns:
+            dict: Pipeline execution results
+        """
+        pass