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)[source]

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)[source]

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)[source]

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)[source]

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)[source]

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()[source]

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()[source]

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()[source]

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()[source]

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()[source]

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()[source]

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()[source]

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)[source]

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, read_only=False, audit_handler=None, audit_callback=None, user_context=None, rate_limit_calls_per_min=None, rate_limit_calls_per_5min=None, rate_limit_calls_per_hour=None, rate_limit_errors_per_min=None, rate_limit_errors_per_5min=None, rate_limit_errors_per_hour=None, rate_limit=False, rate_limit_strategy='queue', rate_limit_max_requests=100, rate_limit_window_seconds=60.0, rate_limit_queue_size=100, rate_limit_queue_timeout=30.0, rate_limit_queue_overflow='block', circuit_breaker=False, circuit_breaker_half_open_calls=3, circuit_breaker_auto_retry=False, circuit_breaker_max_retries=3, circuit_breaker_retry_delay=5.0)[source]

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)

  • read_only (bool)

  • audit_handler (Optional[Any])

  • audit_callback (Optional[Any])

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

  • rate_limit_calls_per_min (Optional[int])

  • rate_limit_calls_per_5min (Optional[int])

  • rate_limit_calls_per_hour (Optional[int])

  • rate_limit_errors_per_min (Optional[int])

  • rate_limit_errors_per_5min (Optional[int])

  • rate_limit_errors_per_hour (Optional[int])

  • rate_limit (bool)

  • rate_limit_strategy (str)

  • rate_limit_max_requests (int)

  • rate_limit_window_seconds (float)

  • rate_limit_queue_size (int)

  • rate_limit_queue_timeout (float)

  • rate_limit_queue_overflow (str)

  • circuit_breaker (bool)

  • circuit_breaker_half_open_calls (int)

  • circuit_breaker_auto_retry (bool)

  • circuit_breaker_max_retries (int)

  • circuit_breaker_retry_delay (float)

__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, read_only=False, audit_handler=None, audit_callback=None, user_context=None, rate_limit_calls_per_min=None, rate_limit_calls_per_5min=None, rate_limit_calls_per_hour=None, rate_limit_errors_per_min=None, rate_limit_errors_per_5min=None, rate_limit_errors_per_hour=None, rate_limit=False, rate_limit_strategy='queue', rate_limit_max_requests=100, rate_limit_window_seconds=60.0, rate_limit_queue_size=100, rate_limit_queue_timeout=30.0, rate_limit_queue_overflow='block', circuit_breaker=False, circuit_breaker_half_open_calls=3, circuit_breaker_auto_retry=False, circuit_breaker_max_retries=3, circuit_breaker_retry_delay=5.0)[source]

Initialize base HTTP client with shared configuration

Parameters:

  • url (str) – Base URL for the API (required)

  • verify (bool) – Enable SSL certificate verification (default: True)

  • vdom (Optional[str]) – Virtual domain name (optional)

  • max_retries (int) – Maximum retry attempts (default: 3)

  • connect_timeout (float) – Connection timeout in seconds (default: 10.0)

  • read_timeout (float) – Read timeout in seconds (default: 300.0)

  • circuit_breaker_threshold (int) – DEPRECATED - use circuit_breaker=True

  • circuit_breaker_timeout (float) – DEPRECATED - use circuit_breaker=True

  • max_connections (int) – Maximum concurrent connections (default: 100)

  • max_keepalive_connections (int) – Maximum keepalive connections (default: 20)

  • adaptive_retry (bool) – Enable adaptive retry with backpressure detection (default: False). 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).

  • read_only (bool) – Enable read-only mode - simulate write operations without executing (default: False)

  • audit_handler (Optional[Any]) – Handler for audit logging (implements AuditHandler protocol). Essential for compliance.

  • audit_callback (Optional[Any]) – Custom callback function for audit logging. Alternative to audit_handler.

  • user_context (Optional[dict[str, Any]]) – Optional dict with user/application context to include in audit logs.

  • rate_limit_calls_per_min (Optional[int]) – Track calls per minute (tracking only)

  • rate_limit_calls_per_5min (Optional[int]) – Track calls per 5 minutes (tracking only)

  • rate_limit_calls_per_hour (Optional[int]) – Track calls per hour (tracking only)

  • rate_limit_errors_per_min (Optional[int]) – Track errors per minute (tracking only)

  • rate_limit_errors_per_5min (Optional[int]) – Track errors per 5 minutes (tracking only)

  • rate_limit_errors_per_hour (Optional[int]) – Track errors per hour (tracking only)

  • Enforcement (# Rate Limiting)

  • rate_limit (bool) – Enable rate limiting enforcement (default: False). When enabled, enforces request rate limits with queue.

  • rate_limit_strategy (str) – How to handle rate limit exceeded: “queue” - Queue requests (default) “drop” - Drop requests silently “raise” - Raise RateLimitExceededError

  • rate_limit_max_requests (int) – Max requests per window (default: 100)

  • rate_limit_window_seconds (float) – Time window in seconds (default: 60.0)

  • rate_limit_queue_size (int) – Max queued requests (default: 100)

  • rate_limit_queue_timeout (float) – Max wait time in queue (default: 30.0)

  • rate_limit_queue_overflow (str) – What to do when queue is full: “block” - Wait for space (default) “drop” - Drop request silently “raise” - Raise RateLimitQueueFullError

  • Breaker (# Circuit)

  • circuit_breaker (bool) – Enable circuit breaker (default: False). When disabled, no overhead. When enabled, trips open after consecutive failures to protect service.

  • circuit_breaker_half_open_calls (int) – Test calls in half-open state (default: 3)

  • circuit_breaker_auto_retry (bool) – When True, wait and retry instead of immediately raising CircuitBreakerOpenError (default: False)

  • circuit_breaker_max_retries (int) – Max auto-retry attempts when circuit open (default: 3)

  • circuit_breaker_retry_delay (float) – Seconds between auto-retry attempts (default: 5.0)

Return type:

None

get_retry_stats()[source]

Get retry statistics

Return type:

dict[str, Any]

get_circuit_breaker_state()[source]

Get current circuit breaker state

Return type:

dict[str, Any]

configure_endpoint_timeout(endpoint_pattern, connect_timeout=None, read_timeout=None)[source]

Configure custom timeout for specific endpoints

Return type:

None

Parameters:

  • endpoint_pattern (str)

  • connect_timeout (float | None)

  • read_timeout (float | None)

reset_circuit_breaker()[source]

Reset circuit breaker to closed state

Return type:

None

get_health_metrics()[source]

Get comprehensive health metrics including adaptive retry stats

Return type:

dict[str, Any]

Returns:

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

get_connection_stats()[source]

Get HTTP connection pool statistics (base implementation)

Base class provides minimal stats. Child classes override this to provide detailed connection pool metrics.

Returns:

  • circuit_breaker_state: Current circuit breaker state

  • consecutive_failures: Number of consecutive failures

  • last_failure_time: Timestamp of last failure

Return type:

Dictionary with basic connection statistics

Note

Child classes (HTTPClient, AsyncHTTPClient, etc.) override this to include additional metrics like active_requests, pool_exhaustion, etc.

Example

>>> stats = client.get_connection_stats()
>>> if stats['circuit_breaker_state'] == 'open':
...     print("Circuit breaker is open!")
get_rate_limit_status()[source]

Get current rate limit tracking status

Returns detailed rate limit statistics including: - Call counts in different time windows (last min, 5min, hour) - Error counts in different time windows - Total calls and errors since client creation - Configured limits - Whether current usage is within limits

Returns:

  • calls_last_min: API calls in last 60 seconds

  • calls_last_5min: API calls in last 300 seconds

  • calls_last_hour: API calls in last 3600 seconds

  • errors_last_min: Errors in last 60 seconds

  • errors_last_5min: Errors in last 300 seconds

  • errors_last_hour: Errors in last 3600 seconds

  • total_calls: Total API calls since creation

  • total_errors: Total errors since creation

  • limits: Dict of configured limits

  • within_limits: Boolean, True if all limits respected

Return type:

Dictionary with rate limit statistics

Note

This is for monitoring only - does NOT enforce rate limits. Configure limits via rate_limit_* parameters in __init__().

Example

>>> client = HTTPClient(
...     url="...",
...     token="...",
...     rate_limit_calls_per_min=100,
...     rate_limit_calls_per_hour=1000
... )
>>> status = client.get_rate_limit_status()
>>> print(f"Calls/min: {status['calls_last_min']}/100")
>>> print(f"Within limits: {status['within_limits']}")

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=5, circuit_breaker_timeout=60.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, rate_limit_calls_per_min=None, rate_limit_calls_per_5min=None, rate_limit_calls_per_hour=None, rate_limit_errors_per_min=None, rate_limit_errors_per_5min=None, rate_limit_errors_per_hour=None, rate_limit=False, rate_limit_strategy='queue', rate_limit_max_requests=100, rate_limit_window_seconds=60.0, rate_limit_queue_size=100, rate_limit_queue_timeout=30.0, rate_limit_queue_overflow='block', circuit_breaker=False, circuit_breaker_half_open_calls=3)[source]

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]])

  • rate_limit_calls_per_min (Optional[int])

  • rate_limit_calls_per_5min (Optional[int])

  • rate_limit_calls_per_hour (Optional[int])

  • rate_limit_errors_per_min (Optional[int])

  • rate_limit_errors_per_5min (Optional[int])

  • rate_limit_errors_per_hour (Optional[int])

  • rate_limit (bool)

  • rate_limit_strategy (str)

  • rate_limit_max_requests (int)

  • rate_limit_window_seconds (float)

  • rate_limit_queue_size (int)

  • rate_limit_queue_timeout (float)

  • rate_limit_queue_overflow (str)

  • circuit_breaker (bool)

  • circuit_breaker_half_open_calls (int)

__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=5, circuit_breaker_timeout=60.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, rate_limit_calls_per_min=None, rate_limit_calls_per_5min=None, rate_limit_calls_per_hour=None, rate_limit_errors_per_min=None, rate_limit_errors_per_5min=None, rate_limit_errors_per_hour=None, rate_limit=False, rate_limit_strategy='queue', rate_limit_max_requests=100, rate_limit_window_seconds=60.0, rate_limit_queue_size=100, rate_limit_queue_timeout=30.0, rate_limit_queue_overflow='block', circuit_breaker=False, circuit_breaker_half_open_calls=3)[source]

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 – 60.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”}

  • rate_limit_calls_per_min (int | None)

  • rate_limit_calls_per_5min (int | None)

  • rate_limit_calls_per_hour (int | None)

  • rate_limit_errors_per_min (int | None)

  • rate_limit_errors_per_5min (int | None)

  • rate_limit_errors_per_hour (int | None)

  • rate_limit (bool)

  • rate_limit_strategy (str)

  • rate_limit_max_requests (int)

  • rate_limit_window_seconds (float)

  • rate_limit_queue_size (int)

  • rate_limit_queue_timeout (float)

  • rate_limit_queue_overflow (str)

  • circuit_breaker (bool)

  • circuit_breaker_half_open_calls (int)

Raises:

  • ValueError – If parameters are invalid or both token and

  • username/password provided –

Return type:

None

login()[source]

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()[source]

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()[source]

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)[source]

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()[source]

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)[source]

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)

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)[source]

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)[source]

GET request returning binary data (for file downloads)

Parameters:
Return type:

bytes

Returns:

Raw binary response data

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

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)[source]

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)[source]

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')[source]

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)[source]

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')[source]

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')[source]

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)[source]

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()[source]

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()[source]

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: False for executed operations; blocked operations

use blocked_by_read_only: True instead

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': False
}
get_write_operations()[source]

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)[source]

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):

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

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

HTTPClientJSONRPC (Synchronous JSON-RPC, FortiManager/FortiAnalyzer)

hfortix_core.HTTPClientFMG is a backwards-compatibility alias for HTTPClientJSONRPC.

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=5, circuit_breaker_timeout=60.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, rate_limit_calls_per_min=None, rate_limit_calls_per_5min=None, rate_limit_calls_per_hour=None, rate_limit_errors_per_min=None, rate_limit_errors_per_5min=None, rate_limit_errors_per_hour=None, rate_limit=False, rate_limit_strategy='queue', rate_limit_max_requests=100, rate_limit_window_seconds=60.0, rate_limit_queue_size=100, rate_limit_queue_timeout=30.0, rate_limit_queue_overflow='block', circuit_breaker=False, circuit_breaker_half_open_calls=3)[source]

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]])

  • rate_limit_calls_per_min (Optional[int])

  • rate_limit_calls_per_5min (Optional[int])

  • rate_limit_calls_per_hour (Optional[int])

  • rate_limit_errors_per_min (Optional[int])

  • rate_limit_errors_per_5min (Optional[int])

  • rate_limit_errors_per_hour (Optional[int])

  • rate_limit (bool)

  • rate_limit_strategy (str)

  • rate_limit_max_requests (int)

  • rate_limit_window_seconds (float)

  • rate_limit_queue_size (int)

  • rate_limit_queue_timeout (float)

  • rate_limit_queue_overflow (str)

  • circuit_breaker (bool)

  • circuit_breaker_half_open_calls (int)

__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=5, circuit_breaker_timeout=60.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, rate_limit_calls_per_min=None, rate_limit_calls_per_5min=None, rate_limit_calls_per_hour=None, rate_limit_errors_per_min=None, rate_limit_errors_per_5min=None, rate_limit_errors_per_hour=None, rate_limit=False, rate_limit_strategy='queue', rate_limit_max_requests=100, rate_limit_window_seconds=60.0, rate_limit_queue_size=100, rate_limit_queue_timeout=30.0, rate_limit_queue_overflow='block', circuit_breaker=False, circuit_breaker_half_open_calls=3)[source]

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”}

  • rate_limit (bool) – Enable rate limiting with token bucket algorithm (default: False). When enabled, enforces request limits with configurable queue and overflow strategies. Zero overhead when disabled.

  • rate_limit_strategy (str) – Rate limiting strategy (default: “queue”). Currently only “queue” is supported - uses token bucket algorithm with FIFO queue for overflow handling.

  • rate_limit_max_requests (int) – Maximum requests allowed per window (default: 100). Controls token bucket capacity and refill rate (tokens/second = max_requests/window_seconds).

  • rate_limit_window_seconds (float) – Time window in seconds for rate limit (default: 60.0). Tokens refill at max_requests/window_seconds per second.

  • rate_limit_queue_size (int) – Maximum requests to queue when rate limit exceeded (default: 100). Set to 0 to disable queuing (requests dropped immediately with clear message).

  • rate_limit_queue_timeout (float) – Maximum seconds to wait in queue (default: 30.0). Raises RateLimitQueueTimeoutError if exceeded.

  • rate_limit_queue_overflow (str) –

    Queue overflow strategy (default:

    ”block”):

    • ”block”: Block/wait until space available (respects

    queue_timeout) - “drop”: Drop the new request silently (return False) - “raise”: Raise RateLimitQueueFullError

  • circuit_breaker (bool) – Enable circuit breaker pattern (default: False). When enabled, opens circuit after threshold failures to prevent cascading failures. Requires successful test requests in half-open state before closing. BREAKING CHANGE: Previously always enabled, now opt-in for zero overhead.

  • circuit_breaker_half_open_calls (int) – Number of successful requests required in half-open state before closing circuit (default: 3). Prevents premature closing on transient recovery.

  • rate_limit_calls_per_min (int | None)

  • rate_limit_calls_per_5min (int | None)

  • rate_limit_calls_per_hour (int | None)

  • rate_limit_errors_per_min (int | None)

  • rate_limit_errors_per_5min (int | None)

  • rate_limit_errors_per_hour (int | None)

Raises:

  • ValueError – If parameters are invalid or both token and

  • username/password provided –

Return type:

None

async login()[source]

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()[source]

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()[source]

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()[source]

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)[source]

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)[source]

Async GET request

Return type:

dict[str, Any]

Parameters:
async get_binary(api_type, path, params=None, vdom=None)[source]

Async GET request returning binary data

Return type:

bytes

Parameters:
async post(api_type, path, data, params=None, vdom=None, raw_json=False)[source]

Async POST request - Create new object

Return type:

dict[str, Any]

Parameters:
async put(api_type, path, data, params=None, vdom=None, raw_json=False)[source]

Async PUT request - Update existing object

Return type:

dict[str, Any]

Parameters:
async delete(api_type, path, params=None, vdom=None, raw_json=False)[source]

Async DELETE request - Delete object

Return type:

dict[str, Any]

Parameters:
static validate_mkey(mkey, parameter_name='mkey')[source]

Validate and convert mkey to string

Return type:

str

Parameters:

  • mkey (Any)

  • parameter_name (str)

static validate_required_params(params, required)[source]

Validate that required parameters are present

Return type:

None

Parameters:
static validate_range(value, min_val, max_val, parameter_name='value')[source]

Validate that a numeric value is within a specified range

Return type:

None

Parameters:
static validate_choice(value, choices, parameter_name='value')[source]

Validate that a value is one of the allowed choices

Return type:

None

Parameters:
static build_params(**kwargs)[source]

Build parameters dict, filtering out None values

Return type:

dict[str, Any]

Parameters:

kwargs (Any)

async close()[source]

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

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

Example

client = AsyncHTTPClient(…) try:

…

finally:

await client.close()

get_operations()[source]

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()[source]

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)[source]

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):

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

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

CloudHTTPClient (FortiCloud REST)

Used by hfortix-forticare and hfortix-fortiztp for OAuth2-authenticated FortiCloud REST APIs.

class hfortix_core.http.CloudHTTPClient(url, oauth_token, verify=True, 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, user_agent=None, read_only=False, track_operations=False, audit_handler=None, audit_callback=None, user_context=None, token_callback=None, rate_limit_calls_per_min=None, rate_limit_calls_per_5min=None, rate_limit_calls_per_hour=None, rate_limit_errors_per_min=None, rate_limit_errors_per_5min=None, rate_limit_errors_per_hour=None, rate_limit=False, rate_limit_strategy='queue', rate_limit_max_requests=100, rate_limit_window_seconds=60.0, rate_limit_queue_size=100, rate_limit_queue_timeout=30.0, rate_limit_queue_overflow='block', circuit_breaker=False, circuit_breaker_half_open_calls=3, circuit_breaker_auto_retry=False, circuit_breaker_max_retries=3, circuit_breaker_retry_delay=5.0)[source]

Bases: BaseHTTPClient

HTTP client for Fortinet Cloud APIs with OAuth 2.0 authentication.

Designed for cloud services like FortiCare Asset Management API v3, FortiCloud, and other OAuth-protected Fortinet cloud endpoints.

Key Differences from HTTPClient: - Uses OAuth 2.0 Bearer tokens (Authorization: Bearer <token>) - No API key authentication - No VDOM support - Cloud-specific error handling - Rate limiting aware (100/min, 1000/hour for FortiCare)

Authentication:

The client expects an OAuth access token obtained from: https://customerapiauth.fortinet.com/api/v1/oauth/token

Rate Limits (FortiCare Asset Management):

  • 100 calls per minute

  • 1000 calls per hour

  • 10 errors per hour

  • Batch operations: max 10 units, max 5 errors per batch

Example

>>> client = CloudHTTPClient(
...     url="https://support.fortinet.com",
...     oauth_token="your_oauth_token_here"
... )
>>> response = client.get("/ES/api/registration/v3/products/list")
>>> client.logout()

Note

The interface differs from HTTPClient — no vdom/api_type params, and methods return a response envelope dict rather than the raw JSON body.

Parameters:

  • url (str)

  • oauth_token (str)

  • verify (bool)

  • 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)

  • user_agent (Optional[str])

  • read_only (bool)

  • track_operations (bool)

  • audit_handler (Optional[Any])

  • audit_callback (Optional[Any])

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

  • token_callback (Optional[Callable[[], str]])

  • rate_limit_calls_per_min (Optional[int])

  • rate_limit_calls_per_5min (Optional[int])

  • rate_limit_calls_per_hour (Optional[int])

  • rate_limit_errors_per_min (Optional[int])

  • rate_limit_errors_per_5min (Optional[int])

  • rate_limit_errors_per_hour (Optional[int])

  • rate_limit (bool)

  • rate_limit_strategy (str)

  • rate_limit_max_requests (int)

  • rate_limit_window_seconds (float)

  • rate_limit_queue_size (int)

  • rate_limit_queue_timeout (float)

  • rate_limit_queue_overflow (str)

  • circuit_breaker (bool)

  • circuit_breaker_half_open_calls (int)

  • circuit_breaker_auto_retry (bool)

  • circuit_breaker_max_retries (int)

  • circuit_breaker_retry_delay (float)

__init__(url, oauth_token, verify=True, 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, user_agent=None, read_only=False, track_operations=False, audit_handler=None, audit_callback=None, user_context=None, token_callback=None, rate_limit_calls_per_min=None, rate_limit_calls_per_5min=None, rate_limit_calls_per_hour=None, rate_limit_errors_per_min=None, rate_limit_errors_per_5min=None, rate_limit_errors_per_hour=None, rate_limit=False, rate_limit_strategy='queue', rate_limit_max_requests=100, rate_limit_window_seconds=60.0, rate_limit_queue_size=100, rate_limit_queue_timeout=30.0, rate_limit_queue_overflow='block', circuit_breaker=False, circuit_breaker_half_open_calls=3, circuit_breaker_auto_retry=False, circuit_breaker_max_retries=3, circuit_breaker_retry_delay=5.0)[source]

Initialize Cloud HTTP client.

Parameters:

  • url (str) – Base URL of the cloud service (e.g., “https://support.fortinet.com”)

  • oauth_token (str) – OAuth 2.0 Bearer token for authentication

  • verify (bool) – Enable SSL certificate verification (default: True)

  • max_retries (int) – Maximum number of retry attempts (default: 3)

  • connect_timeout (float) – Connection timeout in seconds (default: 10.0)

  • read_timeout (float) – Read timeout in seconds (default: 300.0)

  • circuit_breaker_threshold (int) – Failures before circuit opens (default: 5)

  • circuit_breaker_timeout (float) – Circuit breaker timeout in seconds (default: 60.0)

  • max_connections (int) – Maximum number of connections (default: 100)

  • max_keepalive_connections (int) – Max keepalive connections (default: 20)

  • adaptive_retry (bool) – Enable adaptive retry based on response times

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

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

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

  • read_only (bool) – Enable read-only mode - simulate write operations without executing (default: False)

  • track_operations (bool) – Enable operation tracking - maintain audit log of all API calls (default: False)

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

  • audit_callback (Optional[Any]) – Custom callback function for audit logging (alternative to audit_handler)

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

  • token_callback (Optional[Callable[[], str]]) – Optional callback to get fresh token before each request (returns str) Useful with CloudSession to ensure token is valid before each request

  • rate_limit (bool) – Enable rate limiting enforcement (default: False)

  • rate_limit_strategy (str) – ‘queue’, ‘drop’, or ‘raise’ (default: ‘queue’)

  • rate_limit_max_requests (int) – Max requests per window (default: 100)

  • rate_limit_window_seconds (float) – Time window in seconds (default: 60.0)

  • rate_limit_queue_size (int) – Max queue size (default: 100)

  • rate_limit_queue_timeout (float) – Max wait time in queue (default: 30.0)

  • rate_limit_queue_overflow (str) – ‘block’ or ‘drop’ on overflow (default: ‘block’)

  • circuit_breaker (bool) – Enable circuit breaker (default: False)

  • circuit_breaker_half_open_calls (int) – Calls to test in half-open state (default: 3)

  • circuit_breaker_auto_retry (bool) – Wait and retry instead of raising immediately when circuit breaker is open (default: False)

  • circuit_breaker_max_retries (int) – Max auto-retry attempts when circuit open (default: 3)

  • circuit_breaker_retry_delay (float) – Seconds between auto-retry attempts (default: 5.0)

  • rate_limit_calls_per_min (int | None)

  • rate_limit_calls_per_5min (int | None)

  • rate_limit_calls_per_hour (int | None)

  • rate_limit_errors_per_min (int | None)

  • rate_limit_errors_per_5min (int | None)

  • rate_limit_errors_per_hour (int | None)

Raises:

ValueError – If oauth_token is empty or invalid parameters

Return type:

None

get(path, params=None, timeout=None)[source]

Send GET request to cloud API.

Parameters:

  • path (str) – API endpoint path (e.g., “/ES/api/registration/v3/products/list”)

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

  • timeout (Optional[float]) – Override default timeout in seconds (optional)

Returns:

  • data: JSON response body

  • http_status_code: HTTP status code

  • response_time: Response time in seconds

  • request_info: Request metadata (method, url, params)

Return type:

Response envelope containing

Raises:

  • httpx.HTTPStatusError – For HTTP error responses

  • httpx.TimeoutException – If request times out

  • httpx.RequestError – For network errors

post(path, data=None, params=None, timeout=None)[source]

Send POST request to cloud API.

Parameters:
Returns:

  • data: JSON response body

  • http_status_code: HTTP status code

  • response_time: Response time in seconds

  • request_info: Request metadata (method, url, params, data)

Return type:

Response envelope containing

Raises:

  • httpx.HTTPStatusError – For HTTP error responses

  • httpx.TimeoutException – If request times out

  • httpx.RequestError – For network errors

put(path, data=None, params=None, timeout=None)[source]

Send PUT request to cloud API.

Parameters:
Returns:

  • data: JSON response body

  • http_status_code: HTTP status code

  • response_time: Response time in seconds

  • request_info: Request metadata (method, url, params, data)

Return type:

Response envelope containing

Raises:

  • httpx.HTTPStatusError – For HTTP error responses

  • httpx.TimeoutException – If request times out

  • httpx.RequestError – For network errors

delete(path, params=None, timeout=None)[source]

Send DELETE request to cloud API.

Parameters:

  • path (str) – API endpoint path

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

  • timeout (Optional[float]) – Override default timeout in seconds (optional)

Returns:

  • data: JSON response body

  • http_status_code: HTTP status code

  • response_time: Response time in seconds

  • request_info: Request metadata (method, url, params)

Return type:

Response envelope containing

Raises:

  • httpx.HTTPStatusError – For HTTP error responses

  • httpx.TimeoutException – If request times out

  • httpx.RequestError – For network errors

get_operations()[source]

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)

  • path: API endpoint path

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

  • status_code: HTTP response status code

  • read_only_simulated: True if operation was simulated in read-only mode

Return type:

List of operation dictionaries with keys

Example

>>> client = CloudHTTPClient(
...     url="https://support.fortinet.com",
...     oauth_token="...",
...     track_operations=True
... )
>>> client.post("/api/v3/products/list", data={"serial_number": "FGT*"})
>>> ops = client.get_operations()
>>> print(ops[0])
{
    'timestamp': '2026-02-06T10:30:15Z',
    'method': 'POST',
    'path': '/api/v3/products/list',
    'data': {'serial_number': 'FGT*'},
    'status_code': 200,
    'read_only_simulated': False
}
get_write_operations()[source]

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 = CloudHTTPClient(
...     url="https://support.fortinet.com",
...     oauth_token="...",
...     track_operations=True
... )
>>> client.get("/api/v3/products/list")  # GET - excluded
>>> client.post("/api/v3/products/register", data={...})  # POST - included
>>> client.delete("/api/v3/products/123")  # DELETE - included
>>> write_ops = client.get_write_operations()
>>> len(write_ops)  # Returns 2 (POST and DELETE only)
2
get_connection_stats()[source]

Get connection pool statistics.

Returns:

  • http2_enabled: Whether HTTP/2 is enabled

  • max_connections: Maximum allowed connections

  • max_keepalive_connections: Maximum keepalive connections

  • active_requests: Number of currently active requests

  • total_requests: Total number of requests made

  • client_active: Whether HTTP session is initialized

  • circuit_breaker_state: Current circuit breaker state

  • consecutive_failures: Number of consecutive failures

  • last_failure_time: Timestamp of last failure

Return type:

Dictionary with connection pool metrics

Example

>>> client = CloudHTTPClient(url="https://support.fortinet.com", oauth_token="...")
>>> stats = client.get_connection_stats()
>>> print(f"Active: {stats['active_requests']}/{stats['max_connections']}")
Active: 2/100
inspect_last_request()[source]

Get detailed information about the last HTTP request/response.

Useful for debugging and understanding what was sent/received.

Returns:

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

    • endpoint: API endpoint path (without query string)

    • url: Full URL with query string

    • params: Query parameters

    • response_time_ms: Response time in milliseconds

    • status_code: HTTP status code (if response available)

Or {“error”: “…”} if no requests have been made yet.

Return type:

Dictionary with last request details

Example

>>> client = CloudHTTPClient(url="https://support.fortinet.com", oauth_token="...")
>>> client.get("/api/v3/products/list")
>>> last = client.inspect_last_request()
>>> print(f"Last request took {last['response_time_ms']}ms")
Last request took 234.5ms
close()[source]

Close the HTTP session and clean up resources.

Alias for logout() — conforms to the standard client interface.

Return type:

None

logout()[source]

Close the HTTP session and clean up resources.

Note

OAuth token revocation should be handled separately via the authentication service. This method only closes the HTTP connection.

Return type:

None

FortiCloudAuth (OAuth2 Authentication)

class hfortix_core.http.FortiCloudAuth(api_id, password, client_id='assetmanagement', auth_url=None)[source]

Bases: object

FortiCloud OAuth 2.0 Authentication Helper.

Handles OAuth token acquisition from FortiCloud authentication API for various Fortinet cloud services.

Supported Services:

  • assetmanagement: FortiCare Asset Management

  • FortiManager: FortiManager Cloud

  • FortiAnalyzer: FortiAnalyzer Cloud

  • fortigatecloud: FortiGate Cloud

  • fortipresence: FortiPresence Cloud

  • And more…

Example

>>> from hfortix_core.http.oauth import FortiCloudAuth
>>>
>>> auth = FortiCloudAuth(
...     api_id="your_api_id",
...     password="your_password",
...     client_id="assetmanagement"
... )
>>> token = auth.get_token()
>>> print(f"Token: {token[:20]}...")
Parameters:

  • api_id (str)

  • password (str)

  • client_id (str)

  • auth_url (Optional[str])

DEFAULT_AUTH_URL = 'https://customerapiauth.fortinet.com/api/v1/oauth/token/'
__init__(api_id, password, client_id='assetmanagement', auth_url=None)[source]

Initialize FortiCloud authentication helper.

Parameters:

  • api_id (str) – FortiCloud API ID (username)

  • password (str) – FortiCloud API password

  • client_id (str) – Client ID for the service (default: assetmanagement)

  • auth_url (Optional[str]) – Authentication URL (default: FortiCloud OAuth endpoint)

Raises:

ValueError – If api_id or password is empty

get_token(force_refresh=False)[source]

Get OAuth access token.

Requests a new OAuth token from FortiCloud authentication API. Caches the token for reuse unless force_refresh is True.

Parameters:

force_refresh (bool) – Force request new token even if cached (default: False)

Return type:

str

Returns:

OAuth access token string

Raises:

  • httpx.HTTPError – If authentication request fails

  • KeyError – If response doesn’t contain access_token

Example

>>> auth = FortiCloudAuth(api_id="...", password="...")
>>> token = auth.get_token()
clear_token()[source]

Clear cached token.

Use this to force a new token request on the next get_token() call.

Return type:

None

get_oauth_token

hfortix_core.http.get_oauth_token(api_id, password, client_id='assetmanagement', auth_url=None)[source]

Convenience function to get OAuth token.

This is a simplified wrapper around FortiCloudAuth for one-time token acquisition without needing to manage the auth object.

Parameters:

  • api_id (str) – FortiCloud API ID (username)

  • password (str) – FortiCloud API password

  • client_id (str) – Client ID for the service (default: assetmanagement)

  • auth_url (Optional[str]) – Authentication URL (default: FortiCloud OAuth endpoint)

Return type:

str

Returns:

OAuth access token string

Raises:

  • httpx.HTTPError – If authentication request fails

  • KeyError – If response doesn’t contain access_token

Example

>>> from hfortix_core.http.oauth import get_oauth_token
>>>
>>> token = get_oauth_token(
...     api_id="your_api_id",
...     password="your_password",
...     client_id="assetmanagement"
... )
>>> print(f"Token: {token}")

Caching

TTLCache

readonly_cache

Logging

RequestLogger

log_operation

StructuredFormatter

TextFormatter

LogFormatter (Protocol)

class hfortix_core.logging.LogFormatter(*args, **kwargs)[source]

Bases: Protocol

Protocol for log formatters

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

format(record)[source]

Format a log record

Parameters:

record (Any) – logging.LogRecord instance

Return type:

str

Returns:

Formatted string

LogRecord (TypedDict)

class hfortix_core.logging.LogRecord[source]

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

debug_timer

format_connection_stats

format_request_info

DebugFormatter (Protocol)

class hfortix_core.debug.DebugFormatter(*args, **kwargs)[source]

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)[source]

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)[source]

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[source]

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[source]

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)[source]

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=', ')[source]

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)[source]

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')[source]

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)[source]

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=', ')[source]

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=' | ')[source]

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)[source]

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)[source]

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')[source]

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)[source]

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)[source]

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)[source]

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)

SyslogHandler

FileHandler

StreamHandler

CompositeHandler

NullHandler

AuditFormatter (Protocol)

JSONFormatter

SyslogFormatter

CEFFormatter

AuditOperation (TypedDict)

Request Hooks

The hfortix_core.hooks module defines protocol classes (BeforeRequestHook, AfterRequestHook, RequestContext) that are reserved for a future request-interception feature. They are not yet consumed by any HTTP client — passing hook objects to client constructors has no effect in the current release.