Core Package API Reference#

The hfortix-core package provides the foundation for all HFortix packages, including HTTP client functionality, exception handling, and retry logic.

Installation#

pip install hfortix-core

Core Modules#

Exceptions#

FortiOS-Specific Exceptions FortiOS error codes and product-specific exception handling

exception hfortix_core.exceptions.FortinetError[source]#

Bases: Exception

Base exception for all Fortinet API errors

exception hfortix_core.exceptions.APIError[source]#

Bases: FortinetError

Generic API error with optional metadata

message#

Error message

http_status#

HTTP status code (e.g., 400, 404, 500)

error_code#

FortiOS internal error code (e.g., -5, -3)

response#

Full API response dict

endpoint#

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

method#

HTTP method (GET, POST, PUT, DELETE)

params#

Request parameters (sanitized)

hint#

Helpful suggestion for resolving the error

request_id#

Unique identifier for this request

timestamp#

ISO 8601 timestamp when error occurred

__init__(message, http_status=None, error_code=None, response=None, endpoint=None, method=None, params=None, hint=None, request_id=None)[source]#
__str__()[source]#

Format error with full context for better debugging

Return type:

str

__repr__()[source]#

Developer-friendly representation for debugging

Return type:

str

exception hfortix_core.exceptions.AuthenticationError[source]#

Bases: FortinetError

HTTP 401 - Authentication failed (invalid credentials)

exception hfortix_core.exceptions.AuthorizationError[source]#

Bases: FortinetError

HTTP 403 - Authorization failed (insufficient permissions)

exception hfortix_core.exceptions.RetryableError[source]#

Bases: APIError

Base exception for errors that should trigger automatic retry

These errors are typically transient and may succeed on retry: - Rate limiting (429) - Service unavailable (503) - Timeouts - Circuit breaker open

exception hfortix_core.exceptions.NonRetryableError[source]#

Bases: APIError

Base exception for errors that should NOT be retried

These errors indicate client-side mistakes or permanent failures: - Bad request (400) - Resource not found (404) - Duplicate entry - Entry in use - Permission denied

exception hfortix_core.exceptions.ConfigurationError[source]#

Bases: FortinetError

Raised when FortiOS instance is misconfigured

Examples: - Both token and username/password provided - Missing required authentication - Invalid parameter combinations

exception hfortix_core.exceptions.VDOMError[source]#

Bases: FortinetError

Raised when VDOM operation fails or VDOM doesn’t exist

vdom#

The VDOM name that caused the error

Parameters:
__init__(message, vdom)[source]#
Parameters:
exception hfortix_core.exceptions.OperationNotSupportedError[source]#

Bases: FortinetError

Raised when attempting unsupported operation on endpoint

operation#

The operation that was attempted (e.g., ‘DELETE’)

endpoint#

The endpoint that doesn’t support it

Parameters:

  • message (str)

  • operation (str)

  • endpoint (str)

__init__(message, operation, endpoint)[source]#
Parameters:

  • message (str)

  • operation (str)

  • endpoint (str)

exception hfortix_core.exceptions.ReadOnlyModeError[source]#

Bases: FortinetError

Operation blocked by read-only mode

Raised when attempting POST/PUT/DELETE operations with read_only=True. This is a client-side block to prevent accidental writes in safe mode.

exception hfortix_core.exceptions.BadRequestError[source]#

Bases: NonRetryableError

HTTP 400 - Bad Request

__init__(message='Bad request', **kwargs)[source]#
exception hfortix_core.exceptions.ResourceNotFoundError[source]#

Bases: NonRetryableError

HTTP 404 - Resource not found

__init__(message='Resource not found', **kwargs)[source]#
suggest_recovery()[source]#

Suggest how to recover from this error

Return type:

str

exception hfortix_core.exceptions.MethodNotAllowedError[source]#

Bases: NonRetryableError

HTTP 405 - Method not allowed

__init__(message='Method not allowed', **kwargs)[source]#
exception hfortix_core.exceptions.RateLimitError[source]#

Bases: RetryableError

HTTP 429 - Rate limit exceeded

__init__(message='Rate limit exceeded', **kwargs)[source]#
exception hfortix_core.exceptions.ServerError[source]#

Bases: RetryableError

HTTP 500 - Internal server error

__init__(message='Internal server error', **kwargs)[source]#
exception hfortix_core.exceptions.ServiceUnavailableError[source]#

Bases: RetryableError

HTTP 503 - Service temporarily unavailable

__init__(message='Service temporarily unavailable', **kwargs)[source]#
exception hfortix_core.exceptions.CircuitBreakerOpenError[source]#

Bases: RetryableError

Circuit breaker is open - service appears to be down

__init__(message='Circuit breaker is open', **kwargs)[source]#
exception hfortix_core.exceptions.TimeoutError[source]#

Bases: RetryableError

Request timed out

__init__(message='Request timed out', **kwargs)[source]#
exception hfortix_core.exceptions.DuplicateEntryError[source]#

Bases: NonRetryableError

Duplicate entry exists (error code -5, -15, -100, etc.)

__init__(message='A duplicate entry already exists', **kwargs)[source]#
suggest_recovery()[source]#

Suggest how to recover from this error

Return type:

str

exception hfortix_core.exceptions.EntryInUseError[source]#

Bases: NonRetryableError

Entry cannot be deleted because it’s in use (error code -23, -94, -95, etc.)

__init__(message='Entry is in use and cannot be deleted', **kwargs)[source]#
suggest_recovery()[source]#

Suggest how to recover from this error

Return type:

str

exception hfortix_core.exceptions.InvalidValueError[source]#

Bases: NonRetryableError

Invalid value provided (error code -651, -1, -50, etc.)

__init__(message='Input value is invalid', **kwargs)[source]#
exception hfortix_core.exceptions.PermissionDeniedError[source]#

Bases: NonRetryableError

Permission denied, insufficient privileges (error code -14, -37)

__init__(message='Permission denied. Insufficient privileges.', **kwargs)[source]#
hfortix_core.exceptions.get_error_description(error_code)[source]#

Get human-readable description for FortiOS error code

Parameters:

error_code (int) – FortiOS error code

Returns:

Error description or “Unknown error”

Return type:

str

Examples

>>> get_error_description(-5)
'A duplicate entry already exists'
>>> get_error_description(-651)
'Input value is invalid'
hfortix_core.exceptions.get_http_status_description(status_code)[source]#

Get human-readable description for HTTP status code

Parameters:

status_code (int) – HTTP status code

Returns:

Status description or “Unknown status code”

Return type:

str

hfortix_core.exceptions.raise_for_status(response, endpoint=None, method=None, params=None)[source]#

Raise appropriate exception based on FortiOS API response

Parameters:

  • response (dict) – API response dictionary

  • endpoint (str) – Optional API endpoint for better error context

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

  • params (dict) – Optional request parameters (will be sanitized)

Raises:

APIError – If response indicates an error

Return type:

None

Examples

>>> response = {'status': 'error', 'http_status': 404, 'error': -3}
>>> raise_for_status(
...     response,
...     endpoint='/api/v2/cmdb/firewall/address',
...     method='GET'
... )
# Raises ResourceNotFoundError with helpful context and tip
hfortix_core.exceptions.is_retryable_error(error)[source]#

Check if error should trigger automatic retry

Parameters:

error (Exception) – Exception to check

Return type:

bool

Returns:

True if error is retryable, False otherwise

Example

>>> try:
...     fgt.api.cmdb.firewall.policy.get()
... except Exception as e:
...     if is_retryable_error(e):
...         time.sleep(5)
...         # retry logic here
hfortix_core.exceptions.get_retry_delay(error, attempt, base_delay=1.0, max_delay=60.0)[source]#

Calculate appropriate retry delay based on error type and attempt number

Parameters:

  • error (Exception) – Exception that occurred

  • attempt (int) – Retry attempt number (1, 2, 3, …)

  • base_delay (float, default: 1.0) – Base delay in seconds (default: 1.0)

  • max_delay (float, default: 60.0) – Maximum delay in seconds (default: 60.0)

Return type:

float

Returns:

Recommended delay in seconds

Example

>>> for attempt in range(1, 4):
...     try:
...         result = fgt.api.cmdb.firewall.policy.get()
...         break
...     except Exception as e:
...         if is_retryable_error(e):
...             delay = get_retry_delay(e, attempt)
...             time.sleep(delay)
...         else:
...             raise

Exception Hierarchy#

All HFortix exceptions inherit from FortinetError:

FortinetError (Base exception)
├── APIError (Base for all API errors)
│   ├── HTTPError (Base for HTTP errors)
│   │   ├── BadRequestError (400)
│   │   ├── UnauthorizedError (401)
│   │   ├── ForbiddenError (403)
│   │   ├── ResourceNotFoundError (404)
│   │   ├── MethodNotAllowedError (405)
│   │   ├── RateLimitError (429)
│   │   └── ServerError (500+)
│   ├── DuplicateEntryError (FortiOS: object exists)
│   ├── EntryInUseError (FortiOS: object in use)
│   └── ValidationError (Invalid input)
├── ConnectionError (Network/connection issues)
├── TimeoutError (Request timeout)
└── ConfigurationError (Invalid configuration)

HTTP Client#

Base HTTP Client Interface#

HTTP Client Interface (Protocol)

This module defines the Protocol interface for HTTP clients used by FortiOS API endpoints. Using a Protocol (PEP 544) allows for duck-typing and enables users to provide custom HTTP client implementations.

Benefits: - Users can provide custom HTTP clients (with caching, custom auth, proxying, etc.) - Easier testing with lightweight fake/mock clients - Type-safe client swapping - Maintains backward compatibility

Example

class CustomHTTPClient:

‘’’Custom client with company-specific requirements’’’ def get(self, api_type: str, path: str, **kwargs) -> Union[dict[str, Any], Coroutine[Any, Any, dict[str, Any]]]:

# Custom implementation with logging, auth, etc. …

def post(self, api_type: str, path: str, data: dict, **kwargs) -> Union[dict[str, Any], Coroutine[Any, Any, dict[str, Any]]]:

# Custom implementation …

# … put, delete

# Use custom client fgt = FortiOS(host=’…’, token=’…’, client=CustomHTTPClient())

class hfortix_core.http.interface.IHTTPClient[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]], default: None) – Optional query parameters (filters, pagination, etc.)

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

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

  • False

  • results (return only)

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

  • action (Optional[str], default: None) – 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]], default: None) – Optional query parameters

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

  • raw_json (bool, default: False) – 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]], default: None) – Optional query parameters

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

  • raw_json (bool, default: False) – 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]], default: None) – Optional query parameters

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

  • raw_json (bool, default: False) – 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]], default: None) – Optional query parameters

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

Return type:

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

Returns:

Raw binary response data (bytes)

__init__(*args, **kwargs)#

Synchronous HTTP Client#

Internal HTTP Client for FortiOS API

This module contains the HTTPClient class which handles all HTTP communication with FortiGate devices. It is an internal implementation detail and not part of the public API.

Now powered by httpx for better performance, HTTP/2 support, and modern async capabilities.

class hfortix_core.http.client.HTTPClient[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, default: True)

  • token (Optional[str], default: None)

  • username (Optional[str], default: None)

  • password (Optional[str], default: None)

  • vdom (Optional[str], default: None)

  • max_retries (int, default: 3)

  • connect_timeout (float, default: 10.0)

  • read_timeout (float, default: 300.0)

  • user_agent (Optional[str], default: None)

  • circuit_breaker_threshold (int, default: 10)

  • circuit_breaker_timeout (float, default: 30.0)

  • circuit_breaker_auto_retry (bool, default: False)

  • circuit_breaker_max_retries (int, default: 3)

  • circuit_breaker_retry_delay (float, default: 5.0)

  • max_connections (int, default: 100)

  • max_keepalive_connections (int, default: 20)

  • session_idle_timeout (Union[int, float, None], default: 300)

  • read_only (bool, default: False)

  • track_operations (bool, default: False)

  • adaptive_retry (bool, default: False)

  • retry_strategy (str, default: 'exponential')

  • retry_jitter (bool, default: False)

  • audit_handler (Optional[Any], default: None)

  • audit_callback (Optional[Any], default: None)

  • user_context (Optional[dict[str, Any]], default: None)

Initialize HTTP client

Parameters:

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

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

  • token (Optional[str], default: None) – API authentication token (if using token auth)

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

  • auth)

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

  • auth)

  • vdom (Optional[str], default: None) – Default virtual domain

  • max_retries (int, default: 3) – Maximum number of retry attempts on transient failures

  • (default (all API calls) –

  • connect_timeout (float, default: 10.0) – Timeout for establishing connection in seconds

  • (default – 10.0)

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

  • 300.0)

  • user_agent (Optional[str], default: None) – 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, default: 10) – Number of consecutive failures before

  • (default –

  • circuit_breaker_timeout (float, default: 30.0) – Seconds to wait before transitioning to

  • (default – 30.0)

  • circuit_breaker_auto_retry (bool, default: False) – 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, default: 3) – Maximum retry attempts when

  • (default –

  • circuit_breaker_retry_delay (float, default: 5.0) – 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, default: 100) – Maximum number of connections in the pool

  • (default –

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

  • (default –

  • session_idle_timeout (Union[int, float, None], default: 300) – 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, default: False) – Enable read-only mode - simulate write operations

  • (default – False)

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

  • (default – False)

  • adaptive_retry (bool, default: False) – 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, default: 'exponential') – 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, default: False) – 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], default: None) – 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], default: None) – 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]], default: None) – 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 –

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

Initialize HTTP client

Parameters:

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

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

  • token (Optional[str], default: None) – API authentication token (if using token auth)

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

  • auth)

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

  • auth)

  • vdom (Optional[str], default: None) – Default virtual domain

  • max_retries (int, default: 3) – Maximum number of retry attempts on transient failures

  • (default (all API calls) –

  • connect_timeout (float, default: 10.0) – Timeout for establishing connection in seconds

  • (default – 10.0)

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

  • 300.0)

  • user_agent (Optional[str], default: None) – 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, default: 10) – Number of consecutive failures before

  • (default –

  • circuit_breaker_timeout (float, default: 30.0) – Seconds to wait before transitioning to

  • (default – 30.0)

  • circuit_breaker_auto_retry (bool, default: False) – 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, default: 3) – Maximum retry attempts when

  • (default –

  • circuit_breaker_retry_delay (float, default: 5.0) – 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, default: 100) – Maximum number of connections in the pool

  • (default –

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

  • (default –

  • session_idle_timeout (Union[int, float, None], default: 300) – 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, default: False) – Enable read-only mode - simulate write operations

  • (default – False)

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

  • (default – False)

  • adaptive_retry (bool, default: False) – 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, default: "exponential") – 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, default: False) – 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], default: None) – 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], default: None) – 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]], default: None) – 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()[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:
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]], default: None) – Request body data (for POST/PUT)

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

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

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

  • response (return full)

  • request_id (Optional[str], default: None) – Optional correlation ID for tracking requests across

  • logs

  • silent (bool, default: False) – 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]], default: None) – Query parameters

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

  • raw_json (bool, default: False) – Return raw JSON response

  • silent (bool, default: False) – 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:

  • api_type (str) – API type

  • path (str) – Endpoint path

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

  • vdom (Union[str, bool, None], default: 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)[source]#

POST request - Create new object

Parameters:

  • scope (Optional[str], default: None) – 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 (Optional[dict[str, Any]], default: None)

  • vdom (Union[str, bool, None], default: None)

  • raw_json (bool, default: False)

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], default: None) – 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 (Optional[dict[str, Any]], default: None)

  • vdom (Union[str, bool, None], default: None)

  • raw_json (bool, default: False)

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], default: None) – Optional scope parameter for global objects (‘global’ or ‘vdom’). Required when deleting objects in global scope.

  • api_type (str)

  • path (str)

  • params (Optional[dict[str, Any]], default: None)

  • vdom (Union[str, bool, None], default: None)

  • raw_json (bool, default: False)

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, default: "mkey") – 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, default: "value") – 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, default: "value") – 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}
__enter__()[source]#

Enter context manager - returns self for use in ‘with’ statements

Return type:

HTTPClient

__exit__(exc_type, exc_val, exc_tb)[source]#

Exit context manager - ensures session is closed

Parameters:
Return type:

None

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_simulated: True if operation was simulated in

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

hfortix_core.http.client.encode_path_component(component)[source]#

Encode a single path component for use in URLs.

This encodes special characters including forward slashes, which are commonly used in FortiOS object names (e.g., IP addresses with CIDR notation).

Parameters:

component (str | int) – Path component to encode (e.g., object name or ID)

Return type:

str

Returns:

URL-encoded string safe for use in URL paths

Examples

>>> encode_path_component("Test_NET_192.0.2.0/24")
'Test_NET_192.0.2.0%2F24'
>>> encode_path_component("test@example.com")
'test%40example.com'
>>> encode_path_component(123)
'123'

Asynchronous HTTP Client#

Internal Async HTTP Client for FortiOS API

This module contains the AsyncHTTPClient class which handles all async HTTP communication with FortiGate devices. It mirrors HTTPClient but uses async/await with httpx.AsyncClient.

This is an internal implementation detail and not part of the public API.

class hfortix_core.http.async_client.AsyncHTTPClient[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, default: True)

  • token (Optional[str], default: None)

  • username (Optional[str], default: None)

  • password (Optional[str], default: None)

  • vdom (Optional[str], default: None)

  • max_retries (int, default: 3)

  • connect_timeout (float, default: 10.0)

  • read_timeout (float, default: 300.0)

  • user_agent (Optional[str], default: None)

  • circuit_breaker_threshold (int, default: 10)

  • circuit_breaker_timeout (float, default: 30.0)

  • circuit_breaker_auto_retry (bool, default: False)

  • circuit_breaker_max_retries (int, default: 3)

  • circuit_breaker_retry_delay (float, default: 5.0)

  • max_connections (int, default: 100)

  • max_keepalive_connections (int, default: 20)

  • session_idle_timeout (Optional[float], default: 300.0)

  • read_only (bool, default: False)

  • track_operations (bool, default: False)

  • adaptive_retry (bool, default: False)

  • retry_strategy (str, default: 'exponential')

  • retry_jitter (bool, default: False)

  • audit_handler (Optional[Any], default: None)

  • audit_callback (Optional[Any], default: None)

  • user_context (Optional[dict[str, Any]], default: None)

Initialize async HTTP client

Parameters:

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

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

  • token (Optional[str], default: None) – API authentication token (if using token auth)

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

  • auth)

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

  • auth)

  • vdom (Optional[str], default: None) – Default virtual domain

  • max_retries (int, default: 3) – Maximum number of retry attempts on transient failures

  • (default (all API calls) –

  • connect_timeout (float, default: 10.0) – Timeout for establishing connection in seconds

  • (default – 10.0)

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

  • 300.0)

  • user_agent (Optional[str], default: None) – Custom User-Agent header

  • circuit_breaker_threshold (int, default: 10) – Number of consecutive failures before

  • (default –

  • circuit_breaker_timeout (float, default: 30.0) – Seconds to wait before transitioning to

  • (default – 30.0)

  • circuit_breaker_auto_retry (bool, default: False) – 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, default: 3) – Maximum number of auto-retry attempts

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

  • circuit_breaker_retry_delay (float, default: 5.0) – 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, default: 100) – Maximum number of connections in the pool

  • (default –

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

  • (default –

  • session_idle_timeout (Optional[float], default: 300.0) – 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, default: False) – Enable read-only mode - simulate write operations

  • (default – False)

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

  • (default – False)

  • adaptive_retry (bool, default: False) – 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, default: 'exponential') – Retry backoff strategy - ‘exponential’ (default) or ‘linear’. Exponential: 1s, 2s, 4s, 8s, 16s, 30s. Linear: 1s, 2s, 3s, 4s, 5s.

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

  • audit_handler (Optional[Any], default: None) – 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], default: None) – Custom callback function for audit logging. Alternative to audit_handler. Receives operation dict as parameter.

  • user_context (Optional[dict[str, Any]], default: None) – 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 –

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

Initialize async HTTP client

Parameters:

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

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

  • token (Optional[str], default: None) – API authentication token (if using token auth)

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

  • auth)

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

  • auth)

  • vdom (Optional[str], default: None) – Default virtual domain

  • max_retries (int, default: 3) – Maximum number of retry attempts on transient failures

  • (default (all API calls) –

  • connect_timeout (float, default: 10.0) – Timeout for establishing connection in seconds

  • (default – 10.0)

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

  • 300.0)

  • user_agent (Optional[str], default: None) – Custom User-Agent header

  • circuit_breaker_threshold (int, default: 10) – Number of consecutive failures before

  • (default –

  • circuit_breaker_timeout (float, default: 30.0) – Seconds to wait before transitioning to

  • (default – 30.0)

  • circuit_breaker_auto_retry (bool, default: False) – 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, default: 3) – Maximum number of auto-retry attempts

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

  • circuit_breaker_retry_delay (float, default: 5.0) – 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, default: 100) – Maximum number of connections in the pool

  • (default –

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

  • (default –

  • session_idle_timeout (Optional[float], default: 300.0) – 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, default: False) – Enable read-only mode - simulate write operations

  • (default – False)

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

  • (default – False)

  • adaptive_retry (bool, default: False) – 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, default: "exponential") – Retry backoff strategy - ‘exponential’ (default) or ‘linear’. Exponential: 1s, 2s, 4s, 8s, 16s, 30s. Linear: 1s, 2s, 3s, 4s, 5s.

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

  • audit_handler (Optional[Any], default: None) – 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], default: None) – Custom callback function for audit logging. Alternative to audit_handler. Receives operation dict as parameter.

  • user_context (Optional[dict[str, Any]], default: None) – 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()[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:
Return type:

None

Example

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

None

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]], default: None) – Request body data (for POST/PUT)

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

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

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

  • response (full)

  • request_id (Optional[str], default: None) – 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

Parameters:
Return type:

dict[str, Any]

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

Async GET request returning binary data

Parameters:
Return type:

bytes

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

Async POST request - Create new object

Parameters:
Return type:

dict[str, Any]

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

Async PUT request - Update existing object

Parameters:
Return type:

dict[str, Any]

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

Async DELETE request - Delete object

Parameters:
Return type:

dict[str, Any]

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

Validate and convert mkey to string

Parameters:

  • mkey (Any)

  • parameter_name (str, default: "mkey")

Return type:

str

static validate_required_params(params, required)[source]#

Validate that required parameters are present

Parameters:
Return type:

None

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

Validate that a numeric value is within a specified range

Parameters:
Return type:

None

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

Validate that a value is one of the allowed choices

Parameters:

  • value (Any)

  • choices (list[Any])

  • parameter_name (str, default: "value")

Return type:

None

static build_params(**kwargs)[source]#

Build parameters dict, filtering out None values

Parameters:

kwargs (Any)

Return type:

dict[str, Any]

async __aenter__()[source]#

Async context manager entry - auto-login if using username/password

Return type:

AsyncHTTPClient

async __aexit__(exc_type, exc_val, exc_tb)[source]#

Async context manager exit - ensures session is closed

This method is called automatically when exiting an async with block. It ensures that all network connections and sessions are properly closed.

Example

async with AsyncHTTPClient(…) as client:

…

# Session is closed automatically here

Parameters:
Return type:

None

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)

Base HTTP Classes#

Base HTTP Client - Shared Logic for Sync and Async Clients

This module contains BaseHTTPClient with shared validation, retry logic, circuit breaker, statistics, and utilities used by both HTTPClient and AsyncHTTPClient.

class hfortix_core.http.base.BaseHTTPClient[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, default: True)

  • vdom (Optional[str], default: None)

  • max_retries (int, default: 3)

  • connect_timeout (float, default: 10.0)

  • read_timeout (float, default: 300.0)

  • circuit_breaker_threshold (int, default: 5)

  • circuit_breaker_timeout (float, default: 60.0)

  • max_connections (int, default: 100)

  • max_keepalive_connections (int, default: 20)

  • adaptive_retry (bool, default: False)

  • retry_strategy (str, default: 'exponential')

  • retry_jitter (bool, default: False)

Initialize base HTTP client with shared configuration

Parameters:

  • adaptive_retry (bool, default: False) – 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, default: 'exponential') – Retry backoff strategy - ‘exponential’ (default) or ‘linear’. Exponential: 1s, 2s, 4s, 8s, 16s, 30s. Linear: 1s, 2s, 3s, 4s, 5s.

  • retry_jitter (bool, default: False) – 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, default: True)

  • vdom (Optional[str], default: None)

  • max_retries (int, default: 3)

  • connect_timeout (float, default: 10.0)

  • read_timeout (float, default: 300.0)

  • circuit_breaker_threshold (int, default: 5)

  • circuit_breaker_timeout (float, default: 60.0)

  • max_connections (int, default: 100)

  • max_keepalive_connections (int, default: 20)

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

Initialize base HTTP client with shared configuration

Parameters:

  • adaptive_retry (bool, default: False) – 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, default: "exponential") – Retry backoff strategy - ‘exponential’ (default) or ‘linear’. Exponential: 1s, 2s, 4s, 8s, 16s, 30s. Linear: 1s, 2s, 3s, 4s, 5s.

  • retry_jitter (bool, default: False) – 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, default: True)

  • vdom (Optional[str], default: None)

  • max_retries (int, default: 3)

  • connect_timeout (float, default: 10.0)

  • read_timeout (float, default: 300.0)

  • circuit_breaker_threshold (int, default: 5)

  • circuit_breaker_timeout (float, default: 60.0)

  • max_connections (int, default: 100)

  • max_keepalive_connections (int, default: 20)

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

Parameters:
Return type:

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.

Usage Examples#

Basic Exception Handling#

from hfortix import FortiOS
from hfortix_core import (
    APIError,
    ResourceNotFoundError,
    DuplicateEntryError,
    ValidationError
)

fgt = FortiOS(host='192.168.1.99', token='your-token')

try:
    result = fgt.api.cmdb.firewall.address.post(
        name='test-server',
        subnet='10.0.0.1/32'
    )
except DuplicateEntryError:
    print("Address already exists!")
except ValidationError as e:
    print(f"Invalid input: {e.message}")
except ResourceNotFoundError as e:
    print(f"Resource not found: {e.message}")
except APIError as e:
    print(f"API error: {e.message} (code: {e.error_code})")

Custom HTTP Client#

Advanced users can extend the base HTTP client:

from hfortix_core.http.client import HTTPClient

class CustomHTTPClient(HTTPClient):
    def __init__(self, *args, custom_header='value', **kwargs):
        super().__init__(*args, **kwargs)
        # Add custom initialization
        self.default_headers['X-Custom-Header'] = custom_header

See Also#

  • /core/api-reference/fortios - FortiOS package that uses this core functionality

  • /core/user-guide/error-handling - Error handling guide

  • /core/user-guide/async-usage - Async usage guide

On this page
Edit on GitHub