BaseContentApiModule

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

python · ApiModules

Details

ID	`BaseContentApiModule`
Language	python
From Version	6.8.0
Docker Image	`demisto/fastapi:0.123.5.6111791`
Tags	infra server

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.

Source

BaseContentApiModule.py BaseContentApiModule_test.py BaseContentApiModule.yml

Download History Raw

"""Unit tests for BaseContentApiModule.

This module tests the base classes provided by BaseContentApiModule:
- ContentBaseModel: Pydantic model with user-friendly validation
- BaseParams: Base parameters with common connection settings
- BaseExecutionConfig: Centralized execution configuration
"""

import pytest
from pydantic import Field
from BaseContentApiModule import ContentBaseModel, BaseParams, BaseExecutionConfig
from CommonServerPython import DemistoException


class TestContentBaseModel:
    """Test ContentBaseModel validation and error formatting."""

    def test_valid_model(self):
        """Test that valid data creates a model successfully."""

        class TestModel(ContentBaseModel):
            name: str
            age: int

        model = TestModel(name="John", age=30)
        assert model.name == "John"
        assert model.age == 30

    def test_invalid_model_raises_demisto_exception(self):
        """Test that invalid data raises DemistoException with formatted errors."""

        class TestModel(ContentBaseModel):
            name: str
            age: int

        with pytest.raises(DemistoException) as exc_info:
            TestModel(name="John", age="invalid")

        error_message = str(exc_info.value)
        assert "Invalid Inputs:" in error_message
        assert "age:" in error_message

    def test_model_str_representation(self):
        """Test that model string representation uses aliases."""

        class TestModel(ContentBaseModel):
            internal_name: str = Field(alias="externalName")

        model = TestModel(externalName="test")
        assert "externalName" in str(model)

    def test_extra_fields_ignored(self):
        """Test that extra fields are ignored per Config."""

        class TestModel(ContentBaseModel):
            name: str

        model = TestModel(name="John", extra_field="ignored")
        assert model.name == "John"
        assert not hasattr(model, "extra_field")


class TestBaseParams:
    """Test BaseParams common connection settings."""

    def test_default_values(self):
        """Test that default values are set correctly."""
        params = BaseParams()
        assert params.insecure is False
        assert params.proxy is False
        assert params.verify is True

    def test_verify_property(self):
        """Test that verify property returns inverse of insecure."""
        params_secure = BaseParams(insecure=False)
        assert params_secure.verify is True

        params_insecure = BaseParams(insecure=True)
        assert params_insecure.verify is False

    def test_inheritance(self):
        """Test that BaseParams can be inherited."""

        class CustomParams(BaseParams):
            api_key: str

        params = CustomParams(api_key="test-key", insecure=True, proxy=True)
        assert params.api_key == "test-key"
        assert params.insecure is True
        assert params.proxy is True
        assert params.verify is False


class TestBaseExecutionConfig:
    """Test BaseExecutionConfig centralized configuration."""

    def test_command_property(self, mocker):
        """Test that command property returns the current command."""
        mocker.patch("demistomock.command", return_value="test-module")
        mocker.patch("demistomock.params", return_value={})
        mocker.patch("demistomock.args", return_value={})
        mocker.patch("demistomock.getLastRun", return_value={})

        config = BaseExecutionConfig()
        assert config.command == "test-module"

    def test_raw_params_stored(self, mocker):
        """Test that raw params are stored correctly."""
        test_params = {"url": "https://api.example.com", "api_key": "test"}
        mocker.patch("demistomock.command", return_value="test-module")
        mocker.patch("demistomock.params", return_value=test_params)
        mocker.patch("demistomock.args", return_value={})
        mocker.patch("demistomock.getLastRun", return_value={})

        config = BaseExecutionConfig()
        assert config._raw_params == test_params

    def test_raw_args_stored(self, mocker):
        """Test that raw args are stored correctly."""
        test_args = {"limit": 10, "severity": "high"}
        mocker.patch("demistomock.command", return_value="my-command")
        mocker.patch("demistomock.params", return_value={})
        mocker.patch("demistomock.args", return_value=test_args)
        mocker.patch("demistomock.getLastRun", return_value={})

        config = BaseExecutionConfig()
        assert config._raw_args == test_args

    def test_last_run_for_fetch_commands(self, mocker):
        """Test that last_run is retrieved for fetch commands."""
        test_last_run = {"offset": 100}
        mocker.patch("demistomock.command", return_value="fetch-incidents")
        mocker.patch("demistomock.params", return_value={})
        mocker.patch("demistomock.args", return_value={})
        mocker.patch("demistomock.getLastRun", return_value=test_last_run)

        config = BaseExecutionConfig()
        assert config._raw_last_run == test_last_run

    def test_assets_last_run_for_fetch_assets(self, mocker):
        """Test that assets_last_run is retrieved for fetch-assets command."""
        test_assets_last_run = {"stage": "assets", "offset": 50}
        mocker.patch("demistomock.command", return_value="fetch-assets")
        mocker.patch("demistomock.params", return_value={})
        mocker.patch("demistomock.args", return_value={})
        mocker.patch("demistomock.getLastRun", return_value={})
        mocker.patch("demistomock.getAssetsLastRun", return_value=test_assets_last_run)

        config = BaseExecutionConfig()
        assert config._raw_assets_last_run == test_assets_last_run

BaseContentApiModule

Details

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

Source