jupyter-naas abi .cursorrules file for Python (stars: 46)

// User Intent: Create new integration

Required input: IntegrationName (PascalCase if not convert to it)

Template to use:

```python
from lib.abi.integration.integration import Integration, IntegrationConfiguration
from dataclasses import dataclass

@dataclass
class YourIntegrationConfiguration(IntegrationConfiguration):
    """Configuration for YourIntegration.
    
    Attributes:
        attribute_1 (str): Description of attribute_1
        attribute_2 (int): Description of attribute_2
    """
    attribute_1: str
    attribute_2: int

class YourIntegration(Integration):
    """YourIntegration class for interacting with YourService.
    
    This class provides methods to interact with YourService's API endpoints.
    It handles authentication and request management.
    
    Attributes:
        __configuration (YourIntegrationConfiguration): Configuration instance
            containing necessary credentials and settings.
    
    Example:
        >>> config = YourIntegrationConfiguration(
        ...     attribute_1="value1",
        ...     attribute_2=42
        ... )
        >>> integration = YourIntegration(config)
    """

    __configuration: YourIntegrationConfiguration

    def __init__(self, configuration: YourIntegrationConfiguration):
        super().__init__(configuration)
        self.__configuration = configuration

```

Output: Create file in `src/integrations/IntegrationName.py`

// User Intent: Create new pipeline or Update existing pipeline

Required input: PipelineName (PascalCase if not convert to it)

Template to use:

```python
from abi.pipeline import Pipeline, PipelineConfiguration, PipelineParameters
from dataclasses import dataclass
from src.data.integrations import YourIntegration
from abi.utils.Graph import ABIGraph
from rdflib import Graph
from abi.services.ontology_store.OntologyStorePorts import IOntologyStoreService
from src import secret
from fastapi import APIRouter
from langchain_core.tools import StructuredTool

@dataclass
class YourPipelineConfiguration(PipelineConfiguration):
    """Configuration for YourPipeline.
    
    Attributes:
        integration (YourIntegration): The integration instance to use
        ontology_store (IOntologyStoreService): The ontology store service to use
        ontology_store_name (str): Name of the ontology store to use. Defaults to "yourstorename"
    """
    integration: YourIntegration
    ontology_store: IOntologyStoreService
    ontology_store_name: str = "yourstorename"

class YourPipelineParameters(PipelineParameters):
    """Parameters for YourPipeline execution.
    
    Attributes:
        parameter_1 (str): Description of parameter_1
        parameter_2 (int): Description of parameter_2
    """
    parameter_1: str
    parameter_2: int

class YourPipeline(Pipeline):
    __configuration: YourPipelineConfiguration
    
    def __init__(self, configuration: YourPipelineConfiguration):
        self.__configuration = configuration
        
    def as_tools(self) -> list[StructuredTool]:
        """Returns a list of LangChain tools for this pipeline.
        
        Returns:
            list[StructuredTool]: List containing the pipeline tool
        """
        return [StructuredTool(
            name="your_pipeline",
            description="Executes the pipeline with the given parameters",
            func=lambda **kwargs: self.run(YourPipelineParameters(**kwargs)),
            args_schema=YourPipelineParameters
        )]

    def as_api(self, router: APIRouter) -> None:
        """Adds API endpoints for this pipeline to the given router.
        
        Args:
            router (APIRouter): FastAPI router to add endpoints to
        """
        @router.post("/YourPipeline")
        def run(parameters: YourPipelineParameters):
            return self.run(parameters).serialize(format="turtle")

    def run(self, parameters: YourPipelineParameters) -> Graph:        
        graph = ABIGraph()
        
        # ... Add your code here
        
        self.__configuration.ontology_store.insert(self.__configuration.ontology_store_name, graph)
        return graph

```

Output: Create file in `src/data/pipelines/ToolNameIfRelevent/PipelineName.py`

// User Intent: Create new workflow

Required input: WorkflowName (PascalCase if not convert to it)

Template to use:

```python
from abi.workflow import Workflow, WorkflowConfiguration
from abi.workflow.workflow import WorkflowParameters
from src.integrations import YourIntegration, YourIntegrationConfiguration
from src import secret, config
from dataclasses import dataclass
from pydantic import BaseModel, Field
from typing import Optional, List, Dict
from abi import logger
from fastapi import APIRouter
from langchain_core.tools import StructuredTool
from typing import Any
@dataclass
class YourWorkflowConfiguration(WorkflowConfiguration):
    """Configuration for YourWorkflow.
    
    Attributes:
        integration_config (YourIntegrationConfiguration): Configuration for the integration
    """
    integration_config: YourIntegrationConfiguration

class YourWorkflowParameters(WorkflowParameters):
    """Parameters for YourWorkflow execution.
    
    Attributes:
        parameter_1 (str): Description of parameter_1
        parameter_2 (int): Description of parameter_2
    """
    parameter_1: str = Field(..., description="Description of parameter_1")
    parameter_2: int = Field(..., description="Description of parameter_2")

class YourWorkflow(Workflow):
    __configuration: YourWorkflowConfiguration
    
    def __init__(self, configuration: YourWorkflowConfiguration):
        self.__configuration = configuration
        self.__integration = YourIntegration(self.__configuration.integration_config)

    def run(self, parameters: YourWorkflowParameters) -> Any:
        # Add your workflow logic here
        return "Your result"

    def as_tools(self) -> list[StructuredTool]:
        """Returns a list of LangChain tools for this workflow.
        
        Returns:
            list[StructuredTool]: List containing the workflow tool
        """
        return [StructuredTool(
            name="your_workflow_name",
            description="Description of what your workflow does",
            func=lambda **kwargs: self.run(YourWorkflowParameters(**kwargs)),
            args_schema=YourWorkflowParameters
        )]

    def as_api(self, router: APIRouter) -> None:
        """Adds API endpoints for this workflow to the given router.
        
        Args:
            router (APIRouter): FastAPI router to add endpoints to
        """
        @router.post("/your_endpoint")
        def run_workflow(parameters: YourWorkflowParameters):
            return self.run(parameters)

```

Output: Create file in `src/workflows/WorkflowName.py`