BaseContentApiModule

Base classes for building robust Cortex integrations with user-friendly validation, common connection settings, and centralized execution configuration.

python · ApiModules

Source

"""Base Content API Module

This module provides base classes for building robust Cortex integrations with:
- User-friendly validation error formatting via Pydantic models
- Common connection settings (proxy, SSL verification)
- Centralized execution configuration to minimize redundant system calls

Classes:
    ContentBaseModel: Base Pydantic model with enhanced validation error formatting
    BaseParams: Base class for integration parameters with common connection settings
    BaseExecutionConfig: Centralized entry point holding command, params, args, and last_run
"""

from pydantic import BaseModel, ValidationError, Extra  # pylint: disable=no-name-in-module

import demistomock as demisto
from CommonServerPython import *


class ContentBaseModel(BaseModel):
    """Base Pydantic model with user-friendly validation error formatting.

    INTEGRATION DEVELOPER TIP:
    This base class enhances Pydantic's default validation by:
    1. Catching ValidationError exceptions
    2. Formatting them in a user-friendly way
    3. Raising `DemistoException` with clear error messages

    All parameter and argument models should inherit from this class to provide
    consistent, readable error messages to users when validation fails.
    """

    def __init__(self, **data):
        try:
            super().__init__(**data)
        except ValidationError as e:
            # Format errors in a user-friendly way
            error_messages = []
            for error in e.errors():
                field = error["loc"][0] if error["loc"] else "unknown"
                msg = error["msg"]
                error_messages.append(f"- {field}: {msg}")

            raise DemistoException("Invalid Inputs:\n" + "\n".join(error_messages)) from e

    def __str__(self):
        return str(self.dict(by_alias=True))

    def __repr__(self):
        return str(self.dict(by_alias=True))

    class Config:
        extra = Extra.ignore
        allow_population_by_field_name = True


class BaseParams(ContentBaseModel):
    """Base class for integration parameters with common connection settings.

    INTEGRATION DEVELOPER TIP:
    This class provides common parameters that most integrations need:
    - insecure: Whether to skip SSL certificate verification
    - proxy: Whether to use system proxy settings
    - verify: Computed property that returns the inverse of insecure

    Your integration's parameter class should inherit from this to get these
    common settings automatically.
    """

    insecure: bool = False
    proxy: bool = False

    @property
    def verify(self) -> bool:
        """Return SSL verification setting (inverse of insecure).

        Returns:
            bool: True if SSL certificates should be verified, False otherwise.
        """
        return not self.insecure


class BaseExecutionConfig:
    """Centralized entry point for the integration that holds command, params, args, and last_run.

    This class encapsulates all the information needed to execute a command, including:
    - command: The command being executed
    - params: Integration parameters (validated via Pydantic)
    - args: Command-specific arguments (different per command, validated via Pydantic)
    - last_run: State from when the previous fetch ended (for fetch commands)
    """

    def __init__(self):
        # INTEGRATION DEVELOPER TIP:
        # Centralize all your `demisto` class usages in the `ExecutionConfig`
        # class constructor and create an instance of the class *once* in the
        # `main` function to avoid redundant system calls. Access the required
        # configurations as validated and type-safe properties.
        self._raw_command: str = demisto.command()
        self._raw_params: dict = demisto.params()
        self._raw_args: dict = demisto.args()
        self._raw_last_run: dict = demisto.getLastRun() if self._raw_command in FETCH_COMMANDS else {}
        self._raw_assets_last_run: dict = demisto.getAssetsLastRun() if self._raw_command == "fetch-assets" else {}

    @property
    def command(self) -> str:
        """Get the current command being executed.

        Returns:
            str: The command name (e.g., 'test-module', 'ip', 'fetch-incidents').
        """
        return self._raw_command

README

BaseContentApiModule

Base classes for building robust Cortex integrations with user-friendly validation, common connection settings, and centralized execution configuration.

Overview
Key Features
Installation
Quick Start
Classes
Complete Integration Example
API Reference

Overview

This API module provides foundational classes that integrations can use to:

Validate configuration parameters and command arguments with user-friendly error messages
Manage common connection settings (proxy, SSL verification).
Centralize execution configuration to minimize redundant system calls.

These classes work seamlessly with ContentClient from ContentClientApiModule to provide a complete foundation for building production-ready integrations.

Key Features

Feature	Description
User-Friendly Validation	Pydantic-based validation with clear, actionable error messages.
Common Connection Settings	Standardized proxy and SSL verification parameters.
Centralized Configuration	Single entry point for command, params, args, and last_run.
Type Safety	Full type hints and validation for integration parameters.
Minimal Boilerplate	Code streamlining with reduced redundant `demisto` class calls.

Installation

Import the module in your integration:

from BaseContentApiModule import *

Quick Start

Basic Example

from BaseContentApiModule import ContentBaseModel, BaseParams, BaseExecutionConfig
from pydantic import AnyUrl

# Define your integration parameters
class MyIntegrationParams(BaseParams):
    url: AnyUrl
    api_key: str
    max_fetch: int = 50

# Define command arguments
class MyCommandArgs(ContentBaseModel):
    limit: int = 10
    severity: str

# Create execution configuration
class MyExecutionConfig(BaseExecutionConfig):
    @property
    def params(self) -> MyIntegrationParams:
        return MyIntegrationParams(**self._raw_params)
    
    @property
    def my_command_args(self) -> MyCommandArgs:
        return MyCommandArgs(**self._raw_args)

# Use in main function
def main():
    execution = MyExecutionConfig()
    params = execution.params  # Validated parameters
    
    if execution.command == "my-command":
        args = execution.my_command_args  # Validated arguments
        # Execute command logic...

Classes

ContentBaseModel

Base Pydantic model with enhanced validation error formatting.

Features

Catches ValidationError exceptions from Pydantic
Formats validation errors in a user-friendly way
Raises DemistoException with clear, readable error messages
Ignores extra fields automatically
Supports field aliases for parameter name mapping

Usage

from CommonServerPython import *
from BaseContentApiModule import ContentBaseModel
from pydantic import Field


class MyArgs(ContentBaseModel):
    """Arguments for my-command."""
    
    name: str
    age: int
    email: str | None = None
    is_active: bool = Field(default=True, alias="isActive")

# Valid usage
args = MyArgs(name="John", age=30, isActive=False)

# Invalid usage - raises DemistoException with clear message
try:
    args = MyArgs(name="John", age="invalid")
except DemistoException as e:
    # Error message: "Invalid Inputs:\n- age: value is not a valid integer"
    print(str(e))

String Representation

args = MyArgs(name="John", age=30, isActive=False)
print(str(args))  # Uses aliases: {'name': 'John', 'age': 30, 'isActive': False}

BaseParams

Base class for integration parameters with common connection settings.

Attributes

Attribute	Type	Default	Description
`insecure`	`bool`	`False`	Whether to skip SSL certificate verification
`proxy`	`bool`	`False`	Whether to use system proxy settings
`verify`	`bool` (property)	`not insecure`	SSL verification setting (computed)

Usage

from BaseContentApiModule import BaseParams
from pydantic import AnyUrl, Field

class MyIntegrationParams(BaseParams):
    """Integration parameters with validation.
    
    Attributes:
        url: API base URL (trailing slash removed automatically).
        api_key: API key for authentication.
        max_fetch: Maximum incidents per fetch.
    """
    # proxy and insecure are already defined in BaseParams
    url: AnyUrl
    api_key: str
    max_fetch: int = Field(default=50, ge=1, le=1000)
    
    @validator('url', allow_reuse=True)
    def clean_url(cls, v):
        """Remove trailing slash from URL."""
        return v.rstrip('/')

# Usage
params = MyIntegrationParams(
    url="https://api.example.com/",
    api_key="secret-key",
    max_fetch=100,
    insecure=True,
    proxy=False
)

print(params.url)     # "https://api.example.com" (trailing slash removed)
print(params.verify)  # False (inverse of insecure)

BaseExecutionConfig

Centralized entry point for integration execution that holds commands, parameters, arguments, and last_run.

Features

Contains all the information needed to execute a command.
Centralizes demisto class usages to avoid redundant system calls.
Provides type-safe access to configuration via properties.
Automatically retrieves last_run for fetch commands.
Supports both regular fetch and fetch-assets commands.

Attributes

Attribute	Type	Description
`_raw_command`	`str`	The command being executed.
`_raw_params`	`dict`	Raw integration parameters dictionary.
`_raw_args`	`dict`	Raw command arguments dictionary.
`_raw_last_run`	`dict`	State from previous fetch execution.
`_raw_assets_last_run`	`dict`	State from previous fetch-assets execution.

Usage

from BaseContentApiModule import BaseExecutionConfig, BaseParams, ContentBaseModel

class MyIntegrationParams(BaseParams):
    url: str
    api_key: str

class MyCommandArgs(ContentBaseModel):
    limit: int = 10

class MyLastRun(ContentBaseModel):
    offset: int = 0
    last_id: str | None = None

class MyExecutionConfig(BaseExecutionConfig):
    """Centralized execution configuration for MyIntegration."""
    
    @property
    def params(self) -> MyIntegrationParams:
        """Get validated integration parameters."""
        return MyIntegrationParams(**self._raw_params)
    
    @property
    def my_command_args(self) -> MyCommandArgs:
        """Get validated arguments for my-command."""
        return MyCommandArgs(**self._raw_args)
    
    @property
    def last_run(self) -> MyLastRun:
        """Get validated last run state for fetch commands."""
        return MyLastRun(**self._raw_last_run)

# Use in main function
def main():
    execution = MyExecutionConfig()
    
    # Access command name
    command = execution.command
    
    # Access validated parameters (only called once)
    params = execution.params
    
    # Route to command functions
    if command == "my-command":
        args = execution.my_command_args
        # Execute command...
    elif command == "fetch-incidents":
        last_run = execution.last_run
        # Execute fetch...

Complete Integration Example

The following is an end-to-end example of a production-ready integration using the BaseContentApiModule with ContentClient:

import demistomock as demisto
from CommonServerPython import *

from BaseContentApiModule import *
from ContentClientApiModule import *

from pydantic import AnyUrl, Field, validator


# ===== Parameters =====

class Credentials(ContentBaseModel):
    password: str


class MyIntegrationParams(BaseParams):
    """Integration parameters with validation."""
    
    url: AnyUrl
    credentials: Credentials
    max_fetch: int = Field(default=50, ge=1, le=1000)
    severity: str = "high"
    
    @validator('url', allow_reuse=True)
    def clean_url(cls, v):
        """Remove trailing slash from URL."""
        return v.rstrip('/')


# ===== Command Arguments =====

class GetAlertsArgs(ContentBaseModel):
    """Arguments for get-alerts command."""
    
    limit: int = Field(default=10, ge=1, le=100)
    severity: str | None = None


class LastRun(ContentBaseModel):
    """State management for fetch-incidents."""
    
    offset: int = 0
    last_id: str | None = None
    
    def set(self):
        """Save state for next fetch."""
        demisto.setLastRun(self.dict(by_alias=True))


# ===== Execution Configuration =====

class MyExecutionConfig(BaseExecutionConfig):
    """Centralized execution configuration."""
    
    @property
    def params(self) -> MyIntegrationParams:
        return MyIntegrationParams(**self._raw_params)
    
    @property
    def get_alerts_args(self) -> GetAlertsArgs:
        return GetAlertsArgs(**self._raw_args)
    
    @property
    def last_run(self) -> LastRun:
        return LastRun(**self._raw_last_run)


# ===== Client =====

class MyIntegrationClient(ContentClient):
    """Client for MyIntegration API."""
    
    def __init__(self, params: MyIntegrationParams):
        super().__init__(
            base_url=params.url,
            verify=params.verify,
            proxy=params.proxy,
            auth_handler=BearerTokenAuthHandler(token=params.credentials.password),
            retry_policy=RetryPolicy(max_attempts=3),
            diagnostic_mode=is_debug_mode(),
            client_name="MyIntegration"
        )
    
    def get_alerts(self, limit: int, severity: str | None = None) -> list[dict]:
        """Fetch alerts from the API."""
        params = {"limit": limit}
        if severity:
            params["severity"] = severity
        return self.get("/api/v1/alerts", params=params)


# ===== Commands =====

def get_alerts_command(client: MyIntegrationClient, args: GetAlertsArgs) -> CommandResults:
    """Execute get-alerts command."""
    alerts = client.get_alerts(limit=args.limit, severity=args.severity)
    
    return CommandResults(
        outputs_prefix="MyIntegration.Alert",
        outputs_key_field="id",
        outputs=alerts,
        readable_output=tableToMarkdown("Alerts", alerts)
    )

def test_module(client: MyIntegrationClient) -> str:
    """Test the integration connection."""
    try:
        client.get_alerts(limit=1)
        return "ok"
    except Exception as e:
        return f"Test failed: {str(e)}"


# ===== Main =====

def main():
    execution = MyExecutionConfig()
    command = execution.command
    
    try:
        params = execution.params
        client = MyIntegrationClient(params)
        
        if command == "test-module":
            return_results(test_module(client))
        
        elif command == "my-integration-get-alerts":
            args = execution.get_alerts_args
            return_results(get_alerts_command(client, args))

        else:
            raise NotImplementedError(f"Command {command} not implemented")
    
    except Exception as e:
        demisto.error(f"Error executing {command}: {str(e)}")
        return_error(str(e))


if __name__ in ("__main__", "__builtin__", "builtins"):
    main()

API Reference

ContentBaseModel

Base Pydantic model with user-friendly validation error formatting.

Methods

Method	Description
`__init__(**data)`	Initialize model with validation.
`__str__()`	String representation using aliases.
`__repr__()`	Representation using aliases.
`dict(by_alias=True)`	Convert to dictionary.

Configuration

Setting	Value	Description
`extra`	`Extra.ignore`	Ignore extra fields not defined in model.
`allow_population_by_field_name`	`True`	Allow both field names and aliases.

Example

from BaseContentApiModule import ContentBaseModel
from pydantic import Field, validator

class MyModel(ContentBaseModel):
    name: str
    age: int = Field(ge=0, le=150)
    email: str | None = None
    is_active: bool = Field(default=True, alias="isActive")
    
    @validator('email')
    def validate_email(cls, v):
        if v and '@' not in v:
            raise ValueError('Invalid email format')
        return v

# Valid usage
model = MyModel(name="John", age=30, isActive=False)

# Invalid usage - clear error message
try:
    model = MyModel(name="John", age=200)
except DemistoException as e:
    # Error: "Invalid Inputs:\n- age: ensure this value is less than or equal to 150"
    pass

BaseParams

Base class for integration parameters with common connection settings.

Attributes

Attribute	Type	Default	Description
`insecure`	`bool`	`False`	Skip SSL certificate verification.
`proxy`	`bool`	`False`	Use system proxy settings.
`verify`	`bool` (property)	`not insecure`	SSL verification (computed).

Example

from BaseContentApiModule import BaseParams
from pydantic import AnyUrl, Field, validator

class MyIntegrationParams(BaseParams):
    """Integration parameters with validation.
    
    Attributes:
        url: API base URL.
        api_key: API key for authentication.
        max_fetch: Maximum incidents per fetch (1-1000).
    """
    url: AnyUrl
    api_key: str
    max_fetch: int = Field(default=50, ge=1, le=1000)
    
    @validator('url', allow_reuse=True)
    def clean_url(cls, v):
        """Remove trailing slash from URL."""
        return v.rstrip('/')
    
    @validator('max_fetch', allow_reuse=True)
    def validate_max_fetch(cls, v):
        """Ensure max_fetch is within limits."""
        if v > 1000:
            raise ValueError("max_fetch must not exceed 1000")
        return v

# Usage
params = MyIntegrationParams(
    url="https://api.example.com/",
    api_key="secret",
    max_fetch=100,
    insecure=True
)

print(params.url)     # "https://api.example.com"
print(params.verify)  # False

BaseExecutionConfig

Centralized entry point for integration execution.

Attributes

Attribute	Type	Description
`_raw_command`	`str`	Current command being executed.
`_raw_params`	`dict`	Raw integration parameters.
`_raw_args`	`dict`	Raw command arguments.
`_raw_last_run`	`dict`	State from previous fetch.
`_raw_assets_last_run`	`dict`	State from previous the fetch-assets execution.

Properties

Property	Type	Description
`command`	`str`	The current command name.

Example

from BaseContentApiModule import BaseExecutionConfig, BaseParams, ContentBaseModel

class MyParams(BaseParams):
    url: str
    api_key: str

class MyCommandArgs(ContentBaseModel):
    limit: int = 10
    severity: str | None = None

class MyLastRun(ContentBaseModel):
    offset: int = 0

class MyExecutionConfig(BaseExecutionConfig):
    """Centralized configuration for MyIntegration."""
    
    @property
    def params(self) -> MyParams:
        """Get validated integration parameters."""
        return MyParams(**self._raw_params)
    
    @property
    def my_command_args(self) -> MyCommandArgs:
        """Get validated arguments for my-command."""
        return MyCommandArgs(**self._raw_args)
    
    @property
    def last_run(self) -> MyLastRun:
        """Get validated last run state."""
        return MyLastRun(**self._raw_last_run)

# Usage in main function
def main():
    execution = MyExecutionConfig()
    
    # Single call to demisto.command()
    command = execution.command
    
    # Single call to demisto.params() with validation
    params = execution.params
    
    # Route to commands
    if command == "my-command":
        # Single call to demisto.args() with validation
        args = execution.my_command_args
        # Execute command...
    
    elif command == "fetch-incidents":
        # Single call to demisto.getLastRun() with validation
        last_run = execution.last_run
        # Execute fetch...

Dependencies

pydantic: Data validation and settings management.
CommonServerPython: Common integration utilities.
demistomock: Local debugging support.

Best Practices

Inherit from ContentBaseModel for parameter and argument classes to get user-friendly error messages.
Inherit from BaseParams to automatically include common connection settings in integration parameters.
Create one ExecutionConfig instance in main() to minimize redundant system calls.
Define properties in the ExecutionConfig subclass for each command’s arguments and for fetch last run.
Use validators to add custom validation logic and data cleaning.
Document your models** with docstrings and type hints for better IDE and linter support.

Integration Example

See the Hello World v2 integration for a complete example of using these base classes in a production integration.

Related Modules

ContentClientApiModule: High-performance HTTP client with retry logic, rate limiting, and authentication.
CommonServerPython: Core utilities and helper functions for Cortex integrations.

BaseContentApiModule

Source

README

BaseContentApiModule

Table of Contents

Overview

Key Features

Installation

Quick Start

Basic Example

Classes

ContentBaseModel

Features

Usage

String Representation

BaseParams

Attributes

Usage

BaseExecutionConfig

Features

Attributes

Usage

Complete Integration Example

API Reference

ContentBaseModel

Methods

Configuration

Example

BaseParams

Attributes

Example

BaseExecutionConfig

Attributes

Properties

Example

Dependencies

Best Practices

Integration Example

Related Modules