FortiOS Client

The main FortiOS client class for connecting to FortiGate devices.

Client Class

class hfortix_fortios.FortiOS(host=None, token=None, *, username=None, password=None, client=None, mode='sync', verify=True, vdom=None, port=None, debug=None, debug_options=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=10, max_keepalive_connections=5, session_idle_timeout=300, read_only=False, track_operations=False, adaptive_retry=False, retry_strategy='exponential', retry_jitter=False, error_mode='raise', error_format='detailed', audit_handler=None, audit_callback=None, user_context=None, trace_id=None)[source]

Bases: object

FortiOS REST API Client

Python client for interacting with Fortinet FortiGate firewalls via REST API. Supports configuration management (CMDB), monitoring, logging, and services.

This client uses token-based authentication and provides a stateless interface to FortiOS devices. No login/logout required - just initialize with your token and start making API calls.

Main API categories:

  • api.cmdb: Configuration Management Database (firewall policies, objects, etc.)

  • api.monitor: Real-time monitoring and status

  • api.log: Log queries and analysis

  • api.service: System services (sniffer, security rating, etc.)

Parameters:

  • host (Optional[str])

  • token (Optional[str])

  • username (Optional[str])

  • password (Optional[str])

  • client (Optional[IHTTPClient])

  • mode (Literal['sync', 'async'])

  • verify (bool)

  • vdom (Optional[str])

  • port (Union[int, str, None])

  • debug (Union[str, bool, None])

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

  • 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 (Literal['exponential', 'linear'])

  • retry_jitter (bool)

  • error_mode (Literal['raise', 'return', 'print'])

  • error_format (Literal['detailed', 'simple', 'code_only'])

  • audit_handler (Optional[AuditHandler])

  • audit_callback (Optional[Any])

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

  • trace_id (Optional[str])

api

API namespace containing cmdb, monitor, log, service

Type:

API

Example:

>>> from hfortix import FortiOS
>>> fgt = FortiOS("fortigate.example.com", token="your_token_here")
>>>
>>> # List firewall addresses
>>> addresses = fgt.api.cmdb.firewall.address.get()
>>>
>>> # Create a firewall address
>>> fgt.api.cmdb.firewall.address.create(
...     name='test-host',
...     subnet='192.0.2.100/32',
...     comment='Example host'
... )
>>>
>>> # Get system status
>>> status = fgt.api.monitor.system.status.get()

Note

  • Requires FortiOS 6.0+ with REST API enabled

  • API token must be created in FortiOS: System > Admin > API Users

  • Use verify=False only in development with self-signed certificates

See also

__init__(host=None, token=None, *, username=None, password=None, client=None, mode='sync', verify=True, vdom=None, port=None, debug=None, debug_options=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=10, max_keepalive_connections=5, session_idle_timeout=300, read_only=False, track_operations=False, adaptive_retry=False, retry_strategy='exponential', retry_jitter=False, error_mode='raise', error_format='detailed', audit_handler=None, audit_callback=None, user_context=None, trace_id=None)[source]

Initialize FortiOS API client (sync or async mode)

Supports two authentication methods: 1. API Token authentication (stateless, recommended for production) 2. Username/Password authentication (session-based, requires login/logout)

Parameters:

  • host (Optional[str]) – FortiGate IP/hostname (e.g., “192.0.2.10” or

  • "fortigate.example.com") – Not required if providing a custom client

  • token (Optional[str]) – API token for authentication (mutually exclusive with

  • username/password) – Not required if providing a custom client or using username/password

  • username (Optional[str]) – Username for password authentication (must be used with password) Mutually exclusive with token

  • password (Optional[str]) – Password for username authentication (must be used with username) Mutually exclusive with token

  • client (Optional[IHTTPClient]) – Optional custom HTTP client implementing IHTTPClient protocol If provided, host/token/verify/etc. are ignored and the custom client is used Allows for custom authentication, proxying, caching, etc.

  • mode (Literal['sync', 'async']) –

    Client mode - ‘sync’ (default) or ‘async’

    • ’sync’: Traditional synchronous API calls

    • ’async’: Asynchronous API calls with async/await

    Ignored if custom client is provided

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

  • vdom (Optional[str]) – Virtual domain (default: None = FortiGate’s default VDOM)

  • port (Union[int, str, None]) – HTTPS port (default: None = use 443, or specify custom port

  • 8443) (like) – Accepts both int and str types - string values are automatically converted to int. This allows passing environment variable values directly: port=os.getenv(“FORTIOS_PORT”, “443”)

  • debug (Union[str, bool, None]) – Logging level for this instance (‘debug’, ‘info’, ‘warning’,

  • 'error' – Can be a string level or boolean True for ‘debug’ level If not specified, uses the global log level set via hfortix.set_log_level()

  • 'off') – Can be a string level or boolean True for ‘debug’ level If not specified, uses the global log level set via hfortix.set_log_level()

  • debug_options (Optional[dict[str, Any]]) – Optional dict with debugging configuration options

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

  • (default (attempts when auto-retry enabled) –

  • 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 (default: ‘hfortix/{version}’) Useful for identifying different applications/teams in FortiGate logs

  • 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). 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 – 10) Conservative default (50% below lowest-performing device tested). Should work for most FortiGate models and network conditions. Most devices serialize API requests internally, so high concurrency doesn’t improve throughput. Increase based on performance testing: 20 for remote-wan, 30 for fast-lan, 60+ for high-performance local deployments.

  • max_keepalive_connections (int) – Maximum number of keepalive connections

  • (default – 5) Conservative default for connection reuse. If max_keepalive_connections exceeds max_connections, it will be automatically adjusted with a warning. Increase proportionally with max_connections based on your device profile.

  • 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 or False to disable proactive re-authentication. Note: The idle timer resets on each API request. Proactive re-auth triggers when time since last request exceeds threshold (not time since login). API token authentication is stateless and doesn’t use sessions. Important: Proactive re-auth only works when using context manager (with statement).

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

  • them (without executing) – (default: False). When enabled, POST/PUT/DELETE requests are logged but not sent to FortiGate. Useful for testing, dry-run, CI/CD pipelines, and training. GET requests are executed normally.

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

  • calls (all API) – (default: False). When enabled, all requests (GET/POST/PUT/DELETE) are recorded with timestamp, method, URL, and data. Access via get_operations() or get_write_operations(). Useful for debugging, auditing, and documentation.

  • 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. Access health metrics via get_health_metrics().

  • retry_strategy (Literal['exponential', 'linear']) – Retry backoff strategy (default: “exponential”). - “exponential”: 1s, 2s, 4s, 8s, 16s, 30s (recommended for transient failures) - “linear”: 1s, 2s, 3s, 4s, 5s (better for rate limiting scenarios)

  • retry_jitter (bool) – Add random jitter to retry delays (default: False). Adds 0-25% random variation to prevent thundering herd when multiple clients retry simultaneously. Recommended for production deployments.

  • error_mode (Literal['raise', 'return', 'print']) – How convenience wrappers handle errors (default:

  • "raise"). –

    • “raise”: Raise exceptions (stops program unless caught with try/except)

    • ”return”: Return error dict instead of raising (program always continues)

    • ”log”: Log error and return None (program always continues)

    Can be overridden per method call.

  • error_format (Literal['detailed', 'simple', 'code_only']) –

    Error message detail level (default: “detailed”).

    • ”detailed”: Full context with endpoint, parameters, and helpful hints

    • ”simple”: Just error message and code

    • ”code_only”: Just the error code number

    Can be overridden per method call. Affects both raised exceptions and returned error dicts depending on error_mode.

  • audit_handler (Optional[AuditHandler]) – Handler for enterprise audit logging (default: None). Automatically logs all API operations for compliance (SOC 2, HIPAA, PCI-DSS). Use built-in handlers: SyslogHandler (SIEM), FileHandler (local logs), StreamHandler (container logs), CompositeHandler (multiple destinations). Example: SyslogHandler(“siem.company.com:514”)

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

  • (default – None). Alternative to audit_handler. Called with operation dict for each API call. Use for custom logging destinations (Kafka, database, cloud services). Example: lambda op: send_to_kafka(op)

  • user_context (Optional[dict[str, Any]]) – Optional user/application context for audit logs

  • (default – None). Dict with metadata to include in every audit entry. Useful for tracking which user/script/ticket caused each change. Example: {“username”: “admin”, “app”: “backup_script”, “ticket”: “CHG-12345”}

  • trace_id (Optional[str]) – Optional distributed tracing ID for request correlation

  • (default – None). String identifier to track requests across multiple systems. Automatically included in user_context and all audit logs. Useful for debugging and distributed tracing systems (Jaeger, Zipkin, etc.). Example: “request-12345” or UUID

Return type:

None

Important

Username/password authentication still works in FortiOS 7.4.x but is removed in FortiOS 7.6.x and later. Use API token authentication for production deployments.

Performance Note:

Most FortiGate devices serialize API requests internally, meaning concurrent requests don’t improve throughput and actually increase response times (10-15x slower). Sequential requests are recommended for most deployments. Use async mode only when integrating with async frameworks or managing multiple devices in parallel. Performance testing shows ~5 req/s for most devices, ~30 req/s for high-performance local deployments. See COMPARATIVE_ANALYSIS.md for detailed performance profiles.

Examples:

# Token authentication (recommended)
fgt = FortiOS("fortigate.example.com", token="your_token_here",
verify=True)
addresses = fgt.api.cmdb.firewall.address.get("test-host")

# Enterprise audit logging to SIEM (compliance)
from hfortix_core.audit import SyslogHandler
fgt = FortiOS("192.0.2.10", token="token",
             audit_handler=SyslogHandler("siem.company.com:514"))
# All API operations now logged to SIEM automatically

# Multi-destination audit logging
from hfortix_core.audit import CompositeHandler, FileHandler,
StreamHandler
handler = CompositeHandler([
    SyslogHandler("siem.company.com:514"),  # Compliance
    FileHandler("/var/log/fortinet-audit.jsonl"),  # Backup
])
fgt = FortiOS("192.0.2.10", token="token", audit_handler=handler)

# Custom audit callback
def my_audit(op):
    send_to_kafka(op)
    update_cmdb(op)
fgt = FortiOS("192.0.2.10", token="token",
audit_callback=my_audit)

# Audit logging with user context
fgt = FortiOS("192.0.2.10", token="token",
             audit_handler=SyslogHandler("siem.company.com:514"),
             user_context={"username": "admin", "ticket":
             "CHG-12345"})

# Distributed tracing with trace_id
fgt = FortiOS("192.0.2.10", token="token",
             trace_id="request-abc123",
             audit_handler=SyslogHandler("siem.company.com:514"))
# trace_id automatically added to all audit logs and user_context

# Username/Password authentication with context manager (sync)
with FortiOS("192.0.2.10", username="admin", password="password",
verify=False) as fgt:
    addresses = fgt.api.cmdb.firewall.address.get("test-host")
    # Auto-logout on exit

# Username/Password authentication with context manager (async)
async with FortiOS("192.0.2.10", username="admin",
password="password",
                  mode="async", verify=False) as fgt:
    status = await fgt.api.monitor.system.status.get()
    # Auto-logout on exit

# Asynchronous mode with token
fgt = FortiOS("fortigate.example.com", token="your_token_here",
mode="async")
addresses = await fgt.api.cmdb.firewall.address.get("test-host")

# Custom HTTP client
class MyHTTPClient:
    def get(self, api_type, path, **kwargs):
        # Custom implementation with company auth, logging, etc.
        ...
    def post(self, api_type, path, data, **kwargs):
        ...
    # ... put, delete

fgt = FortiOS(client=MyHTTPClient())
addresses = fgt.api.cmdb.firewall.address.get("test-host")

# Production - with valid SSL certificate
fgt = FortiOS("fortigate.example.com", token="your_token_here",
verify=True)

# Development/Testing - with self-signed certificate (example IP from RFC 5737)  # noqa: E501
fgt = FortiOS("192.0.2.10", token="your_token_here", verify=False)

# Environment variables (credentials from environment)
# Set: export FORTIOS_HOST="192.0.2.10"
#      export FORTIOS_TOKEN="your_token_here"
fgt = FortiOS()  # Reads from FORTIOS_HOST, FORTIOS_TOKEN

# Environment variables with username/password
# Set: export FORTIOS_HOST="192.0.2.10"
#      export FORTIOS_USERNAME="admin"
#      export FORTIOS_PASSWORD="your_password"
fgt = FortiOS()  # Reads from FORTIOS_HOST, FORTIOS_USERNAME,
FORTIOS_PASSWORD

# Environment variables with custom port
# Set: export FORTIOS_HOST="192.0.2.10"
#      export FORTIOS_TOKEN="your_token_here"
#      export FORTIOS_PORT="8443"
fgt = FortiOS()  # Reads from FORTIOS_HOST, FORTIOS_TOKEN, FORTIOS_PORT

# Custom port
fgt = FortiOS("192.0.2.10", token="your_token_here", verify=False,
port=8443)

# Port in hostname (alternative)
fgt = FortiOS("192.0.2.10:8443", token="your_token_here",
verify=False)

# Enable debug logging for this instance only
fgt = FortiOS("192.0.2.10", token="your_token_here", verify=False,
debug='info')

# Custom timeouts (e.g., slower network)
fgt = FortiOS("192.0.2.10", token="your_token_here",
connect_timeout=30.0, read_timeout=600.0)

# Custom User-Agent for multi-team environments
fgt = FortiOS("192.0.2.10", token="your_token_here",
user_agent="BackupScript/2.1.0")

# Read-only mode for testing (simulates writes without executing)
fgt = FortiOS("192.0.2.10", token="your_token_here",
read_only=True)
fgt.api.cmdb.firewall.address.create(name="test")  # Logged but not
executed

# Operation tracking for debugging/auditing
fgt = FortiOS("192.0.2.10", token="your_token_here",
track_operations=True)
fgt.api.cmdb.firewall.address.create(name="test",
subnet="10.0.0.1/32")
operations = fgt.get_operations()  # Get all operations
write_ops = fgt.get_write_operations()  # Only POST/PUT/DELETE
property api: API

Primary entry point to FortiOS endpoints (cmdb/monitor/log/service).

__dir__()[source]

Prefer showing api early in interactive completion.

Return type:

list[str]

property host: str | None

FortiGate hostname or IP address

property port: int | None

HTTPS port number

property vdom: str | None

Active virtual domain

property error_mode: Literal['raise', 'return', 'print']

Default error handling mode for convenience wrappers

property error_format: Literal['detailed', 'simple', 'code_only']

Default error message format for convenience wrappers

request(config)[source]

Execute a generic API request from FortiGate GUI API preview JSON

This method accepts the JSON configuration directly from the FortiGate GUI’s API preview feature, making it easy to test and execute API calls without manually constructing requests.

Parameters:

config (dict[str, Any]) –

Dictionary containing the API request configuration with:

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

  • url: Full API URL path (e.g., “/api/v2/cmdb/firewall/address”)

  • params: Optional query parameters dict

  • data: Optional request body for POST/PUT

Return type:

Any

Returns:

Full API response dictionary with http_status, results, etc.

Raises:

  • ValueError – If config is missing required fields or has invalid format

  • APIError – For API errors (404, 500, etc.)

Example

>>> fgt = FortiOS("192.168.1.99", token="...")
>>>
>>> # Copy this directly from FortiGate GUI API preview
>>> config = {
...     "method": "POST",
...     "url": "/api/v2/cmdb/firewall/address",
...     "params": {
...         "datasource": 1,
...         "vdom": "test"
...     },
...     "data": {
...         "name": "test999999",
...         "subnet": "192.168.1.0/24",
...         "color": "0"
...     }
... }
>>> result = fgt.request(config)
>>>
>>> # Example: GET request
>>> get_config = {
...     "method": "GET",
...     "url": "/api/v2/cmdb/firewall/address",
...     "params": {"vdom": "root"}
... }
>>> addresses = fgt.request(get_config)
>>>
>>> # Example: PUT request
>>> update_config = {
...     "method": "PUT",
...     "url": "/api/v2/cmdb/firewall/address/test999999",
...     "params": {"vdom": "test"},
...     "data": {"comment": "Updated via API"}
... }
>>> result = fgt.request(update_config)
>>>
>>> # Example: DELETE request
>>> delete_config = {
...     "method": "DELETE",
...     "url": "/api/v2/cmdb/firewall/address/test999999",
...     "params": {"vdom": "test"}
... }
>>> result = fgt.request(delete_config)

Note

  • The URL should include /api/v2/ prefix (as shown in GUI)

  • The vdom parameter can be in params dict or will use default

  • This method is perfect for testing API calls from the GUI before implementing in code

get_connection_stats()[source]

Get HTTP connection pool statistics and metrics

Provides insights into connection health, retry behavior, and circuit breaker state. Useful for monitoring, debugging, and capacity planning.

Returns:

  • total_requests: Total number of API requests made

  • successful_requests: Number of successful requests

  • failed_requests: Number of failed requests

  • total_retries: Total number of retry attempts

  • success_rate: Percentage of successful requests

  • retry_by_reason: Breakdown of retries by failure reason

  • retry_by_endpoint: Breakdown of retries by endpoint

  • circuit_breaker_state: Current circuit breaker state (closed/open/half_open)

  • circuit_breaker_failures: Consecutive failure count

  • last_retry_time: Timestamp of last retry (if any)

Return type:

Dictionary containing connection statistics

Example:

>>> fgt = FortiOS("192.0.2.10", token="...")
>>> stats = fgt.get_connection_stats()
>>> print(f"Success rate: {stats['success_rate']:.1f}%")
>>> print(f"Total retries: {stats['total_retries']}")
>>> print(f"Circuit breaker: {stats['circuit_breaker_state']}")
>>> if stats['retry_by_reason']:
...     print("Retry reasons:")
...     for reason, count in stats['retry_by_reason'].items():
...         print(f"  {reason}: {count}")

Note

Statistics are collected from the time the FortiOS instance was created. Use this method to monitor connection health and identify issues.

get_operations()[source]

Get audit log of all API operations (requires track_operations=True)

Returns list of all operations (GET/POST/PUT/DELETE) with details about each request. Only available when track_operations=True was passed to FortiOS constructor.

Returns:

  • timestamp: ISO 8601 timestamp when operation was executed

  • 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 for GET/DELETE

  • status_code: HTTP response status code

  • vdom: Virtual domain (if specified)

Return type:

List of operation dictionaries containing

Raises:

RuntimeError – If track_operations was not enabled

Example:

>>> fgt = FortiOS("192.0.2.10", token="...", track_operations=True)
>>> fgt.api.cmdb.firewall.address.create(name="test",
...                                       subnet="10.0.0.1/32")
>>> fgt.api.cmdb.firewall.policy.update("10", action="deny")
>>>
>>> operations = fgt.get_operations()
>>> for op in operations:
...     print(f"{op['timestamp']} {op['method']} {op['path']}")
2024-12-20T10:30:15Z POST /api/v2/cmdb/firewall/address
2024-12-20T10:30:16Z PUT /api/v2/cmdb/firewall/policy/10

Note

Use get_write_operations() to filter only write operations (POST/PUT/DELETE).

get_write_operations()[source]

Get audit log of write operations only (requires track_operations=True)

Returns list of only write operations (POST/PUT/DELETE), excluding GET requests. Only available when track_operations=True was passed to FortiOS constructor.

Return type:

list[dict[str, Any]]

Returns:

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

Raises:

RuntimeError – If track_operations was not enabled

Example

>>> fgt = FortiOS("192.0.2.10", token="...", track_operations=True)
>>> fgt.api.cmdb.firewall.address.get("test")  # GET - not included
>>> fgt.api.cmdb.firewall.address.create(name="test2",
subnet="10.0.0.2/32")  # POST
>>> fgt.api.cmdb.firewall.address.delete("test")  # DELETE
>>>
>>> write_ops = fgt.get_write_operations()
>>> # Only POST and DELETE are returned, GET is excluded
>>> for op in write_ops:
...     print(f"{op['method']} {op['path']} - {op['data']}")
POST /api/v2/cmdb/firewall/address - {'name': 'test2', 'subnet':
'10.0.0.2/32'}
DELETE /api/v2/cmdb/firewall/address/test - None

Note

Useful for generating change logs, rollback scripts, and audit reports.

export_audit_logs(filepath=None, format='json', filter_method=None, filter_api_type=None, since=None)[source]

Export audit logs to file or return as string

Exports all tracked operations (requires track_operations=True) to a file or returns as formatted string. Useful for compliance reporting, change documentation, and integration with external SIEM systems.

Parameters:

  • filepath (Optional[str]) – Path to export file (optional). If None, returns string

  • format (str) – Export format - “json” (default), “csv”, or “text”

  • filter_method (Optional[str]) – Filter by HTTP method (e.g., “POST”, “PUT”,

  • "DELETE")

  • filter_api_type (Optional[str]) – Filter by API type (e.g., “cmdb”, “monitor”)

  • since (Optional[str]) – Filter operations since ISO 8601 timestamp (e.g., “2025-01-01T00:00:00Z”)

Return type:

Optional[str]

Returns:

Formatted string if filepath is None, otherwise None (writes to file)

Raises:

Example

>>> fgt = FortiOS("192.0.2.10", token="...", track_operations=True)
>>> # Make some changes
>>> fgt.api.cmdb.firewall.address.create(name="test",
...                                       subnet="10.0.0.1/32")
>>> fgt.api.cmdb.firewall.policy.update("10", action="deny")
>>>
>>> # Export to JSON file
>>> fgt.export_audit_logs("audit.json", format="json")
>>>
>>> # Export only write operations to CSV
>>> fgt.export_audit_logs("changes.csv", format="csv",
...                       filter_method="POST,PUT,DELETE")
>>>
>>> # Get as string for processing
>>> audit_json = fgt.export_audit_logs(format="json")
>>> send_to_siem(audit_json)

Note

For real-time audit logging to SIEM, use audit_handler parameter when initializing FortiOS client instead.

get_retry_stats()[source]

Get retry statistics from HTTP client

Returns statistics about retry attempts, including total retries, reasons for retries, and per-endpoint retry counts. Useful for monitoring FortiGate health and diagnosing connectivity issues.

Returns:

  • total_retries: Total number of retry attempts across all requests

  • total_requests: Total number of requests made

  • successful_requests: Number of requests that succeeded

  • failed_requests: Number of requests that ultimately failed

  • retry_by_reason: Dict mapping retry reason to count (e.g., {“timeout”: 10, “rate_limit”: 8, “server_error”: 5})

  • retry_by_endpoint: Dict mapping endpoint to retry count

  • last_retry_time: Unix timestamp of most recent retry

Return type:

Dictionary containing

Example

>>> fgt = FortiOS("192.0.2.10", token="...", max_retries=5)
>>> # Make some requests that might retry
>>> fgt.api.cmdb.firewall.policy.get()
>>>
>>> stats = fgt.get_retry_stats()
>>> print(f"Total retries: {stats['total_retries']}")
>>> print(f"Success rate: {stats['successful_requests'] / stats['total_requests'] * 100:.1f}%")  # noqa: E501
>>> for reason, count in stats['retry_by_reason'].items():
...     print(f"  {reason}: {count} retries")
Total retries: 23
Success rate: 98.5%
  timeout: 10 retries
  rate_limit: 8 retries
  server_error: 5 retries

Note

Stats are cumulative for the lifetime of the FortiOS client instance. # noqa: E501

get_circuit_breaker_state()[source]

Get current circuit breaker state

Returns the current state of the circuit breaker, including whether it’s open, closed, or half-open, the number of consecutive failures, and the configured threshold.

Returns:

  • state: Current state - “closed”, “open”, or “half_open”

  • consecutive_failures: Number of consecutive failures

  • failure_threshold: Threshold for opening circuit

  • timeout: Seconds to wait before transitioning to half-open

  • last_failure_time: Unix timestamp of most recent failure

Return type:

Dictionary containing

Example

>>> fgt = FortiOS("192.0.2.10", token="...",
...               circuit_breaker_threshold=10)
>>> # Make requests
>>> try:
...     fgt.api.cmdb.firewall.policy.get()
... except CircuitBreakerOpenError:
...     state = fgt.get_circuit_breaker_state()
...     print(f"Circuit is {state['state']}")
...     print(f"Failures: {state['consecutive_failures']}/{state['failure_threshold']}")  # noqa: E501
Circuit is open
Failures: 10/10

Note

Circuit breaker automatically resets after successful requests. You can manually reset with fgt._client.reset_circuit_breaker().

get_health_metrics()[source]

Get comprehensive health metrics for HTTP client

Returns health metrics including: - Circuit breaker state and failures - Retry statistics by endpoint and reason - Response time metrics (if adaptive_retry=True) - Endpoint health status (slow vs normal)

Returns:

  • circuit_breaker: Circuit breaker state, consecutive failures, threshold

  • retry_stats: Total retries, requests, success/failure counts

  • adaptive_retry_enabled: Whether adaptive retry is active

  • response_times: Per-endpoint metrics (avg, min, max, p50, p95) if adaptive_retry=True

Return type:

Dictionary containing

Example:

>>> fgt = FortiOS("192.0.2.10", token="...", adaptive_retry=True)
>>> # Make some requests
>>> fgt.api.cmdb.firewall.address.get()
>>>
>>> # Check health
>>> metrics = fgt.get_health_metrics()
>>> print(f"Circuit state: {metrics['circuit_breaker']['state']}")
>>> print(f"Total retries:
...       {metrics['retry_stats']['total_retries']}")
>>>
>>> # Check response times (if adaptive_retry=True)
>>> if metrics['response_times']:
...     for endpoint, stats in metrics['response_times'].items():
...         print(f"{endpoint}: avg={stats['avg_ms']}ms,
...               slow={stats['is_slow']}")
Circuit state: closed
Total retries: 2
cmdb/firewall/address: avg=245.5ms, slow=False

Note

Response time metrics only available when adaptive_retry=True

transaction(timeout=60, vdom=None, auto_commit=True, auto_abort=True)[source]

Create a FortiOS batch transaction context manager

Batches multiple API calls into an atomic transaction. All changes are applied together on commit, or rolled back on abort. Ideal for multi-step configuration changes that must succeed or fail as a unit.

Parameters:

  • timeout (int) – Transaction timeout in seconds (default: 60)

  • vdom (Optional[str]) – Virtual Domain for the transaction (default: use client’s vdom)

  • auto_commit (bool) – Auto-commit on successful context exit (default: True)

  • auto_abort (bool) – Auto-abort on exception in context (default: True)

Returns:

Transaction context manager

Return type:

Transaction

Raises:

  • RuntimeError – If another transaction is already active

  • TransactionError – If transaction start fails

Example:

>>> # Context manager - automatic commit/abort
>>> with fgt.transaction() as txn:
...     fgt.api.cmdb.system.interface.post({
...         "name": "port3",
...         "vdom": "root",
...         "mode": "static",
...         "ip": "192.168.1.1 255.255.255.0"
...     })
...     fgt.api.cmdb.firewall.policy.post({
...         "policyid": 1,
...         "name": "test-policy",
...         "srcintf": [{"name": "port1"}],
...         "dstintf": [{"name": "port2"}],
...         "srcaddr": [{"name": "all"}],
...         "dstaddr": [{"name": "all"}],
...         "action": "accept",
...         "schedule": "always",
...         "service": [{"name": "ALL"}]
...     })
>>> # Transaction auto-commits on successful exit
>>>
>>> # Manual control - disable auto_commit
>>> with fgt.transaction(auto_commit=False) as txn:
...     fgt.api.cmdb.system.interface.post(...)
...     # Do validation or checks
...     if everything_ok:
...         txn.commit()
...     else:
...         txn.abort()
>>>
>>> # Longer timeout for complex operations
>>> with fgt.transaction(timeout=300) as txn:
...     for interface in interfaces:
...         fgt.api.cmdb.system.interface.post(interface)

Note

  • Requires FortiOS 6.4.0 or later

  • Only one transaction can be active at a time

  • Transaction automatically aborts on exception

  • All API calls within context use same transaction ID

transactional(timeout=60, vdom=None)[source]

Decorator to run a function within a FortiOS transaction

Wraps a function so all API calls it makes are batched into an atomic transaction. Auto-commits on success, auto-aborts on exception.

Parameters:

  • timeout (int) – Transaction timeout in seconds (default: 60)

  • vdom (Optional[str]) – Virtual Domain for the transaction (default: use client’s vdom)

Returns:

Decorator function

Return type:

Callable

Raises:

  • RuntimeError – If another transaction is already active

  • TransactionError – If transaction start fails

Example:

>>> @fgt.transactional(timeout=120)
... def setup_network_infrastructure():
...     # All these calls happen in one transaction
...     fgt.api.cmdb.system.interface.post({
...         "name": "dmz",
...         "vdom": "root",
...         "mode": "static",
...         "ip": "10.0.0.1 255.255.255.0"
...     })
...     fgt.api.cmdb.firewall.address.post({
...         "name": "dmz-server",
...         "subnet": "10.0.0.10 255.255.255.255"
...     })
...     fgt.api.cmdb.firewall.policy.post({
...         "policyid": 100,
...         "name": "allow-dmz",
...         "srcintf": [{"name": "internal"}],
...         "dstintf": [{"name": "dmz"}],
...         "srcaddr": [{"name": "all"}],
...         "dstaddr": [{"name": "dmz-server"}],
...         "action": "accept",
...         "schedule": "always",
...         "service": [{"name": "HTTP"}]
...     })
...     return {"status": "success"}
>>>
>>> # Function executes within transaction
>>> result = setup_network_infrastructure()
>>> # Transaction auto-commits if no exception raised
>>>
>>> @fgt.transactional()
... def configure_interfaces(interfaces: list[dict]):
...     for intf in interfaces:
...         fgt.api.cmdb.system.interface.post(intf)
>>>
>>> # If any interface fails, all changes are rolled back
>>> configure_interfaces([
...     {"name": "port3", "ip": "192.168.1.1 255.255.255.0"},
...     {"name": "port4", "ip": "192.168.2.1 255.255.255.0"},
... ])

Note

  • Requires FortiOS 6.4.0 or later

  • Always auto-commits on success, auto-aborts on exception

  • Cannot be nested with other transactions

  • Function can access transaction via fgt._active_transaction

list_transactions()[source]

List all active transactions on the FortiGate

Shows transaction IDs, VDOMs, creation times, and other metadata for all currently active transactions across the device.

Returns:

List of active transaction details

Return type:

list[dict]

Raises:

ValueError – If FortiOS version doesn’t support listing (< 7.4.1)

Example:

>>> # Show all active transactions
>>> transactions = fgt.list_transactions()
>>> for txn in transactions:
...     print(f"ID: {txn['transaction_id']}, VDOM: {txn['vdom']}")
ID: 12345, VDOM: root
ID: 12346, VDOM: customer1
>>>
>>> # Check if specific transaction exists
>>> active_ids = [t['transaction_id'] for t in fgt.list_transactions()]
>>> if 12345 in active_ids:
...     print("Transaction 12345 still active")

Note

  • Requires FortiOS 7.4.1 or later

  • Returns transactions from all VDOMs (admin must have access)

  • Includes transactions started by other users/sessions

close()[source]

Close the HTTP session and release resources

Optional: Python automatically cleans up when object is destroyed. Use this for explicit resource management or in long-running apps.

Note

For async mode, use await fgt.aclose() instead.

Return type:

None

async aclose()[source]

Close the async HTTP session and release resources (async mode only)

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

Usage:

  • Call await fgt.aclose() when you are done with the client in async mode.

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

Example

>>> fgt = FortiOS("192.0.2.10", token="...", mode="async")
>>> try:
...     addresses = await fgt.api.cmdb.firewall.address.get()
... finally:
...     await fgt.aclose()

Note

Prefer using ‘async with’ statement for automatic cleanup: >>> async with FortiOS(“192.0.2.10”, token=”…”, mode=”async”) as fgt: … addresses = await fgt.api.cmdb.firewall.address.get()

Return type:

None

property connection_stats: dict[str, Any]

Get connection pool and health statistics

Convenience property that returns real-time connection pool metrics, circuit breaker state, and request statistics.

Returns:

  • http2_enabled: Whether HTTP/2 is enabled

  • max_connections: Maximum connections allowed

  • max_keepalive_connections: Maximum keepalive connections

  • active_requests: Current active requests

  • total_requests: Total requests since initialization

  • pool_exhaustion_count: Times pool reached capacity

  • circuit_breaker_state: Current state (closed/open/half-open)

  • consecutive_failures: Consecutive failure count

  • last_failure_time: Timestamp of last failure

Return type:

Dictionary with connection metrics

Example

>>> fgt = FortiOS("192.168.1.99", token="...")
>>> stats = fgt.connection_stats
>>> print(f"Active: {stats['active_requests']}/{stats['max_connections']}")  # noqa: E501
>>> print(f"Pool exhaustions: {stats['pool_exhaustion_count']}")
>>> print(f"Circuit breaker: {stats['circuit_breaker_state']}")
property last_request: dict[str, Any]

Get details of last API request (for debugging)

Returns information about the most recent API call including method, endpoint, response time, and status code. Useful for troubleshooting and performance analysis.

Returns:

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

  • endpoint: API endpoint path

  • params: Query parameters used

  • response_time_ms: Response time in milliseconds

  • status_code: HTTP status code

  • error: Error message if no requests made yet

Return type:

Dictionary with request details

Example

>>> fgt = FortiOS("192.168.1.99", token="...")
>>> fgt.api.cmdb.firewall.address.get()
>>> info = fgt.last_request
>>> print(f"Last request: {info['method']} {info['endpoint']}")
>>> print(f"Response time: {info['response_time_ms']:.2f}ms")
__enter__()[source]

Context manager entry (sync mode only)

Return type:

FortiOS

__exit__(exc_type, exc_val, exc_tb)[source]

Context manager exit - automatically closes session (sync mode only)

Return type:

None

Parameters:
async __aenter__()[source]

Async context manager entry (async mode only)

Enters the async context for FortiOS. Use with async with to ensure proper resource management.

Returns:

The async client instance.

Return type:

FortiOS

Example

>>> async with FortiOS("192.0.2.10", token="...", mode="async") as
fgt:
...     addresses = await fgt.api.cmdb.firewall.address.get()
async __aexit__(exc_type, exc_val, exc_tb)[source]

Async context manager exit - automatically closes session (async mode only)

Ensures that all resources are cleaned up when exiting the async context.

Returns:

False (exceptions are not suppressed)

Return type:

bool

Parameters:

Usage

Basic Connection

from hfortix_fortios import FortiOS

# Connect with API token (recommended)
fgt = FortiOS(
    host='192.168.1.99',
    token='your-api-token',
    verify=False  # Use True in production
)

# Connect with username/password (FortiOS ≤7.4 only)
fgt = FortiOS(
    host='192.168.1.99',
    username='admin',
    password='password',
    verify=False
)

Optimized Settings

# For high-performance scenarios
fgt = FortiOS(
    host='192.168.1.99',
    token='your-token',
    max_connections=60,
    max_keepalive_connections=30,
    connect_timeout=10.0,
    read_timeout=300.0
)

API Namespaces

The FortiOS client provides access to all API categories through organized namespaces.

API Namespace

class hfortix_fortios.api.API(client)[source]

Bases: object

FortiOS REST API v2 Interface.

Organizes all FortiOS API endpoints into logical categories for easy access.

Parameters:

client (IHTTPClient)

cmdb

Configuration Management Database (CRUD operations on config) - Create objects with POST - Update objects with PUT - Read objects with GET - Delete objects with DELETE - Example: fgt.api.cmdb.firewall.address.post(…)

monitor

Real-time monitoring and status (read-only, GET operations) - System resources, interfaces, sessions - Firewall statistics and connections - Routing tables, VPN status - Example: fgt.api.monitor.system.resource.get()

log

Historical log retrieval (read-only, GET operations) - Traffic logs, event logs, security logs - Disk, memory, FortiAnalyzer logs - Example: fgt.api.log.disk.traffic.get()

service

System services and operations - Special service endpoints - Example: fgt.api.service.system.get()

utils

Utility functions for testing and diagnostics

  • Performance testing and benchmarking

  • Connection pool validation

  • Device profiling and recommendations

  • Example: fgt.api.utils.performance_test()

HTTP Method Guidelines:

  • POST: Create new configuration objects (returns 404 if already exists)

  • PUT: Update existing configuration objects (returns 404 if not found)

  • GET: Retrieve data (config, monitoring, logs) - read-only

  • DELETE: Remove configuration objects (returns 404 if not found)

Example

>>> from hfortix_fortios import FortiOS
>>> fgt = FortiOS(host="192.168.1.99", token="your-token")
>>>
>>> # CMDB: Manage configuration
>>> fgt.api.cmdb.firewall.address.post(name="WebServer",
subnet="10.0.0.5/32")
>>> fgt.api.cmdb.firewall.policy.get(policyid=1)
>>>
>>> # Monitor: Real-time data
>>> fgt.api.monitor.system.interface.get()
>>> fgt.api.monitor.firewall.session.get()
>>>
>>> # Log: Historical data
>>> fgt.api.log.disk.traffic.get(count=100)
__init__(client)[source]

Initialize API namespace with HTTP client implementing IHTTPClient protocol.

Note

Utils requires concrete HTTPClient for internal access. When a protocol-only client is provided, Utils will be unavailable (set to None).

Parameters:

client (IHTTPClient)

Return type:

None

cmdb: CMDB
log: Log
monitor: Monitor
service: Service
utils: Utils | None
__dir__()[source]

Control autocomplete to show only public attributes

Return type:

list[str]

The api namespace provides access to:

  • api.cmdb - Configuration Management Database (886 endpoints)

  • api.monitor - Monitoring and status (295 endpoints)

  • api.log - Log queries (38 endpoints with full parameterization)

Examples

Using CMDB API

# List firewall addresses
addresses = fgt.api.cmdb.firewall.address.get()

# Create firewall address
fgt.api.cmdb.firewall.address.post(
    name='web-server',
    subnet='10.0.1.100/32'
)

# Update firewall address
fgt.api.cmdb.firewall.address.put(
    name='web-server',
    comment='Production server'
)

# Delete firewall address
fgt.api.cmdb.firewall.address.delete(name='web-server')

Using Monitor API

# Get system status
status = fgt.api.monitor.system.status.get()

# Get firewall sessions
sessions = fgt.api.monitor.firewall.session.get()

# Get interface statistics
stats = fgt.api.monitor.system.interface.get()

Using Direct API Methods

# Create firewall policy (using post)
policy = fgt.api.cmdb.firewall.policy.post(
    name='Allow-Web',
    srcintf=[{"name": "internal"}],
    dstintf=[{"name": "wan1"}],
    srcaddr=[{"name": "all"}],
    dstaddr=[{"name": "web-server"}],
    service=[{"name": "HTTP"}, {"name": "HTTPS"}],
    action='accept'
)

# Check if policy exists
if fgt.api.cmdb.firewall.policy.exists(policyid='1'):
    print("Policy exists!")

See Also

  • /fortios/api-reference/cmdb/index - CMDB API reference

  • /fortios/api-reference/monitor/index - Monitor API reference

  • /fortios/user-guide/fortios-overview - User guide

  • /fortios/getting-started/quickstart - Quick start guide