Core Modules

HTTP Client Framework

The HTTP client framework provides both synchronous and asynchronous HTTP clients with built-in retry logic, circuit breakers, connection pooling, and statistics.

IHTTPClient (Protocol Interface)

class hfortix_core.http.IHTTPClient(*args, **kwargs)

Bases: Protocol

Protocol defining the interface for HTTP clients used by FortiOS endpoints.

This protocol allows any class implementing these methods to be used as an HTTP client, enabling: - Custom HTTP client implementations - Easier testing with mock/fake clients - Support for both sync and async clients - Extension by library users

Method signatures support both synchronous (returning dict) and asynchronous (returning Coroutine) implementations. The return type is Union to accommodate both modes.

Implementations: - HTTPClient: Synchronous implementation using httpx.Client - AsyncHTTPClient: Asynchronous implementation using httpx.AsyncClient

Note

All methods should handle vdom=False to skip VDOM parameter in requests. The raw_json parameter controls whether full API response is returned (True) or just the results section (False, default).

get(api_type, path, params=None, vdom=None, raw_json=False, unwrap_single=False, action=None)

Perform GET request to retrieve resource(s) from the API.

Parameters:

• api_type (str) – API category (e.g., ‘cmdb’, ‘monitor’, ‘log’, ‘service’)

• path (str) – Endpoint path (e.g., ‘firewall/address’,

• 'firewall/address/web-server')

• params (Optional[dict[str, Any]]) – Optional query parameters (filters, pagination, etc.)

• vdom (Union[str, bool, None]) – Virtual domain name, or False to skip VDOM parameter

• raw_json (bool) – If True, return full API response with metadata; if

• False

• results (return only)

• unwrap_single (bool) – If True and result is single-item list, return just the item

• action (Optional[str]) – Special action parameter (e.g., ‘schema’, ‘default’)

Returns:

API response (sync mode) or Coroutine[dict] (async mode)

Return type:

dict

Example (Sync):

result = client.get(“cmdb”, “firewall/address/web-server”)

Example (Async):

result = await client.get(“cmdb”, “firewall/address/web-server”)

post(api_type, path, data, params=None, vdom=None, raw_json=False)

Perform POST request to create new resource(s) in the API.

Parameters:

• api_type (str) – API category (e.g., ‘cmdb’, ‘monitor’, ‘log’, ‘service’)

• path (str) – Endpoint path (e.g., ‘firewall/address’)

• data (dict[str, Any]) – Resource data to create

• params (Optional[dict[str, Any]]) – Optional query parameters

• vdom (Union[str, bool, None]) – Virtual domain name, or False to skip VDOM parameter

• raw_json (bool) – If True, return full API response with metadata; if

• False

• results (return only)

Returns:

API response (sync mode) or Coroutine[dict] (async mode)

Return type:

dict

Example (Sync):

result = client.post(“cmdb”, “firewall/address”, data={“name”: “test”, “subnet”: “10.0.0.1/32”})

Example (Async):

result = await client.post(“cmdb”, “firewall/address”, data={…})

put(api_type, path, data, params=None, vdom=None, raw_json=False)

Perform PUT request to update existing resource in the API.

Parameters:

• api_type (str) – API category (e.g., ‘cmdb’, ‘monitor’, ‘log’, ‘service’)

• path (str) – Endpoint path with identifier (e.g.,

• 'firewall/address/web-server')

• data (dict[str, Any]) – Updated resource data

• params (Optional[dict[str, Any]]) – Optional query parameters

• vdom (Union[str, bool, None]) – Virtual domain name, or False to skip VDOM parameter

• raw_json (bool) – If True, return full API response with metadata; if

• False

• results (return only)

Returns:

API response (sync mode) or Coroutine[dict] (async mode)

Return type:

dict

Example (Sync):

result = client.put(“cmdb”, “firewall/address/web-server”, data={“subnet”: “10.0.0.2/32”})

Example (Async):

result = await client.put(“cmdb”, “firewall/address/web-server”, data={…})

delete(api_type, path, params=None, vdom=None, raw_json=False)

Perform DELETE request to remove resource from the API.

Parameters:

• api_type (str) – API category (e.g., ‘cmdb’, ‘monitor’, ‘log’, ‘service’)

• path (str) – Endpoint path with identifier (e.g.,

• 'firewall/address/web-server')

• params (Optional[dict[str, Any]]) – Optional query parameters

• vdom (Union[str, bool, None]) – Virtual domain name, or False to skip VDOM parameter

• raw_json (bool) – If True, return full API response with metadata; if

• False

• results (return only)

Returns:

API response (sync mode) or Coroutine[dict] (async mode)

Return type:

dict

Example (Sync):

result = client.delete(“cmdb”, “firewall/address/web-server”)

Example (Async):

result = await client.delete(“cmdb”, “firewall/address/web-server”)

close()

Close the HTTP client and release resources.

Return type:

Optional[Coroutine[Any, Any, None]]

Returns:

None (sync clients) or Coroutine[None] (async clients)

Optional method - not required for basic protocol compliance. Custom clients may implement this for resource cleanup.

Example (Sync):

client.close()

Example (Async):

await client.close()

get_connection_stats()

Get connection statistics.

Optional method - not required for basic protocol compliance. Returns statistics about HTTP connections if supported.

Return type:

dict[str, Any]

Returns:

Dictionary with connection pool metrics (if available)

get_operations()

Get audit log of all API operations.

Optional method - not required for basic protocol compliance. Only available when operation tracking is enabled.

Return type:

list[dict[str, Any]]

Returns:

List of all API operations with timestamps and details

get_write_operations()

Get audit log of write operations (POST/PUT/DELETE).

Optional method - not required for basic protocol compliance. Only available when operation tracking is enabled.

Return type:

list[dict[str, Any]]

Returns:

List of write operations with timestamps and details

get_retry_stats()

Get retry statistics and metrics.

Optional method - not required for basic protocol compliance. Only available when adaptive retry is enabled.

Return type:

dict[str, Any]

Returns:

Dictionary with retry statistics (retry counts, backoff times, etc.)

get_circuit_breaker_state()

Get current circuit breaker state and metrics.

Optional method - not required for basic protocol compliance. Only available when circuit breaker is enabled.

Return type:

dict[str, Any]

Returns:

Dictionary with circuit breaker state (open/closed, failure count, etc.)

get_health_metrics()

Get health metrics and performance indicators.

Optional method - not required for basic protocol compliance. Only available when adaptive retry is enabled.

Return type:

dict[str, Any]

Returns:

Dictionary with health metrics (response times, backpressure, etc.)

get_binary(api_type, path, params=None, vdom=None)

GET request returning binary data (for file downloads).

Parameters:

• api_type (str) – API category (e.g., ‘cmdb’, ‘monitor’, ‘log’)

• path (str) – Endpoint path

• params (Optional[dict[str, Any]]) – Optional query parameters

• vdom (Union[str, bool, None]) – Virtual domain name, or False to skip VDOM parameter

Return type:

Union[bytes, Coroutine[Any, Any, bytes]]

Returns:

Raw binary response data (bytes)

BaseHTTPClient (Base Client)

class hfortix_core.http.BaseHTTPClient(url, verify=True, vdom=None, max_retries=3, connect_timeout=10.0, read_timeout=300.0, circuit_breaker_threshold=5, circuit_breaker_timeout=60.0, max_connections=100, max_keepalive_connections=20, adaptive_retry=False, retry_strategy='exponential', retry_jitter=False)

Bases: object

Base class for HTTP clients with shared logic.

Provides: - Parameter validation - URL building - Retry statistics - Circuit breaker state management - Endpoint timeout configuration - Path normalization and encoding - Data sanitization

Parameters:

• url (str)

• verify (bool)

• vdom (Optional[str])

• max_retries (int)

• connect_timeout (float)

• read_timeout (float)

• circuit_breaker_threshold (int)

• circuit_breaker_timeout (float)

• max_connections (int)

• max_keepalive_connections (int)

• adaptive_retry (bool)

• retry_strategy (str)

• retry_jitter (bool)

__init__(url, verify=True, vdom=None, max_retries=3, connect_timeout=10.0, read_timeout=300.0, circuit_breaker_threshold=5, circuit_breaker_timeout=60.0, max_connections=100, max_keepalive_connections=20, adaptive_retry=False, retry_strategy='exponential', retry_jitter=False)

Initialize base HTTP client with shared configuration

Parameters:

• adaptive_retry (bool) – Enable adaptive retry with backpressure detection

• (default – False) When enabled, monitors response times and adjusts retry delays based on FortiGate health signals.

• retry_strategy (str) – Retry backoff strategy - ‘exponential’ (default) or ‘linear’. Exponential: 1s, 2s, 4s, 8s, 16s, 30s. Linear: 1s, 2s, 3s, 4s, 5s.

• retry_jitter (bool) – Add random jitter (0-25% of delay) to retry delays to prevent thundering herd problem when multiple clients retry simultaneously (default: False).

• url (str)

• verify (bool)

• vdom (str | None)

• max_retries (int)

• connect_timeout (float)

• read_timeout (float)

• circuit_breaker_threshold (int)

• circuit_breaker_timeout (float)

• max_connections (int)

• max_keepalive_connections (int)

Return type:

None

get_retry_stats()

Get retry statistics

Return type:

dict[str, Any]

get_circuit_breaker_state()

Get current circuit breaker state

Return type:

dict[str, Any]

configure_endpoint_timeout(endpoint_pattern, connect_timeout=None, read_timeout=None)

Configure custom timeout for specific endpoints

Return type:

None

Parameters:

• endpoint_pattern (str)

• connect_timeout (float | None)

• read_timeout (float | None)

reset_circuit_breaker()

Reset circuit breaker to closed state

Return type:

None

get_health_metrics()

Get comprehensive health metrics including adaptive retry stats

Return type:

dict[str, Any]

Returns:

Dictionary with health score, response times, circuit state, etc.

HTTPClient (Synchronous FortiOS)

class hfortix_core.http.HTTPClient(url, verify=True, token=None, username=None, password=None, vdom=None, max_retries=3, connect_timeout=10.0, read_timeout=300.0, user_agent=None, circuit_breaker_threshold=10, circuit_breaker_timeout=30.0, circuit_breaker_auto_retry=False, circuit_breaker_max_retries=3, circuit_breaker_retry_delay=5.0, max_connections=100, max_keepalive_connections=20, session_idle_timeout=300, read_only=False, track_operations=False, adaptive_retry=False, retry_strategy='exponential', retry_jitter=False, audit_handler=None, audit_callback=None, user_context=None)

Bases: BaseHTTPClient

Internal HTTP client for FortiOS API requests (Sync Implementation)

Implements the IHTTPClient protocol for synchronous HTTP operations.

Handles all HTTP communication with FortiGate devices including: - Session management - Authentication headers - SSL verification - Request/response handling - Error handling - Automatic retry with exponential backoff - Context manager support (use with ‘with’ statement)

Query Parameter Encoding:

The requests library automatically handles query parameter encoding: - Lists: Encoded as repeated parameters (e.g., [‘a’, ‘b’] → ?key=a&key=b) - Booleans: Converted to lowercase strings (‘true’/’false’) - None values: Should be filtered out before passing to params - Special characters: URL-encoded automatically

Path Encoding:

Paths are URL-encoded with / and % as safe characters to prevent double-encoding of already-encoded components.

Protocol Implementation:

This class implements the IHTTPClient protocol, allowing it to be used interchangeably with other HTTP client implementations (e.g., AsyncHTTPClient, custom user-provided clients).

This class is internal and not exposed to users directly, but users can provide their own IHTTPClient implementations to FortiOS.__init__().

Parameters:

• url (str)

• verify (bool)

• token (Optional[str])

• username (Optional[str])

• password (Optional[str])

• vdom (Optional[str])

• max_retries (int)

• connect_timeout (float)

• read_timeout (float)

• user_agent (Optional[str])

• circuit_breaker_threshold (int)

• circuit_breaker_timeout (float)

• circuit_breaker_auto_retry (bool)

• circuit_breaker_max_retries (int)

• circuit_breaker_retry_delay (float)

• max_connections (int)

• max_keepalive_connections (int)

• session_idle_timeout (Union[int, float, None])

• read_only (bool)

• track_operations (bool)

• adaptive_retry (bool)

• retry_strategy (str)

• retry_jitter (bool)

• audit_handler (Optional[Any])

• audit_callback (Optional[Any])

• user_context (Optional[dict[str, Any]])

__init__(url, verify=True, token=None, username=None, password=None, vdom=None, max_retries=3, connect_timeout=10.0, read_timeout=300.0, user_agent=None, circuit_breaker_threshold=10, circuit_breaker_timeout=30.0, circuit_breaker_auto_retry=False, circuit_breaker_max_retries=3, circuit_breaker_retry_delay=5.0, max_connections=100, max_keepalive_connections=20, session_idle_timeout=300, read_only=False, track_operations=False, adaptive_retry=False, retry_strategy='exponential', retry_jitter=False, audit_handler=None, audit_callback=None, user_context=None)

Initialize HTTP client

Parameters:

• url (str) – Base URL for API (e.g., “https://192.0.2.10”)

• verify (bool) – Verify SSL certificates

• token (Optional[str]) – API authentication token (if using token auth)

• username (Optional[str]) – Username for authentication (if using username/password

• auth)

• password (Optional[str]) – Password for authentication (if using username/password

• auth)

• vdom (Optional[str]) – Default virtual domain

• max_retries (int) – Maximum number of retry attempts on transient failures

• (default (all API calls) –

• connect_timeout (float) – Timeout for establishing connection in seconds

• (default – 10.0)

• read_timeout (float) – Timeout for reading response in seconds (default:

• 300.0)

• user_agent (Optional[str]) – Custom User-Agent header for identifying application in

• logs. (include in audit) – If None, defaults to ‘hfortix/{version}’. Useful for multi-team environments and troubleshooting in production.

• circuit_breaker_threshold (int) – Number of consecutive failures before

• (default –

• circuit_breaker_timeout (float) – Seconds to wait before transitioning to

• (default – 30.0)

• circuit_breaker_auto_retry (bool) – Enable automatic retry when circuit

• (default – False). When enabled, waits circuit_breaker_retry_delay seconds between retries instead of immediately raising exception. Useful for long-running automation scripts. NOT recommended for tests or interactive use.

• circuit_breaker_max_retries (int) – Maximum retry attempts when

• (default –

• circuit_breaker_retry_delay (float) – Delay in seconds between retry

• (default – 5.0). This is separate from circuit_breaker_timeout, which controls when the circuit transitions from open to half-open.

• max_connections (int) – Maximum number of connections in the pool

• (default –

• max_keepalive_connections (int) – Maximum number of keepalive connections

• (default –

• session_idle_timeout (Union[int, float, None]) – For username/password auth only. Idle timeout

• before (in seconds) – proactively re-authenticating (default: 300 = 5 minutes). This should match your FortiGate’s ‘config system global’ -> ‘remoteauthtimeout’ setting. Set to None to disable proactive re-authentication. Note: API token authentication is stateless and doesn’t use sessions.

• read_only (bool) – Enable read-only mode - simulate write operations

• (default – False)

• track_operations (bool) – Enable operation tracking - maintain audit log of

• (default – False)

• adaptive_retry (bool) – Enable adaptive retry with backpressure detection

• (default – False). When enabled, monitors response times and adjusts retry delays based on FortiGate health signals (slow responses, 503 errors). Increases retry delays when FortiGate is overloaded to prevent cascading failures.

• retry_strategy (str) – Retry backoff strategy - ‘exponential’ (default) or ‘linear’. Exponential: 1s, 2s, 4s, 8s, 16s, 30s. Linear: 1s, 2s, 3s, 4s, 5s. Use exponential for transient failures, linear for rate limiting.

• retry_jitter (bool) – Add random jitter (0-25% of delay) to retry delays to prevent thundering herd problem when multiple clients retry simultaneously (default: False).

• audit_handler (Optional[Any]) – Handler for audit logging (implements AuditHandler

• protocol). – Use built-in handlers: SyslogHandler, FileHandler, StreamHandler, CompositeHandler. Essential for compliance (SOC 2, HIPAA, PCI-DSS). Example: SyslogHandler(“siem.company.com:514”)

• audit_callback (Optional[Any]) – Custom callback function for audit logging. Alternative to audit_handler. Receives operation dict as parameter. Example: lambda op: send_to_kafka(op)

• user_context (Optional[dict[str, Any]]) – Optional dict with user/application context to

• logs. – Example: {“username”: “admin”, “app”: “automation”, “ticket”: “CHG-12345”}

Raises:

• ValueError – If parameters are invalid or both token and

• username/password provided –

Return type:

None

login()

Authenticate using username/password and obtain session token

This method is called automatically if username/password are provided during initialization. Can also be called manually to re-authenticate.

Raises:

• ValueError – If username/password not configured

• AuthenticationError – If login fails

Return type:

None

logout()

Logout and invalidate session token

This method is called automatically when using context manager (with statement). Can also be called manually to explicitly logout.

Note

Only applicable when using username/password authentication. Token-based authentication doesn’t require logout.

Return type:

None

get_connection_stats()

Get HTTP connection pool statistics

Returns:

Connection statistics including:

• http2_enabled: Whether HTTP/2 is enabled

• max_connections: Maximum number of connections allowed

• max_keepalive_connections: Maximum keepalive connections

• active_requests: Current number of active requests

• total_requests: Total requests made since initialization

• pool_exhaustion_count: Times pool reached capacity

• circuit_breaker_state: Current circuit breaker state

• consecutive_failures: Number of consecutive failures

Return type:

dict

Example

>>> stats = client.get_connection_stats()
>>> print(f"Circuit breaker: {stats['circuit_breaker_state']}")
>>> print(f"Active requests: {stats['active_requests']}")

set_transaction(transaction_id)

Set active transaction ID for automatic header injection.

When a transaction ID is set, all subsequent requests will automatically include the ‘X-TRANSACTION-ID’ header required by FortiOS batch transactions.

Parameters:

transaction_id (Optional[int]) – Transaction ID to use (None to clear)

Return type:

None

Examples

>>> # Start transaction
>>> client.set_transaction(19)
>>> # All requests now include X-TRANSACTION-ID: 19
>>>
>>> # Clear transaction
>>> client.set_transaction(None)

inspect_last_request()

Get details of last API request for debugging

Returns:

Request information including:

• method: HTTP method used

• endpoint: API endpoint path

• params: Query parameters

• response_time_ms: Response time in milliseconds

• status_code: HTTP status code

• error: Error message if no requests made

Return type:

dict

Example

>>> client.get("/api/v2/cmdb/firewall/address")
>>> info = client.inspect_last_request()
>>> print(f"Last request took {info['response_time_ms']:.2f}ms")

request(method, api_type, path, data=None, params=None, vdom=None, raw_json=False, request_id=None, silent=False)

Generic request method for all API calls

Parameters:

• method (str) – HTTP method (GET, POST, PUT, DELETE)

• api_type (str) – API type (cmdb, monitor, log, service)

• path (str) – API endpoint path (e.g., ‘firewall/address’, ‘system/status’)

• data (Optional[dict[str, Any]]) – Request body data (for POST/PUT)

• params (Optional[dict[str, Any]]) – Query parameters dict

• vdom (Union[str, bool, None]) – Virtual domain (None=use default, or specify vdom name)

• raw_json (bool) – If False (default), return only ‘results’ field. If True,

• response (return full)

• request_id (Optional[str]) – Optional correlation ID for tracking requests across

• logs

• silent (bool) – If True, suppress error logging (for exists() checks)

Returns:

If raw_json=False, returns response[‘results’] (or full response if no ‘results’ key)

System Message: ERROR/3 (/home/docs/checkouts/readthedocs.org/user_builds/hfortix-core/checkouts/latest/hfortix_core/http/client.py:docstring of hfortix_core.http.client.HTTPClient.request, line 26)

Unexpected indentation.

If raw_json=True, returns complete API response with status, http_status, etc.

Return type:

dict

get(api_type, path, params=None, vdom=None, raw_json=False, silent=False)

GET request

Parameters:

• api_type (str) – API type (cmdb, monitor, etc.)

• path (str) – Endpoint path

• params (Optional[dict[str, Any]]) – Query parameters

• vdom (Union[str, bool, None]) – Virtual domain

• raw_json (bool) – Return raw JSON response

• silent (bool) – If True, suppress error logging (for exists() checks)

Return type:

Union[dict[str, Any], Coroutine[Any, Any, dict[str, Any]]]

get_binary(api_type, path, params=None, vdom=None)

GET request returning binary data (for file downloads)

Parameters:

• api_type (str) – API type

• path (str) – Endpoint path

• params (Optional[dict[str, Any]]) – Query parameters

• vdom (Union[str, bool, None]) – Virtual domain

Return type:

bytes

Returns:

Raw binary response data

post(api_type, path, data, params=None, vdom=None, scope=None, raw_json=False)

POST request - Create new object

Parameters:

• scope (Optional[str]) – Optional scope parameter for global objects (‘global’ or ‘vdom’). Required when creating objects in global scope.

• api_type (str)

• path (str)

• data (dict[str, Any])

• params (dict[str, Any] | None)

• vdom (str | bool | None)

• raw_json (bool)

Return type:

Union[dict[str, Any], Coroutine[Any, Any, dict[str, Any]]]

put(api_type, path, data, params=None, vdom=None, scope=None, raw_json=False)

PUT request - Update existing object

Parameters:

• scope (Optional[str]) – Optional scope parameter for global objects (‘global’ or ‘vdom’). Required when updating objects in global scope.

• api_type (str)

• path (str)

• data (dict[str, Any])

• params (dict[str, Any] | None)

• vdom (str | bool | None)

• raw_json (bool)

Return type:

Union[dict[str, Any], Coroutine[Any, Any, dict[str, Any]]]

delete(api_type, path, params=None, vdom=None, scope=None, raw_json=False)

DELETE request - Delete object

Parameters:

• scope (Optional[str]) – Optional scope parameter for global objects (‘global’ or ‘vdom’). Required when deleting objects in global scope.

• api_type (str)

• path (str)

• params (dict[str, Any] | None)

• vdom (str | bool | None)

• raw_json (bool)

Return type:

Union[dict[str, Any], Coroutine[Any, Any, dict[str, Any]]]

static validate_mkey(mkey, parameter_name='mkey')

Validate and convert mkey to string

Parameters:

• mkey (Any) – The management key value to validate

• parameter_name (str) – Name of the parameter (for error messages)

Return type:

str

Returns:

String representation of mkey

Raises:

ValueError – If mkey is None, empty, or invalid

Example

>>> mkey = HTTPClient.validate_mkey(user_id, 'user_id')

static validate_required_params(params, required)

Validate that required parameters are present in params dict

Parameters:

• params (dict[str, Any]) – Dictionary of parameters to validate

• required (list[str]) – List of required parameter names

Raises:

ValueError – If any required parameters are missing

Return type:

None

Example

>>> HTTPClient.validate_required_params(data, ['name', 'type'])

static validate_range(value, min_val, max_val, parameter_name='value')

Validate that a numeric value is within a specified range

Parameters:

• value (Union[int, float]) – The value to validate

• min_val (Union[int, float]) – Minimum allowed value (inclusive)

• max_val (Union[int, float]) – Maximum allowed value (inclusive)

• parameter_name (str) – Name of the parameter (for error messages)

Raises:

ValueError – If value is outside the specified range

Return type:

None

Example

>>> HTTPClient.validate_range(port, 1, 65535, 'port')

static validate_choice(value, choices, parameter_name='value')

Validate that a value is one of the allowed choices

Parameters:

• value (Any) – The value to validate

• choices (list[Any]) – List of allowed values

• parameter_name (str) – Name of the parameter (for error messages)

Raises:

ValueError – If value is not in the allowed choices

Return type:

None

Example

>>> HTTPClient.validate_choice(protocol, ['tcp', 'udp'],
'protocol')

static build_params(**kwargs)

Build parameters dict, filtering out None values

Parameters:

**kwargs (Any) – Keyword arguments to build params from

Return type:

dict[str, Any]

Returns:

Dictionary with None values removed

Example

>>> params = HTTPClient.build_params(format=['name'],
datasource=True, other=None)
>>> # Returns: {'format': ['name'], 'datasource': True}

close()

Close the HTTP session and release resources

If using username/password authentication, this will also logout to properly clean up the session.

Return type:

None

get_operations()

Get audit log of all tracked API operations

Returns all tracked operations (GET/POST/PUT/DELETE) in chronological order. Only available when track_operations=True was passed to constructor.

Returns:

• timestamp: ISO 8601 timestamp

• method: HTTP method (GET/POST/PUT/DELETE)

• api_type: API type (cmdb/monitor/log/service)

• path: API endpoint path

• data: Request payload (for POST/PUT), None otherwise

• status_code: HTTP response status code

• vdom: Virtual domain (if specified)

• read_only_simulated: True if operation was simulated in

System Message: WARNING/2 (/home/docs/checkouts/readthedocs.org/user_builds/hfortix-core/checkouts/latest/hfortix_core/http/client.py:docstring of hfortix_core.http.client.HTTPClient.get_operations, line 15)

Bullet list ends without a blank line; unexpected unindent.

read-only mode

Return type:

List of operation dictionaries with keys

Example

>>> client = HTTPClient(url="https://192.0.2.10", token="...",
track_operations=True)
>>> client.post("cmdb", "/firewall/address", {"name": "test"})
>>> ops = client.get_operations()
>>> print(ops[0])
{
    'timestamp': '2024-12-20T10:30:15Z',
    'method': 'POST',
    'api_type': 'cmdb',
    'path': '/firewall/address',
    'data': {'name': 'test'},
    'status_code': 200,
    'vdom': 'root',
    'read_only_simulated': False
}

get_write_operations()

Get audit log of write operations only (POST/PUT/DELETE)

Filters tracked operations to return only write operations, excluding GET requests.

Return type:

list[dict[str, Any]]

Returns:

List of write operation dictionaries (same format as get_operations())

Example

>>> client = HTTPClient(url="https://192.0.2.10", token="...",
track_operations=True)
>>> client.get("cmdb", "/firewall/address/test")  # GET - excluded
>>> client.post("cmdb", "/firewall/address", {"name": "test2"})  #
POST - included
>>> client.delete("cmdb", "/firewall/address/test")  # DELETE -
included
>>> write_ops = client.get_write_operations()
>>> len(write_ops)  # Returns 2 (POST and DELETE only)
2

static make_exists_method(get_method)

Create an exists() helper that works with both sync and async modes.

This utility wraps a get() method and returns a function that: - Returns True if the object exists - Returns False if ResourceNotFoundError is raised - Returns False if response has error status (e.g., {‘status’: ‘error’}) - Works transparently with both sync and async clients

Parameters:

• get_method (Callable[..., Any]) – The get() method to wrap (bound method from endpoint

• instance)

Return type:

Callable[..., bool]

Returns:

A function that returns bool (sync) or Coroutine[bool] (async)

Example

class Address:

def __init__(self, client):

self._client = client

def get(self, name, **kwargs):

System Message: WARNING/2 (/home/docs/checkouts/readthedocs.org/user_builds/hfortix-core/checkouts/latest/hfortix_core/http/client.py:docstring of hfortix_core.http.client.HTTPClient.make_exists_method, line 22); backlink

Inline strong start-string without end-string.

return self._client.get(“cmdb”, f”/firewall/address/{name}”, **kwargs)

System Message: WARNING/2 (/home/docs/checkouts/readthedocs.org/user_builds/hfortix-core/checkouts/latest/hfortix_core/http/client.py:docstring of hfortix_core.http.client.HTTPClient.make_exists_method, line 23); backlink

Inline strong start-string without end-string.

# Create exists method using the helper exists = HTTPClient.make_exists_method(get)

HTTPClientFMG (Synchronous FortiManager)

class hfortix_core.HTTPClientFMG(url, username, password, verify=True, adom=None, max_retries=3, connect_timeout=10.0, read_timeout=300.0, circuit_breaker_threshold=5, circuit_breaker_timeout=60.0, max_connections=100, max_keepalive_connections=20, adaptive_retry=False, retry_strategy='exponential', retry_jitter=False)

Bases: BaseHTTPClient

HTTP client for FortiManager JSON-RPC API.

Provides session-based authentication and JSON-RPC request handling while reusing the retry logic, circuit breaker, connection pooling, and statistics from BaseHTTPClient.

FortiManager uses a different authentication model than FortiOS: - FortiOS: REST API with Bearer token in headers - FortiManager: JSON-RPC API with session token in request body

Example

>>> client = HTTPClientFMG(
...     url="https://fmg.example.com",
...     username="admin",
...     password="password",
... )
>>> client.login()
>>> response = client.execute("get", [{"url": "/dvmdb/device"}])
>>> client.logout()

Parameters:

• url (str)

• username (str)

• password (str)

• verify (bool)

• adom (Optional[str])

• max_retries (int)

• connect_timeout (float)

• read_timeout (float)

• circuit_breaker_threshold (int)

• circuit_breaker_timeout (float)

• max_connections (int)

• max_keepalive_connections (int)

• adaptive_retry (bool)

• retry_strategy (str)

• retry_jitter (bool)

__init__(url, username, password, verify=True, adom=None, max_retries=3, connect_timeout=10.0, read_timeout=300.0, circuit_breaker_threshold=5, circuit_breaker_timeout=60.0, max_connections=100, max_keepalive_connections=20, adaptive_retry=False, retry_strategy='exponential', retry_jitter=False)

Initialize FortiManager HTTP client.

Parameters:

• url (str) – Base URL for FMG (e.g., “https://fmg.example.com”)

• username (str) – Admin username

• password (str) – Admin password

• verify (bool) – Verify SSL certificates (default: True)

• adom (Optional[str]) – Default ADOM for operations

• max_retries (int) – Maximum retry attempts on transient failures

• connect_timeout (float) – Connection timeout in seconds

• read_timeout (float) – Read timeout in seconds

• circuit_breaker_threshold (int) – Failures before opening circuit

• circuit_breaker_timeout (float) – Seconds before retrying after circuit opens

• max_connections (int) – Maximum connection pool size

• max_keepalive_connections (int) – Maximum keepalive connections

• adaptive_retry (bool) – Enable adaptive retry with backpressure detection

• retry_strategy (str) – ‘exponential’ or ‘linear’ backoff

• retry_jitter (bool) – Add random jitter to retry delays

Return type:

None

property jsonrpc_url: str

JSON-RPC endpoint URL.

property is_authenticated: bool

Check if we have a valid session.

property adom: str | None

Default ADOM.

login()

Authenticate with FortiManager.

Return type:

dict[str, Any]

Returns:

FMG login response dict with session and status information

Raises:

RuntimeError – If authentication fails

logout()

End FortiManager session.

Return type:

dict[str, Any]

Returns:

FMG logout response dict with status information

execute(method, params, verbose=1)

Execute a FortiManager JSON-RPC request.

Parameters:

• method (Literal['exec', 'get', 'set', 'add', 'update', 'delete']) – JSON-RPC method

• params (list[dict[str, Any]]) – Request parameters

• verbose (int) – Verbosity level (0 or 1)

Return type:

dict[str, Any]

Returns:

FMG response dict

Raises:

RuntimeError – If not authenticated or request fails

proxy_request(action, resource, targets, payload=None, timeout=60)

Execute a FortiOS API call through the FMG proxy endpoint.

This is the core method for routing FortiOS REST API calls through FortiManager to managed devices.

Parameters:

• action (Literal['get', 'post', 'put', 'delete']) – HTTP method (get, post, put, delete)

• resource (str) – FortiOS API resource path (e.g., “/api/v2/cmdb/firewall/address”)

• targets (list[str]) – List of target devices/groups (e.g., [“adom/root/device/fw-01”])

• payload (dict[str, Any] | None) – Request body for POST/PUT

• timeout (int) – Request timeout in seconds

Return type:

dict[str, Any]

Returns:

FMG response dict containing results from each target device

close()

Close the session and HTTP client.

Return type:

None

get_health_metrics()

Get health metrics for monitoring.

Return type:

dict[str, Any]

get_connection_stats()

Get connection pool statistics.

Return type:

dict[str, Any]

AsyncHTTPClient (Asynchronous)

class hfortix_core.http.AsyncHTTPClient(url, verify=True, token=None, username=None, password=None, vdom=None, max_retries=3, connect_timeout=10.0, read_timeout=300.0, user_agent=None, circuit_breaker_threshold=10, circuit_breaker_timeout=30.0, circuit_breaker_auto_retry=False, circuit_breaker_max_retries=3, circuit_breaker_retry_delay=5.0, max_connections=100, max_keepalive_connections=20, session_idle_timeout=300.0, read_only=False, track_operations=False, adaptive_retry=False, retry_strategy='exponential', retry_jitter=False, audit_handler=None, audit_callback=None, user_context=None)

Bases: BaseHTTPClient

Internal async HTTP client for FortiOS API requests (Async Implementation)

Implements the IHTTPClient protocol for asynchronous HTTP operations.

Async version of HTTPClient using httpx.AsyncClient. Handles all HTTP communication with FortiGate devices including: - Async session management - Authentication headers - SSL verification - Request/response handling - Error handling - Automatic retry with exponential backoff - Async context manager support (use with ‘async with’ statement)

Protocol Implementation:

This class implements the IHTTPClient protocol, allowing it to be used interchangeably with other HTTP client implementations (e.g., HTTPClient, custom user-provided async clients). All methods return coroutines that must be awaited.

This class is internal and not exposed to users directly, but users can provide their own async IHTTPClient implementations to FortiOS.__init__().

Parameters:

• url (str)

• verify (bool)

• token (Optional[str])

• username (Optional[str])

• password (Optional[str])

• vdom (Optional[str])

• max_retries (int)

• connect_timeout (float)

• read_timeout (float)

• user_agent (Optional[str])

• circuit_breaker_threshold (int)

• circuit_breaker_timeout (float)

• circuit_breaker_auto_retry (bool)

• circuit_breaker_max_retries (int)

• circuit_breaker_retry_delay (float)

• max_connections (int)

• max_keepalive_connections (int)

• session_idle_timeout (Optional[float])

• read_only (bool)

• track_operations (bool)

• adaptive_retry (bool)

• retry_strategy (str)

• retry_jitter (bool)

• audit_handler (Optional[Any])

• audit_callback (Optional[Any])

• user_context (Optional[dict[str, Any]])

__init__(url, verify=True, token=None, username=None, password=None, vdom=None, max_retries=3, connect_timeout=10.0, read_timeout=300.0, user_agent=None, circuit_breaker_threshold=10, circuit_breaker_timeout=30.0, circuit_breaker_auto_retry=False, circuit_breaker_max_retries=3, circuit_breaker_retry_delay=5.0, max_connections=100, max_keepalive_connections=20, session_idle_timeout=300.0, read_only=False, track_operations=False, adaptive_retry=False, retry_strategy='exponential', retry_jitter=False, audit_handler=None, audit_callback=None, user_context=None)

Initialize async HTTP client

Parameters:

• url (str) – Base URL for API (e.g., “https://192.0.2.10”)

• verify (bool) – Verify SSL certificates

• token (Optional[str]) – API authentication token (if using token auth)

• username (Optional[str]) – Username for authentication (if using username/password

• auth)

• password (Optional[str]) – Password for authentication (if using username/password

• auth)

• vdom (Optional[str]) – Default virtual domain

• max_retries (int) – Maximum number of retry attempts on transient failures

• (default (all API calls) –

• connect_timeout (float) – Timeout for establishing connection in seconds

• (default – 10.0)

• read_timeout (float) – Timeout for reading response in seconds (default:

• 300.0)

• user_agent (Optional[str]) – Custom User-Agent header

• circuit_breaker_threshold (int) – Number of consecutive failures before

• (default –

• circuit_breaker_timeout (float) – Seconds to wait before transitioning to

• (default – 30.0)

• circuit_breaker_auto_retry (bool) – When True, automatically wait and retry

• breaker (when circuit) – opens instead of raising error immediately (default: False). WARNING: Not recommended for test environments - may cause long delays.

• circuit_breaker_max_retries (int) – Maximum number of auto-retry attempts

• breaker – opens (default: 3). Only used when circuit_breaker_auto_retry=True.

• circuit_breaker_retry_delay (float) – Delay in seconds between retry

• (default – 5.0). Separate from circuit_breaker_timeout, which controls when the circuit transitions from open to half-open.

• max_connections (int) – Maximum number of connections in the pool

• (default –

• max_keepalive_connections (int) – Maximum number of keepalive connections

• (default –

• session_idle_timeout (Optional[float]) – For username/password auth only. Idle timeout

• before (in seconds) – proactively re-authenticating (default: 300 = 5 minutes). Set to None or False to disable. Note: Async client does not yet implement proactive re-auth; this parameter is accepted for API compatibility.

• read_only (bool) – Enable read-only mode - simulate write operations

• (default – False)

• track_operations (bool) – Enable operation tracking - maintain audit log of

• (default – False)

• adaptive_retry (bool) – Enable adaptive retry with backpressure detection

• (default – False). When enabled, monitors response times and adjusts retry delays based on FortiGate health signals (slow responses, 503 errors).

• retry_strategy (str) – Retry backoff strategy - ‘exponential’ (default) or ‘linear’. Exponential: 1s, 2s, 4s, 8s, 16s, 30s. Linear: 1s, 2s, 3s, 4s, 5s.

• retry_jitter (bool) – Add random jitter (0-25% of delay) to retry delays to prevent thundering herd problem (default: False).

• audit_handler (Optional[Any]) – Handler for audit logging (implements AuditHandler

• protocol). – Use built-in handlers: SyslogHandler, FileHandler, StreamHandler, CompositeHandler. Essential for compliance (SOC 2, HIPAA, PCI-DSS).

• audit_callback (Optional[Any]) – Custom callback function for audit logging. Alternative to audit_handler. Receives operation dict as parameter.

• user_context (Optional[dict[str, Any]]) – Optional dict with user/application context to

• logs. (include in audit) – Example: {“username”: “admin”, “app”: “automation”, “ticket”: “CHG-12345”}

Raises:

• ValueError – If parameters are invalid or both token and

• username/password provided –

Return type:

None

async login()

Authenticate using username/password and obtain session token (async)

Must be called manually for async clients (cannot be called in __init__). Alternatively, use async context manager which handles login/logout automatically.

Raises:

• ValueError – If username/password not configured

• AuthenticationError – If login fails

Return type:

None

Example

>>> client = AsyncHTTPClient(url, username="admin",
password="password")
>>> await client.login()

async logout()

Logout and invalidate session token (async)

This method is called automatically when using async context manager. Can also be called manually to explicitly logout.

Note

Only applicable when using username/password authentication. Token-based authentication doesn’t require logout.

Return type:

None

get_connection_stats()

Get HTTP connection pool statistics (async client)

Returns:

Connection statistics including:

• http2_enabled: Whether HTTP/2 is enabled

• max_connections: Maximum number of connections allowed

• max_keepalive_connections: Maximum keepalive connections

• active_requests: Current number of active requests

• total_requests: Total requests made since initialization

• pool_exhaustion_count: Times pool reached capacity

• circuit_breaker_state: Current circuit breaker state

• consecutive_failures: Number of consecutive failures

Return type:

dict

Example

>>> stats = client.get_connection_stats()
>>> print(f"Circuit breaker: {stats['circuit_breaker_state']}")
>>> print(f"Active requests: {stats['active_requests']}")

inspect_last_request()

Get details of last API request for debugging

Returns:

Request information including:

• method: HTTP method used

• endpoint: API endpoint path

• params: Query parameters

• response_time_ms: Response time in milliseconds

• status_code: HTTP status code

• error: Error message if no requests made

Return type:

dict

Example

>>> await client.get("/api/v2/cmdb/firewall/address")
>>> info = client.inspect_last_request()
>>> print(f"Last request took {info['response_time_ms']:.2f}ms")

async request(method, api_type, path, data=None, params=None, vdom=None, raw_json=False, request_id=None)

Generic async request method for all API calls

Parameters:

• method (str) – HTTP method (GET, POST, PUT, DELETE)

• api_type (str) – API type (cmdb, monitor, log, service)

• path (str) – API endpoint path

• data (Optional[dict[str, Any]]) – Request body data (for POST/PUT)

• params (Optional[dict[str, Any]]) – Query parameters dict

• vdom (Union[str, bool, None]) – Virtual domain

• raw_json (bool) – If False, return only ‘results’ field. If True, return

• response (full)

• request_id (Optional[str]) – Optional correlation ID for tracking requests

Returns:

API response (results or full response based on raw_json)

Return type:

dict

async get(api_type, path, params=None, vdom=None, raw_json=False)

Async GET request

Return type:

dict[str, Any]

Parameters:

• api_type (str)

• path (str)

• params (dict[str, Any] | None)

• vdom (str | bool | None)

• raw_json (bool)

async get_binary(api_type, path, params=None, vdom=None)

Async GET request returning binary data

Return type:

bytes

Parameters:

• api_type (str)

• path (str)

• params (dict[str, Any] | None)

• vdom (str | bool | None)

async post(api_type, path, data, params=None, vdom=None, raw_json=False)

Async POST request - Create new object

Return type:

dict[str, Any]

Parameters:

• api_type (str)

• path (str)

• data (dict[str, Any])

• params (dict[str, Any] | None)

• vdom (str | bool | None)

• raw_json (bool)

async put(api_type, path, data, params=None, vdom=None, raw_json=False)

Async PUT request - Update existing object

Return type:

dict[str, Any]

Parameters:

• api_type (str)

• path (str)

• data (dict[str, Any])

• params (dict[str, Any] | None)

• vdom (str | bool | None)

• raw_json (bool)

async delete(api_type, path, params=None, vdom=None, raw_json=False)

Async DELETE request - Delete object

Return type:

dict[str, Any]

Parameters:

• api_type (str)

• path (str)

• params (dict[str, Any] | None)

• vdom (str | bool | None)

• raw_json (bool)

static validate_mkey(mkey, parameter_name='mkey')

Validate and convert mkey to string

Return type:

str

Parameters:

• mkey (Any)

• parameter_name (str)

static validate_required_params(params, required)

Validate that required parameters are present

Return type:

None

Parameters:

• params (dict[str, Any])

• required (list[str])

static validate_range(value, min_val, max_val, parameter_name='value')

Validate that a numeric value is within a specified range

Return type:

None

Parameters:

• value (int | float)

• min_val (int | float)

• max_val (int | float)

• parameter_name (str)

static validate_choice(value, choices, parameter_name='value')

Validate that a value is one of the allowed choices

Return type:

None

Parameters:

• value (Any)

• choices (list[Any])

• parameter_name (str)

static build_params(**kwargs)

Build parameters dict, filtering out None values

Return type:

dict[str, Any]

Parameters:

kwargs (Any)

async close()

Close the async HTTP session and release resources

This method should be called to properly clean up resources when using AsyncHTTPClient. It ensures that all network connections and sessions are closed.

Return type:

None

Usage:

• Call await client.close() when you are done with the client in

System Message: WARNING/2 (/home/docs/checkouts/readthedocs.org/user_builds/hfortix-core/checkouts/latest/hfortix_core/http/async_client.py:docstring of hfortix_core.http.async_client.AsyncHTTPClient.close, line 11)

Bullet list ends without a blank line; unexpected unindent.

async mode. - Prefer using the async context manager (async with) for automatic cleanup.

Example

client = AsyncHTTPClient(…) try:

System Message: ERROR/3 (/home/docs/checkouts/readthedocs.org/user_builds/hfortix-core/checkouts/latest/hfortix_core/http/async_client.py:docstring of hfortix_core.http.async_client.AsyncHTTPClient.close, line 19)

Unexpected indentation.

…

System Message: WARNING/2 (/home/docs/checkouts/readthedocs.org/user_builds/hfortix-core/checkouts/latest/hfortix_core/http/async_client.py:docstring of hfortix_core.http.async_client.AsyncHTTPClient.close, line 20)

Block quote ends without a blank line; unexpected unindent.

finally:

await client.close()

get_operations()

Get audit log of all tracked API operations

Returns all tracked operations (GET/POST/PUT/DELETE) in chronological order. Only available when track_operations=True was passed to constructor.

Return type:

list[dict[str, Any]]

Returns:

List of operation dictionaries (same format as HTTPClient.get_operations())

get_write_operations()

Get audit log of write operations only (POST/PUT/DELETE)

Filters tracked operations to return only write operations, excluding GET requests.

Return type:

list[dict[str, Any]]

Returns:

List of write operation dictionaries (same format as HTTPClient.get_write_operations())

static make_exists_method(get_method)

Create an exists() helper that works with both sync and async modes.

This utility wraps a get() method and returns a function that: - Returns True if the object exists - Returns False if ResourceNotFoundError is raised - Works transparently with both sync and async clients

Parameters:

• get_method (Callable[..., Any]) – The get() method to wrap (bound method from endpoint

• instance)

Return type:

Callable[..., bool]

Returns:

A function that returns bool (sync) or Coroutine[bool] (async)

Example

class Address:

def __init__(self, client):

self._client = client

def get(self, name, **kwargs):

System Message: WARNING/2 (/home/docs/checkouts/readthedocs.org/user_builds/hfortix-core/checkouts/latest/hfortix_core/http/async_client.py:docstring of hfortix_core.http.async_client.AsyncHTTPClient.make_exists_method, line 21); backlink

Inline strong start-string without end-string.

return self._client.get(“cmdb”, f”/firewall/address/{name}”, **kwargs)

System Message: WARNING/2 (/home/docs/checkouts/readthedocs.org/user_builds/hfortix-core/checkouts/latest/hfortix_core/http/async_client.py:docstring of hfortix_core.http.async_client.AsyncHTTPClient.make_exists_method, line 22); backlink

Inline strong start-string without end-string.

# Create exists method using the helper exists = AsyncHTTPClient.make_exists_method(get)

Caching

TTLCache

class hfortix_core.TTLCache(default_ttl=3600)

Bases: Generic[T]

Simple TTL-based cache for readonly reference data.

Thread-safe in-memory cache with time-based expiration. Designed for caching readonly reference tables that rarely change.

Example

>>> cache = TTLCache[dict](default_ttl=3600)  # 1 hour
>>> cache.set("geography/countries", country_data)
>>> data = cache.get("geography/countries")

Parameters:

default_ttl (float)

__init__(default_ttl=3600)

Initialize TTL cache.

Parameters:

default_ttl (float) – Default time-to-live in seconds (default: 1 hour)

get(key)

Get cached value if not expired.

Parameters:

key (str) – Cache key

Return type:

Optional[TypeVar(T)]

Returns:

Cached value or None if expired/not found

set(key, value, ttl=None)

Store value in cache with TTL.

Parameters:

• key (str) – Cache key

• value (TypeVar(T)) – Value to cache

• ttl (float | None) – Time-to-live in seconds (uses default if None)

Return type:

None

invalidate(key)

Remove entry from cache.

Parameters:

key (str) – Cache key to remove

Return type:

None

clear()

Clear all cached entries.

Return type:

None

cleanup()

Remove all expired entries.

Return type:

int

Returns:

Number of entries removed

readonly_cache

hfortix_core.readonly_cache = <hfortix_core.cache.TTLCache object>

Simple TTL-based cache for readonly reference data.

Thread-safe in-memory cache with time-based expiration. Designed for caching readonly reference tables that rarely change.

Example

>>> cache = TTLCache[dict](default_ttl=3600)  # 1 hour
>>> cache.set("geography/countries", country_data)
>>> data = cache.get("geography/countries")

Logging

RequestLogger

class hfortix_core.RequestLogger(method, endpoint, extra=None)

Bases: object

Context manager for logging API requests with timing and status

Automatically logs request start/completion and calculates duration. Logs errors with full context if the request fails.

Example

>>> with RequestLogger("POST", "/api/v2/cmdb/firewall/address", extra={"vdom": "root"}):  # noqa: E501
...     response = make_request()
# Logs: ✓ POST /api/v2/cmdb/firewall/address (0.234s) {vdom: root}

Parameters:

• method (str)

• endpoint (str)

• extra (dict[str, Any] | None)

__init__(method, endpoint, extra=None)

Initialize request logger

Parameters:

• method (str) – HTTP method (GET, POST, PUT, DELETE)

• endpoint (str) – API endpoint path

• extra (Optional[dict[str, Any]]) – Additional context to include in logs

log_operation

hfortix_core.log_operation(logger_name, operation, level='INFO', **kwargs)

Log an operation with structured data

Parameters:

• logger_name (str) – Name of the logger to use (e.g., “hfortix.http”)

• operation (str) – Operation being performed (logged as message)

• level (str) – Log level (DEBUG, INFO, WARNING, ERROR)

• **kwargs – Additional context to log as extra fields

Return type:

None

Example

>>> log_operation(
...     "hfortix.client",
...     "Creating firewall address",
...     level="INFO",
...     address_name="server1",
...     subnet="10.0.0.1/32",
...     vdom="root"
... )
# Logs: Creating firewall address {address_name: server1, subnet: 10.0.0.1/32, vdom: root}  # noqa: E501

StructuredFormatter

class hfortix_core.StructuredFormatter(include_fields=None, exclude_fields=None)

Bases: Formatter

Format log records as structured JSON

Useful for log aggregation systems like ELK, Splunk, CloudWatch that can parse JSON logs for better searching and analysis.

Standard Fields (always present):

• timestamp: ISO 8601 UTC timestamp

• level: Log level (DEBUG, INFO, WARNING, ERROR, CRITICAL)

• logger: Logger name (e.g., “hfortix.http.client”)

• message: Log message

Common Extra Fields (added via extra={…}):

• request_id: Unique request identifier for correlation

• method: HTTP method (GET, POST, PUT, DELETE, PATCH)

• endpoint: API endpoint path

• status_code: HTTP status code

• duration_seconds: Request duration in seconds

• vdom: FortiOS Virtual Domain (multi-tenant environments)

• adom: FortiManager/FortiAnalyzer Administrative Domain

• error_type: Exception class name (for errors)

• attempt: Current retry attempt number

• max_attempts: Maximum retry attempts

Example

>>> import logging
>>> from hfortix_core.logging import StructuredFormatter
>>>
>>> handler = logging.StreamHandler()
>>> handler.setFormatter(StructuredFormatter())
>>> logger = logging.getLogger("hfortix")
>>> logger.addHandler(handler)
>>>
>>> # Basic usage
>>> logger.info("API request completed",
...            extra={"endpoint": "/api/v2/cmdb/firewall/policy",
...                   "duration_seconds": 0.145})

Output: {“timestamp”:”2026-01-02T14:23:45.123Z”,”level”:”INFO”,

System Message: ERROR/3 (/home/docs/checkouts/readthedocs.org/user_builds/hfortix-core/checkouts/latest/hfortix_core/logging/formatters.py:docstring of hfortix_core.logging.formatters.StructuredFormatter, line 41)

Unexpected indentation.

“logger”:”hfortix”,”message”:”API request completed”, “endpoint”:”/api/v2/cmdb/firewall/policy”,”duration_seconds”:0.145}

>>> # Multi-tenant usage
>>> logger.info("Policy created",
...            extra={"vdom": "customer_a", "endpoint": "/api/v2/cmdb/firewall/policy",  # noqa: E501
...                   "request_id": "req-123", "status_code": 200})

Output: {“timestamp”:”2026-01-02T14:23:45.456Z”,”level”:”INFO”,

System Message: ERROR/3 (/home/docs/checkouts/readthedocs.org/user_builds/hfortix-core/checkouts/latest/hfortix_core/logging/formatters.py:docstring of hfortix_core.logging.formatters.StructuredFormatter, line 51)

Unexpected indentation.

“logger”:”hfortix”,”message”:”Policy created”,”vdom”:”customer_a”, “endpoint”:”/api/v2/cmdb/firewall/policy”,”request_id”:”req-123”, “status_code”:200}

Parameters:

• include_fields (list[str] | None)

• exclude_fields (list[str] | None)

__init__(include_fields=None, exclude_fields=None)

Initialize structured formatter

Parameters:

• include_fields (list[str] | None) – List of extra fields to include (None = all)

• exclude_fields (list[str] | None) – List of fields to exclude from output

format(record)

Format log record as JSON

Parameters:

record (LogRecord) – LogRecord to format

Return type:

str

Returns:

JSON string with log data

TextFormatter

class hfortix_core.TextFormatter(use_color=False)

Bases: Formatter

Format log records as human-readable text

Provides colorized output for terminal and clean formatting for log files.

Example

>>> import logging
>>> from hfortix_core.logging import TextFormatter
>>>
>>> handler = logging.StreamHandler()
>>> handler.setFormatter(TextFormatter(use_color=True))
>>> logger = logging.getLogger("hfortix")
>>> logger.addHandler(handler)
>>> logger.info("API request completed")

Output: 2026-01-02 14:23:45 [INFO] hfortix: API request completed

Parameters:

use_color (bool)

COLORS = {'CRITICAL': '\x1b[35m', 'DEBUG': '\x1b[36m', 'ERROR': '\x1b[31m', 'INFO': '\x1b[32m', 'RESET': '\x1b[0m', 'WARNING': '\x1b[33m'}

__init__(use_color=False)

Initialize text formatter

Parameters:

use_color (bool) – Whether to use ANSI color codes

format(record)

Format log record as text

Parameters:

record (LogRecord) – LogRecord to format

Return type:

str

Returns:

Formatted text string

LogFormatter (Protocol)

class hfortix_core.logging.LogFormatter(*args, **kwargs)

Bases: Protocol

Protocol for log formatters

Any class implementing this protocol can be used as a formatter for HFortix logging.

format(record)

Format a log record

Parameters:

record (Any) – logging.LogRecord instance

Return type:

str

Returns:

Formatted string

LogRecord (TypedDict)

class hfortix_core.logging.LogRecord

Bases: TypedDict

Type definition for structured log record data

timestamp

ISO 8601 UTC timestamp

level

Log level (DEBUG, INFO, WARNING, ERROR, CRITICAL)

logger

Logger name (e.g., “hfortix.http.client”)

message

Log message

request_id

Unique request identifier for correlation

method

HTTP method (GET, POST, PUT, DELETE)

endpoint

API endpoint path

status_code

HTTP status code

duration_s

Request duration in seconds

duration_ms

Request duration in milliseconds

vdom

FortiOS Virtual Domain

event

Event type (request_start, request_completed, request_failed)

error

Error message (if applicable)

error_type

Exception class name

attempt

Current retry attempt number

max_attempts

Maximum retry attempts

source

Source location (file, line, function)

timestamp: str

level: str

logger: str

message: str

request_id: str

method: str

endpoint: str

status_code: int

duration_s: float

duration_ms: float

vdom: str

event: str

error: str

error_type: str

attempt: int

max_attempts: int

source: dict[str, Any]

Debugging

DebugSession

class hfortix_core.DebugSession(client, capture_response_data=False, print_on_exit=True)

Bases: object

Context manager for debugging FortiOS API sessions.

Captures detailed information about all API requests made within the session, # noqa: E501 including timing, connection stats, and request/response details.

Parameters:

• client (FortiOS) – FortiOS client instance to debug

• capture_response_data (bool) – Whether to capture full response bodies (default: False) # noqa: E501

• print_on_exit (bool) – Whether to print summary when exiting context (default: True) # noqa: E501

Example

>>> with DebugSession(fgt, capture_response_data=True) as session:
...     fgt.cmdb.firewall.address.get()
...     fgt.cmdb.firewall.policy.create(data={...})
...     session.print_summary()

Session Summary:

Duration: 1.234s Total Requests: 2 Successful: 2 Failed: 0 Average Response Time: 617ms …

__init__(client, capture_response_data=False, print_on_exit=True)

Initialize debug session.

Parameters:

• client (FortiOS) – FortiOS client to monitor

• capture_response_data (bool) – Capture full response bodies

• print_on_exit (bool) – Auto-print summary on context exit

Return type:

None

property duration: float | None

Session duration in seconds.

property request_count: int

Total number of requests captured.

property stats_delta: dict[str, Any] | None

Difference in connection stats from start to end.

capture_request(method, endpoint, params=None, response_time_ms=None, status_code=None, response_data=None, error=None)

Capture information about an API request.

Parameters:

• method (str) – HTTP method (GET, POST, etc.)

• endpoint (str) – API endpoint path

• params (dict[str, Any] | None) – Request parameters

• response_time_ms (float | None) – Response time in milliseconds

• status_code (int | None) – HTTP status code

• response_data (Any) – Response body (if capture_response_data=True)

• error (str | None) – Error message if request failed

Return type:

None

get_summary()

Get session summary as dictionary.

Return type:

dict[str, Any]

Returns:

Dictionary with session metrics and statistics

print_summary(format='text')

Print session summary.

Parameters:

format (Literal['text', 'json']) – Output format - ‘text’ for human-readable, ‘json’ for structured # noqa: E501

Return type:

None

print_requests(verbose=False)

Print detailed information about captured requests.

Parameters:

verbose (bool) – Include full request/response details

Return type:

None

debug_timer

hfortix_core.debug_timer(operation='Operation')

Context manager for timing operations.

Parameters:

operation (str) – Name of the operation being timed

Yields:

Dictionary that will be populated with timing information

Return type:

Iterator[dict[str, Any]]

Example

>>> with debug_timer("Get firewall addresses") as timing:
...     result = fgt.cmdb.firewall.address.get()
>>> print(f"Took {timing['duration_ms']:.1f}ms")

format_connection_stats

hfortix_core.format_connection_stats(stats)

Format connection stats dictionary as human-readable string.

Parameters:

stats (dict[str, Any] | None) – Connection stats from client.connection_stats

Return type:

str

Returns:

Formatted string with connection statistics

Example

>>> stats = fgt.connection_stats
>>> print(format_connection_stats(stats))
Connection Pool Statistics
==========================
Max Connections: 100
Active Requests: 2
Total Requests: 1543
Pool Exhaustion Events: 0

format_request_info

hfortix_core.format_request_info(request_info)

Format request info dictionary as human-readable string.

Parameters:

request_info (dict[str, Any] | None) – Request info from client.last_request

Return type:

str

Returns:

Formatted string with request details

Example

>>> info = fgt.last_request
>>> print(format_request_info(info))
GET /api/v2/cmdb/firewall/address
Response Time: 123.4ms
Status Code: 200

print_debug_info

hfortix_core.print_debug_info(client, include_last_request=True, include_connection_stats=True)

Print comprehensive debug information for a FortiOS client.

Parameters:

• client (FortiOS) – FortiOS client to inspect

• include_last_request (bool) – Include last request details

• include_connection_stats (bool) – Include connection pool statistics

Return type:

None

Example

>>> from hfortix_core.debug import print_debug_info
>>> fgt = FortiOS(host="192.168.1.99", token="your-token")
>>> fgt.cmdb.firewall.address.get()
>>> print_debug_info(fgt)

DebugFormatter (Protocol)

class hfortix_core.debug.DebugFormatter(*args, **kwargs)

Bases: Protocol

Protocol for debug information formatters

Any class implementing this protocol can be used to format debug information for display.

format_request(request_info)

Format request information as string

Parameters:

request_info (RequestInfo | dict[str, Any] | None) – Request information dictionary

Return type:

str

Returns:

Formatted string

format_stats(stats)

Format connection statistics as string

Parameters:

stats (dict[str, Any] | None) – Connection statistics dictionary

Return type:

str

Returns:

Formatted string

DebugInfo (TypedDict)

class hfortix_core.debug.DebugInfo

Bases: TypedDict

Type definition for comprehensive debug information

last_request

Information about the last API request

connection_stats

Connection pool statistics

session_active

Whether a debug session is active

capture_enabled

Whether response capture is enabled

last_request: RequestInfo | None

connection_stats: dict[str, Any]

session_active: bool

capture_enabled: bool

SessionSummary (TypedDict)

class hfortix_core.debug.SessionSummary

Bases: TypedDict

Type definition for debug session summary

duration_seconds

Total session duration in seconds

total_requests

Total number of requests made

successful_requests

Number of successful requests

failed_requests

Number of failed requests

avg_response_time_ms

Average response time in milliseconds

max_response_time_ms

Maximum response time in milliseconds

min_response_time_ms

Minimum response time in milliseconds

stats_delta

Change in connection stats during session

initial_stats

Connection stats at session start

final_stats

Connection stats at session end

duration_seconds: float | None

total_requests: int

successful_requests: int

failed_requests: int

avg_response_time_ms: float | None

max_response_time_ms: float | None

min_response_time_ms: float | None

stats_delta: dict[str, Any] | None

initial_stats: dict[str, Any] | None

final_stats: dict[str, Any] | None

Formatting

fmt Module

Data formatting utilities for FortiOS objects and data structures.

Provides simple, type-agnostic conversion functions that handle any input gracefully. Never raises exceptions - returns sensible defaults for edge cases.

hfortix_core.fmt.to_json(data, indent=2, **kwargs)

Convert any data to formatted JSON string.

Handles objects with __dict__, converts sets/tuples to lists, and uses str() fallback for non-serializable types.

Parameters:

• data (Any) – Any Python object to convert to JSON

• indent (int) – Number of spaces for indentation (default: 2)

• **kwargs (Any) – Additional arguments passed to json.dumps()

Return type:

str

Returns:

Formatted JSON string

hfortix_core.fmt.to_csv(data, separator=', ')

Convert any data to comma-separated string.

Parameters:

• data (Any) – Any Python object to convert

• separator (str) – String to use between items (default: ‘, ‘)

Return type:

str

Returns:

Comma-separated string

Examples

>>> to_csv(['port1', 'port2', 'port3'])
'port1, port2, port3'

>>> to_csv({'x': 1, 'y': 2, 'z': 3})
'x=1, y=2, z=3'

>>> to_csv('already a string')
'already a string'

>>> to_csv(None)
''

>>> to_csv([1, 2, 3], separator=' | ')
'1 | 2 | 3'

>>> class Interface:
...     def __init__(self):
...         self.name = "port1"
...         self.ip = "10.0.0.1"
>>> to_csv(Interface())
'name=port1, ip=10.0.0.1'

hfortix_core.fmt.to_dict(data)

Convert any data to dictionary.

Returns a dict with string keys for most inputs (objects, existing dicts), or integer keys for list/tuple inputs.

Parameters:

data (Any) – Any Python object to convert

Return type:

dict

Returns:

Dictionary representation of the data

Examples

>>> class Policy:
...     def __init__(self):
...         self.name = "Allow-All"
...         self.action = "accept"
>>> to_dict(Policy())
{'name': 'Allow-All', 'action': 'accept'}

>>> to_dict({'already': 'a dict'})
{'already': 'a dict'}

>>> to_dict([('a', 1), ('b', 2)])
{'a': 1, 'b': 2}

>>> to_dict(['x', 'y', 'z'])
{0: 'x', 1: 'y', 2: 'z'}

>>> to_dict('simple string')
{'value': 'simple string'}

>>> to_dict(None)
{'value': None}

hfortix_core.fmt.to_multiline(data, separator='\\n')

Convert any data to newline-separated string.

Parameters:

• data (Any) – Any Python object to convert

• separator (str) – String to use between lines (default: ‘n’)

Return type:

str

Returns:

Newline-separated string

Examples

>>> print(to_multiline(['port1', 'port2', 'port3']))
port1
port2
port3

>>> print(to_multiline({'name': 'policy1', 'action': 'accept'}))
name: policy1
action: accept

>>> to_multiline('already a string')
'already a string'

>>> to_multiline(None)
''

>>> class Policy:
...     def __init__(self):
...         self.name = "Allow-All"
...         self.policyid = 1
>>> print(to_multiline(Policy()))
name: Allow-All
policyid: 1

hfortix_core.fmt.to_list(data, delimiter=None)

Convert any data to list.

Parameters:

• data (Any) – Any Python object to convert

• delimiter (str | None) – If data is a string, split by this delimiter. If None and data is string with spaces, auto-splits by space. Common delimiters: ‘,’, ‘ ‘, ‘|’, ‘;’, etc.

Return type:

list[Any]

Returns:

List representation of the data

Examples

>>> to_list(['already', 'a', 'list'])
['already', 'a', 'list']

>>> to_list(('tuple', 'to', 'list'))
['tuple', 'to', 'list']

>>> to_list({'a', 'b', 'c'})  # set to list
['a', 'b', 'c']

>>> to_list('port1,port2,port3', delimiter=',')
['port1', 'port2', 'port3']

>>> to_list('port1 port2 port3')  # auto-splits on space
['port1', 'port2', 'port3']

>>> to_list('80 443 8080')  # works with numbers as strings
['80', '443', '8080']

>>> to_list('port1 | port2 | port3', delimiter=' | ')
['port1', 'port2', 'port3']

>>> to_list('single_string')  # no spaces, returns as-is
['single_string']

>>> to_list({'name': 'policy1', 'action': 'accept'})
['policy1', 'accept']

>>> to_list(None)
[]

>>> to_list(42)
[42]

>>> class Policy:
...     def __init__(self):
...         self.name = "Allow-All"
...         self.policyid = 1
>>> to_list(Policy())
['Allow-All', 1]

hfortix_core.fmt.to_quoted(data, quote='"', separator=', ')

Convert any data to quoted string representation.

Parameters:

• data (Any) – Any Python object to convert

• quote (str) – Quote character to use (default: ‘”’)

• separator (str) – String to use between quoted items (default: ‘, ‘)

Return type:

str

Returns:

Quoted string representation

Examples

>>> to_quoted(['port1', 'port2', 'port3'])
'"port1", "port2", "port3"'

>>> to_quoted({'x': 1, 'y': 2})
'"x", "y"'

>>> to_quoted('hello')
'"hello"'

>>> to_quoted(None)
'""'

>>> to_quoted([1, 2, 3], quote="'")
"'1', '2', '3'"

>>> class Interface:
...     def __init__(self):
...         self.name = "port1"
...         self.vlan = 10
>>> to_quoted(Interface())
'"name", "vlan"'

hfortix_core.fmt.to_table(data, headers=True, delimiter=' | ')

Convert data to table format.

Parameters:

• data (Any) – Any Python object to convert (list of dicts, list of objects, etc.)

• headers (bool) – Whether to include headers (default: True)

• delimiter (str) – Column delimiter (default: ‘ | ‘)

Return type:

str

Returns:

Table-formatted string

Examples

>>> policies = [
...     {'name': 'Allow-Web', 'action': 'accept', 'policyid': 1},
...     {'name': 'Block-All', 'action': 'deny', 'policyid': 2}
... ]
>>> print(to_table(policies))
name | action | policyid
Allow-Web | accept | 1
Block-All | deny | 2

>>> to_table(policies, headers=False)
'Allow-Web | accept | 1\nBlock-All | deny | 2'

>>> to_table(policies, delimiter=' || ')
'name || action || policyid\nAllow-Web || accept || 1\nBlock-All || deny || 2'

hfortix_core.fmt.to_yaml(data, indent=2)

Convert data to YAML-like format (simple, no external dependencies).

Parameters:

• data (Any) – Any Python object to convert

• indent (int) – Number of spaces for indentation (default: 2)

Return type:

str

Returns:

YAML-style string

Examples

>>> policy = {'name': 'Allow-Web', 'action': 'accept', 'srcintf': ['port1', 'port2']}
>>> print(to_yaml(policy))
name: Allow-Web
action: accept
srcintf:
  - port1
  - port2

>>> print(to_yaml({'nested': {'key': 'value'}}))
nested:
  key: value

hfortix_core.fmt.to_xml(data, root='data', indent=2)

Convert data to simple XML format (no external dependencies).

Parameters:

• data (Any) – Any Python object to convert

• root (str) – Root element name (default: ‘data’)

• indent (int) – Number of spaces for indentation (default: 2)

Return type:

str

Returns:

XML string

Examples

>>> policy = {'name': 'Allow-Web', 'policyid': 1}
>>> print(to_xml(policy, root='policy'))
<policy>
  <name>Allow-Web</name>
  <policyid>1</policyid>
</policy>

>>> policies = [{'name': 'p1'}, {'name': 'p2'}]
>>> print(to_xml(policies, root='policies'))
<policies>
  <item>
    <name>p1</name>
  </item>
  <item>
    <name>p2</name>
  </item>
</policies>

hfortix_core.fmt.to_key_value(data, separator='=', delimiter='\\n')

Convert data to key=value pairs format.

Parameters:

• data (Any) – Any Python object to convert

• separator (str) – Separator between key and value (default: ‘=’)

• delimiter (str) – Delimiter between pairs (default: ‘n’)

Return type:

str

Returns:

Key-value pairs string

Examples

>>> config = {'host': '192.168.1.1', 'port': 443, 'verify': False}
>>> print(to_key_value(config))
host=192.168.1.1
port=443
verify=False

>>> to_key_value(config, separator=': ', delimiter='; ')
'host: 192.168.1.1; port: 443; verify: False'

hfortix_core.fmt.to_markdown_table(data)

Convert data to Markdown table format.

Parameters:

data (Any) – List of dicts or list of objects

Return type:

str

Returns:

Markdown table string

Examples

>>> policies = [
...     {'name': 'Allow-Web', 'action': 'accept'},
...     {'name': 'Block-All', 'action': 'deny'}
... ]
>>> print(to_markdown_table(policies))
| name | action |
| --- | --- |
| Allow-Web | accept |
| Block-All | deny |

hfortix_core.fmt.to_dictlist(data)

Convert dict of lists to list of dicts (columnar to row format).

Useful for transforming columnar data into row-based records.

Parameters:

data (Any) – Dict where values are lists, or any convertible data

Return type:

list[dict[str, Any]]

Returns:

List of dicts where each dict is one row

Examples

>>> columnar = {'name': ['p1', 'p2'], 'action': ['accept', 'deny']}
>>> to_dictlist(columnar)
[{'name': 'p1', 'action': 'accept'}, {'name': 'p2', 'action': 'deny'}]

>>> to_dictlist({'ports': ['80', '443', '8080']})
[{'ports': '80'}, {'ports': '443'}, {'ports': '8080'}]

>>> # Already list of dicts - returns as-is
>>> to_dictlist([{'name': 'p1'}, {'name': 'p2'}])
[{'name': 'p1'}, {'name': 'p2'}]

>>> to_dictlist(None)
[]

hfortix_core.fmt.to_listdict(data)

Convert list of dicts to dict of lists (row to columnar format).

Useful for transforming row-based records into columnar data.

Parameters:

data (Any) – List of dicts, list of objects, or any convertible data

Return type:

dict[str, list[Any]]

Returns:

Dict where keys are field names and values are lists

Examples

>>> rows = [{'name': 'p1', 'action': 'accept'}, {'name': 'p2', 'action': 'deny'}]
>>> to_listdict(rows)
{'name': ['p1', 'p2'], 'action': ['accept', 'deny']}

>>> to_listdict([{'ports': '80'}, {'ports': '443'}, {'ports': '8080'}])
{'ports': ['80', '443', '8080']}

>>> # Single dict becomes dict of single-item lists
>>> to_listdict({'name': 'p1', 'action': 'accept'})
{'name': ['p1'], 'action': ['accept']}

>>> to_listdict(None)
{}

Audit Logging

Enterprise-grade audit logging for compliance and security monitoring.

AuditHandler (Protocol)

class hfortix_core.audit.AuditHandler(*args, **kwargs)

Bases: Protocol

Protocol for audit log handlers

Any class implementing this protocol can be used as an audit handler. This allows for maximum flexibility - users can provide custom handlers for their specific logging infrastructure.

The handler receives sanitized operation data and is responsible for: - Formatting the data as needed - Sending to destination (file, syslog, database, etc.) - Handling errors gracefully (should not raise exceptions)

Example

>>> class CustomHandler:
...     def log_operation(self, operation: dict[str, Any]) -> None:
...         # Send to your logging infrastructure
...         send_to_kafka(operation)
...         update_database(operation)
...
>>> handler = CustomHandler()
>>> fgt = FortiOS("192.168.1.99", token="...", audit_handler=handler)

log_operation(operation)

Log an API operation

Parameters:

operation (dict[str, Any]) – Dictionary containing operation details (see AuditOperation) # noqa: E501

Return type:

None

Note

This method should handle errors internally and not raise exceptions, # noqa: E501 as audit logging failures should not break API operations.

SyslogHandler

class hfortix_core.audit.SyslogHandler(server, formatter=None, timeout=5.0)

Bases: object

Send audit logs to a syslog server

Sends UDP packets to a syslog server in RFC 5424 format. Commonly used for SIEM integration (Splunk, ELK, QRadar, ArcSight).

Example

>>> from hfortix_core.audit import SyslogHandler
>>> handler = SyslogHandler("siem.company.com:514")
>>> # Use with FortiOS client
>>> fgt = FortiOS("192.168.1.99", token="...", audit_handler=handler)

Note

Uses UDP protocol which is fire-and-forget. For guaranteed delivery, consider using FileHandler with external log shipping.

Parameters:

• server (str)

• formatter (AuditFormatter | None)

• timeout (float)

__init__(server, formatter=None, timeout=5.0)

Initialize Syslog handler

Parameters:

• server (str) – Syslog server in format “host:port” or “host” (default port 514) # noqa: E501

• formatter (AuditFormatter | None) – Custom formatter (default: SyslogFormatter)

• timeout (float) – Socket timeout in seconds

log_operation(operation)

Send operation to syslog server

Parameters:

operation (dict[str, Any]) – Operation data dictionary

Return type:

None

close()

Close the socket

Return type:

None

FileHandler

class hfortix_core.audit.FileHandler(filepath, max_bytes=10000000, backup_count=5, formatter=None, mode='a')

Bases: object

Write audit logs to a file

Writes audit logs to a file in JSON Lines format (one JSON object per line). # noqa: E501 Supports automatic log rotation based on file size.

Example

>>> from hfortix_core.audit import FileHandler
>>> handler = FileHandler("/var/log/fortinet-audit.jsonl")
>>> fgt = FortiOS("192.168.1.99", token="...", audit_handler=handler)

File Format:

Each line is a complete JSON object (JSON Lines format): {“timestamp”:”2026-01-02T14:23:45Z”,”method”:”POST”,…} {“timestamp”:”2026-01-02T14:23:46Z”,”method”:”GET”,…}

This format is:

• Easy to parse (one record per line)

• Compatible with log shipping tools (Fluentd, Logstash, Filebeat)

• Human-readable with tools like jq

Parameters:

• filepath (str | Path)

• max_bytes (int)

• backup_count (int)

• formatter (AuditFormatter | None)

• mode (str)

__init__(filepath, max_bytes=10000000, backup_count=5, formatter=None, mode='a')

Initialize File handler

Parameters:

• filepath (str | Path) – Path to log file

• max_bytes (int) – Maximum file size before rotation (bytes)

• backup_count (int) – Number of backup files to keep

• formatter (AuditFormatter | None) – Custom formatter (default: JSONFormatter)

• mode (str) – File open mode (‘a’ for append, ‘w’ for overwrite)

log_operation(operation)

Write operation to file

Parameters:

operation (dict[str, Any]) – Operation data dictionary

Return type:

None

close()

Close the file

Return type:

None

StreamHandler

class hfortix_core.audit.StreamHandler(stream=None, formatter=None)

Bases: object

Write audit logs to a stream (stdout/stderr)

Useful for containerized applications where logs are captured by container orchestration (Docker, Kubernetes) and sent to centralized logging (CloudWatch, Datadog, etc.).

Example

>>> from hfortix_core.audit import StreamHandler
>>> import sys
>>>
>>> # Log to stdout (captured by Docker/Kubernetes)
>>> handler = StreamHandler(sys.stdout, format="json")
>>> fgt = FortiOS("192.168.1.99", token="...", audit_handler=handler)

Parameters:

• stream (TextIO | None)

• formatter (AuditFormatter | None)

__init__(stream=None, formatter=None)

Initialize Stream handler

Parameters:

• stream (TextIO | None) – Output stream (default: sys.stdout)

• formatter (AuditFormatter | None) – Custom formatter (default: JSONFormatter)

log_operation(operation)

Write operation to stream

Parameters:

operation (dict[str, Any]) – Operation data dictionary

Return type:

None

CompositeHandler

class hfortix_core.audit.CompositeHandler(handlers, aggregate_errors=True, error_threshold=10)

Bases: object

Send audit logs to multiple handlers with advanced routing

Features: - Multiple destination routing - Priority-based ordering - Conditional routing with filters - Error aggregation and reporting - Individual handler enable/disable

Example (Basic):

>>> from hfortix_core.audit import CompositeHandler, SyslogHandler, FileHandler  # noqa: E501
>>>
>>> # Send to both SIEM and local file
>>> handler = CompositeHandler([
...     SyslogHandler("siem.company.com:514"),  # Compliance
...     FileHandler("/var/log/fortinet.log"),   # Debugging/backup
... ])
>>>
>>> fgt = FortiOS("192.168.1.99", token="...", audit_handler=handler)

Example (Priority and Filtering):

>>> # Higher priority handlers execute first
>>> handler = CompositeHandler([
...     (critical_handler, 10, lambda op: op['action'] == 'delete'),
...     (normal_handler, 5, lambda op: op['success']),
...     (all_handler, 1, None),  # No filter, always executes
... ])

Error Handling:

If one handler fails, others continue. Errors are aggregated and can be retrieved via error_summary property.

Parameters:

• handlers (list[Any])

• aggregate_errors (bool)

• error_threshold (int)

__init__(handlers, aggregate_errors=True, error_threshold=10)

Initialize Composite handler

Parameters:

• handlers (list[Any]) – List of handlers or tuples (handler, priority, filter_fn) - handler: Implements AuditHandler protocol - priority: Int (higher = executes first), default: 0 - filter_fn: Callable[[dict], bool] or None, default: None

• aggregate_errors (bool) – Track errors for reporting

• error_threshold (int) – Max errors to track per handler

log_operation(operation)

Send operation to all matching handlers

Parameters:

operation (dict[str, Any]) – Operation data dictionary

Return type:

None

property error_summary: dict[str, Any]

Get summary of handler errors

Returns:

Dictionary with error counts and samples per handler

reset_errors()

Clear error tracking

Return type:

None

add_handler(handler, priority=0, filter_fn=None)

Add handler dynamically

Parameters:

• handler (Any) – Handler implementing AuditHandler protocol

• priority (int) – Execution priority (higher = first)

• filter_fn (Any) – Optional filter function

Return type:

None

remove_handler(handler)

Remove handler

Parameters:

handler (Any) – Handler to remove

Return type:

bool

Returns:

True if handler was found and removed

get_handlers()

Get list of all handlers with their priority and filters

Return type:

list[tuple[Any, int, Any]]

Returns:

List of (handler, priority, filter_fn) tuples

close()

Close all handlers that support it

Return type:

None

NullHandler

class hfortix_core.audit.NullHandler

Bases: object

Null handler that does nothing

Useful for disabling audit logging without changing code.

Example

>>> from hfortix_core.audit import NullHandler
>>>
>>> # Disable audit logging
>>> handler = NullHandler()
>>> fgt = FortiOS("192.168.1.99", token="...", audit_handler=handler)

log_operation(operation)

Do nothing

Return type:

None

Parameters:

operation (dict[str, Any])

AuditFormatter (Protocol)

class hfortix_core.audit.AuditFormatter(*args, **kwargs)

Bases: Protocol

Protocol for audit log formatters

format(operation)

Format an operation dict as a string

Parameters:

operation (dict[str, Any]) – Operation data dictionary

Return type:

str

Returns:

Formatted string ready for output

JSONFormatter

class hfortix_core.audit.JSONFormatter(pretty=False, indent=2)

Bases: object

Format audit logs as JSON (default)

Outputs compact JSON on a single line, suitable for log aggregation systems like ELK, Splunk, or cloud logging services.

Example output:

{“timestamp”:”2026-01-02T14:23:45Z”,”method”:”POST”,…}

Parameters:

• pretty (bool)

• indent (int)

__init__(pretty=False, indent=2)

Initialize JSON formatter

Parameters:

• pretty (bool) – If True, format with indentation for readability

• indent (int) – Number of spaces for indentation (when pretty=True)

format(operation)

Format as JSON string

Return type:

str

Parameters:

operation (dict[str, Any])

SyslogFormatter

class hfortix_core.audit.SyslogFormatter(facility='LOCAL0', severity='INFO', app_name='hfortix', hostname=None)

Bases: object

Format audit logs for Syslog (RFC 5424)

Outputs Syslog-formatted messages with proper priority, timestamp, and structured data.

Format:

<PRI>VERSION TIMESTAMP HOSTNAME APP-NAME PROCID MSGID STRUCTURED-DATA MSG # noqa: E501

Example output:

<134>1 2026-01-02T14:23:45Z 192.168.1.99 hfortix - - - {“method”:”POST”,…} # noqa: E501

Parameters:

• facility (str)

• severity (str)

• app_name (str)

• hostname (str | None)

FACILITIES = {'AUTH': 4, 'AUTHPRIV': 10, 'CRON': 9, 'DAEMON': 3, 'FTP': 11, 'KERN': 0, 'LOCAL0': 16, 'LOCAL1': 17, 'LOCAL2': 18, 'LOCAL3': 19, 'LOCAL4': 20, 'LOCAL5': 21, 'LOCAL6': 22, 'LOCAL7': 23, 'LPR': 6, 'MAIL': 2, 'NEWS': 7, 'SYSLOG': 5, 'USER': 1, 'UUCP': 8}

SEVERITIES = {'ALERT': 1, 'CRIT': 2, 'DEBUG': 7, 'EMERG': 0, 'ERR': 3, 'INFO': 6, 'NOTICE': 5, 'WARNING': 4}

__init__(facility='LOCAL0', severity='INFO', app_name='hfortix', hostname=None)

Initialize Syslog formatter

Parameters:

• facility (str) – Syslog facility (LOCAL0-LOCAL7, USER, etc.)

• severity (str) – Syslog severity (INFO, WARNING, ERR, etc.)

• app_name (str) – Application name for syslog

• hostname (str | None) – Hostname to use (None = use from operation data)

format(operation)

Format as RFC 5424 syslog message

Priority is calculated as: (facility * 8) + severity

Return type:

str

Parameters:

operation (dict[str, Any])

CEFFormatter

class hfortix_core.audit.CEFFormatter(device_vendor='Fortinet', device_product='FortiGate', device_version='7.0')

Bases: object

Format audit logs as Common Event Format (CEF)

CEF is widely used by SIEM systems like ArcSight, Splunk, QRadar.

Format:

CEF:Version|Device Vendor|Device Product|Device Version|Signature ID|Name|Severity|Extension # noqa: E501

Example output:

CEF:0|Fortinet|FortiGate|7.0|API_OPERATION|FortiGate API Operation|5| act=POST dst=192.168.1.99 suser=api-token outcome=success …

Parameters:

• device_vendor (str)

• device_product (str)

• device_version (str)

SEVERITY_MAP = {'DELETE': 7, 'GET': 2, 'POST': 5, 'PUT': 5}

__init__(device_vendor='Fortinet', device_product='FortiGate', device_version='7.0')

Initialize CEF formatter

Parameters:

• device_vendor (str) – Vendor name

• device_product (str) – Product name

• device_version (str) – Product version

format(operation)

Format as CEF string

Return type:

str

Parameters:

operation (dict[str, Any])

AuditOperation (TypedDict)

class hfortix_core.audit.AuditOperation

Bases: TypedDict

Type definition for audit operation data

timestamp

ISO 8601 timestamp when operation occurred

request_id

Unique identifier for correlating related operations

method

HTTP method (GET, POST, PUT, DELETE)

endpoint

Full API endpoint path (e.g., ‘/api/v2/cmdb/firewall/policy’)

api_type

API category (cmdb, monitor, log, service)

path

Relative path within api_type

vdom

Virtual domain name (if applicable)

action

High-level action (create, update, delete, read, list)

object_type

Type of object being operated on (e.g., ‘firewall.policy’)

object_name

Name/identifier of specific object (if available)

data

Request payload (sanitized)

params

Query parameters (sanitized)

status_code

HTTP response status code

success

Whether operation succeeded

duration_ms

Operation duration in milliseconds

error

Error message (if operation failed)

user_context

Optional user-provided context dict

host

FortiGate device IP/hostname

read_only_mode

Whether operation was simulated in read-only mode

timestamp: str

request_id: str

method: str

endpoint: str

api_type: str

path: str

vdom: str | None

action: str

object_type: str

object_name: str | None

data: dict[str, Any] | None

params: dict[str, Any] | None

status_code: int

success: bool

duration_ms: int

error: str | None

user_context: dict[str, Any] | None

host: str

read_only_mode: bool

Request Hooks

Protocol-based hooks for intercepting and modifying API requests/responses.

BeforeRequestHook (Protocol)

class hfortix_core.hooks.BeforeRequestHook(*args, **kwargs)

Bases: Protocol

Protocol for before-request hooks

Hooks implementing this protocol are called BEFORE sending API requests. They can: - Validate request data - Transform request data - Add headers or parameters - Cancel requests (raise exception) - Log/audit request details

Example

>>> class ValidationHook:
...     def before_request(
...         self, context: dict[str, Any]
...     ) -> dict[str, Any]:
...         # Validate request
...         if 'ticket' not in context.get('user_context', {}):
...             raise ValueError("Change ticket required!")
...         return context
...
>>> hook = ValidationHook()
>>> fgt = FortiOS(
...     "192.168.1.99", token="...", before_request_hooks=[hook]
... )

before_request(context)

Called before sending API request

Parameters:

context (dict[str, Any]) – Request context (see RequestContext)

Return type:

dict[str, Any]

Returns:

Modified context (changes apply to request)

Raises:

Any exception to cancel the request –

Note

• Modify context[‘data’] or context[‘params’] to change request

• Raise exceptions to block requests

• Keep execution fast - this runs in request path

AfterRequestHook (Protocol)

class hfortix_core.hooks.AfterRequestHook(*args, **kwargs)

Bases: Protocol

Protocol for after-request hooks

Hooks implementing this protocol are called AFTER receiving API responses. They can: - Validate responses - Transform response data - Log successful operations - Trigger side effects - Cache results

Note: The existing audit_handler is a specialized after-request hook.

Example

>>> class CacheHook:
...     def after_request(
...         self, context: dict[str, Any], response: dict[str, Any]
...     ) -> dict[str, Any]:
...         # Cache GET responses
...         if context['method'] == 'GET':
...             self.cache[context['endpoint']] = response
...         return response

after_request(context, response)

Called after receiving API response (success only)

Parameters:

• context (dict[str, Any]) – Request context (see RequestContext)

• response (dict[str, Any]) – API response (mutable)

Return type:

dict[str, Any]

Returns:

Modified response (changes apply to caller)

Note

• Only called on successful requests (2xx status)

• Exceptions in hooks don’t cancel the response

• Keep execution fast - this runs in response path

RequestContext (TypedDict)

class hfortix_core.hooks.RequestContext

Bases: TypedDict

Request context passed to hooks

method

HTTP method (GET, POST, PUT, DELETE)

api_type

API category (cmdb, monitor, log, service)

path

Relative endpoint path

data

Request payload (mutable)

params

Query parameters (mutable)

vdom

Virtual domain

endpoint

Full endpoint path

request_id

Unique request identifier

user_context

User-provided context metadata

method: str

api_type: str

path: str

data: dict[str, Any] | None

params: dict[str, Any] | None

vdom: str | None

endpoint: str

request_id: str

user_context: dict[str, Any] | None