hfortix-core 0.5.162

Core foundation for HFortix - Fully-typed Fortinet SDK with async support and type safety

HFortix Core

Core HTTP client and shared utilities for the HFortix ecosystem

📦 Overview

hfortix-core is the foundational package for all HFortix Fortinet automation tools. It provides:

• 🔐 CloudSession - Multi-service OAuth token manager with advanced features:
  • Pluggable token storage (Redis, PostgreSQL, MySQL, SQLite, File, Memcached)
  • Lifecycle hooks for monitoring and metrics
  • Background token refresh
  • Distributed token sharing across servers
• � Rate Limiting Enforcement (NEW) - Token bucket algorithm with queue support:
  • Three strategies: queue, drop, raise
  • FIFO queue for overflow handling
  • Comprehensive metrics tracking
  • Zero overhead when disabled (default)
• 🔌 Circuit Breaker (Enhanced) - Now configurable, default disabled:
  • State machine: closed → open → half-open → closed
  • Test calls required before full recovery
  • Zero overhead when disabled
• 📊 Rate Limiting Tracking - Monitor API call rates across multiple time windows
• 🌐 HTTP Client Framework - Async and sync HTTP client with retry logic
• ⚠️ Exception System - Comprehensive error handling for Fortinet APIs
• 🛠️ Utilities - Common functions used across all Fortinet product packages
• 📘 Type Definitions - Shared TypedDict and Protocol definitions

🚀 Installation

pip install hfortix-core

Note: This package is usually installed automatically as a dependency of other HFortix packages like hfortix-fortios, hfortix-forticare, hfortix-fortiztp.

📚 Usage

CloudSession - Multi-Service OAuth Management

CloudSession provides centralized OAuth token management for multiple FortiCloud services:

from hfortix_core.session import CloudSession
from hfortix_forticare import FortiCare
from hfortix_fortiztp import FortiZTP

# Simple usage - auto-detects client_id for each service
with CloudSession(api_id="your_api_id", password="your_password") as session:
    fc = FortiCare(session=session)    # Uses "assetmanagement" client_id
    fz = FortiZTP(session=session)     # Uses "fortiztp" client_id
    
    # Both services share the session, each with their own token
    products = fc.api.products.list.post()
    devices = fz.devices.get()

Key Features:

• Auto-detect client_id: Services use their DEFAULT_CLIENT_ID automatically
• Token caching: Multiple clients with same client_id share tokens efficiently
• Background refresh: Optional auto-refresh thread to renew tokens before expiration
• Pluggable storage: Redis, PostgreSQL, MySQL, SQLite, File, Memcached backends
• Lifecycle hooks: Monitor token operations with callbacks for metrics/logging
• Distributed tokens: Share tokens across multiple servers via Redis/Database
• Thread-safe: Concurrent access from multiple services
• Context manager: Auto-cleanup with with statement

Advanced Usage:

# Override client_id for specialized access
with CloudSession(api_id="...", password="...") as session:
    fc_standard = FortiCare(session=session)                      # "assetmanagement"
    fc_elite = FortiCare(session=session, client_id="fcelite")   # "fcelite" 
    # Each gets its own token

# Enable auto-refresh
session = CloudSession(
    api_id="...",
    password="...",
    auto_refresh=True,           # Enable background refresh
    refresh_buffer_seconds=300   # Refresh 5 min before expiry
)
fc = FortiCare(session=session)
# Tokens refresh automatically in background

# Manual token management
session = CloudSession(api_id="...", password="...")
fc_token = session.get_token("assetmanagement")  # Get token for specific client_id
token_info = session.get_token_info("assetmanagement")  # Check expiration
print(f"Token expires in {token_info['time_remaining']}s")

Enterprise Features - Distributed Token Storage:

Share tokens across multiple servers using Redis, PostgreSQL, MySQL, SQLite, or file-based storage:

# Redis (recommended for production multi-server)
import redis
from hfortix_core.session import CloudSession
from hfortix_core.session.storage_examples import RedisTokenStorage

r = redis.Redis(host='redis-cluster', port=6379)
storage = RedisTokenStorage(r, namespace="prod:forticloud")

session = CloudSession(
    api_id="...",
    password="...",
    token_storage=storage  # All servers share tokens
)

# PostgreSQL/MySQL (persistent, ACID-compliant)
from sqlalchemy import create_engine
from hfortix_core.session.storage_examples import DatabaseTokenStorage

# PostgreSQL
engine = create_engine('postgresql://user:pass@localhost/fortinet')
# MySQL
# engine = create_engine('mysql+pymysql://user:pass@localhost/fortinet')
# SQLite (single-server)
# engine = create_engine('sqlite:///tokens.db')

storage = DatabaseTokenStorage(engine)
session = CloudSession(api_id="...", password="...", token_storage=storage)

# File-based (simple, no external dependencies)
from hfortix_core.session.storage_examples import FileTokenStorage

storage = FileTokenStorage(directory="/var/lib/forticloud/tokens")
session = CloudSession(api_id="...", password="...", token_storage=storage)

Enterprise Features - Lifecycle Hooks:

Monitor token operations with callbacks for metrics, logging, and alerting:

# Prometheus metrics
from prometheus_client import start_http_server
from hfortix_core.session.hooks_examples import PrometheusHooks

start_http_server(8000)  # Metrics at :8000/metrics
hooks = PrometheusHooks(namespace="prod", service="api")

session = CloudSession(
    api_id="...",
    password="...",
    lifecycle_hooks=hooks  # Automatic metrics collection
)

# Simple callbacks
from hfortix_core.session import SimpleLifecycleHooks

def log_acquisition(event):
    print(f"Token acquired: {event.client_id}, expires {event.expires_in}s")

def alert_failure(event):
    send_alert(f"Token failed: {event.error}")

hooks = SimpleLifecycleHooks(
    on_acquired=log_acquisition,
    on_failed=alert_failure
)

session = CloudSession(api_id="...", password="...", lifecycle_hooks=hooks)

Storage Backend Comparison:

Backend	Best For	Requires	Persistent	Distributed
Redis	Production multi-server	pip install redis	No	Yes
PostgreSQL/MySQL	Production clusters	pip install sqlalchemy psycopg2-binary	Yes	Yes
SQLite	Single-server production	pip install sqlalchemy	Yes	No
File (JSON)	Development, simple deployments	None (built-in)	Yes	No
Memcached	Lightweight caching	pip install pymemcache	No	Yes
In-Memory	Default, simple use	None (built-in)	No	No

CloudSession Global Rate Limiting (NEW):

Configure rate limiting at the session level to apply globally to all clients:

from hfortix_core.session import CloudSession
from hfortix_forticare import FortiCare
from hfortix_fortiztp import FortiZTP

# Create session with global rate limiting
session = CloudSession(
    api_id="your_api_id",
    password="your_password",
    # Global rate limiting - inherited by all clients
    rate_limit=True,
    rate_limit_max_requests=20,       # 20 requests per window
    rate_limit_window_seconds=60.0,   # 60 second window
    rate_limit_strategy="queue",      # Queue excess requests
    circuit_breaker=True,             # Enable circuit breaker
    circuit_breaker_threshold=5,      # Open after 5 failures
)

# Both clients inherit session's rate limiting
fc = FortiCare(session=session)    # Uses 20 req/min from session
fz = FortiZTP(session=session)     # Uses 20 req/min from session

# Individual client can override session defaults
fc_fast = FortiCare(
    session=session,
    rate_limit_max_requests=50,    # Override session's 20 req/min
)

Benefits:

• Configure once: Set rate limiting at session level, all clients inherit
• Consistent limits: Ensure all services respect the same rate limits
• Flexible overrides: Individual clients can override session defaults as needed
• Global enforcement: Rate limiting applies across all services using the session

Rate Limit Tracking

Track API calls and errors across multiple time windows (no enforcement, just visibility):

from hfortix_forticare import FortiCare

# Configure custom limits
fc = FortiCare(
    api_id="...",
    password="...",
    rate_limit_calls_per_min=100,      # 100 calls per minute
    rate_limit_calls_per_hour=1000,    # 1000 calls per hour
    rate_limit_errors_per_hour=10      # 10 errors per hour
)

# Check rate limit status
status = fc.get_rate_limit_status()
print(f"Calls last min: {status['calls_last_min']}/{status['limits']['calls_per_min']}")
print(f"Calls last 5min: {status['calls_last_5min']}")
print(f"Calls last hour: {status['calls_last_hour']}")
print(f"Within limits: {status['within_limits']}")

# Session-wide tracking
with CloudSession(api_id="...", password="...") as session:
    fc = FortiCare(session=session)
    fz = FortiZTP(session=session)
    
    # Check per-client stats
    fc_status = fc.get_rate_limit_status()
    fz_status = fz.get_rate_limit_status()
    
    # Check session-wide stats (all services combined)
    session_status = session.get_rate_limit_status()

Status Response:

{
    "calls_last_min": 45,       # Calls in last 60 seconds
    "calls_last_5min": 180,     # Calls in last 300 seconds
    "calls_last_hour": 523,     # Calls in last 3600 seconds
    "errors_last_min": 0,       # Errors in last 60 seconds
    "errors_last_5min": 1,      # Errors in last 300 seconds
    "errors_last_hour": 2,      # Errors in last 3600 seconds
    "total_calls": 1247,        # Total since client creation
    "total_errors": 5,          # Total errors since creation
    "limits": {                 # Configured limits
        "calls_per_min": 100,
        "calls_per_5min": None,
        "calls_per_hour": 1000,
        "errors_per_min": None,
        "errors_per_5min": None,
        "errors_per_hour": 10
    },
    "within_limits": True       # Whether within all set limits
}

Rate Limiting Enforcement & Circuit Breaker (NEW)

Protect your FortiGate devices from overload with built-in rate limiting and circuit breaker:

from hfortix_core.http import HTTPClient

# Enable rate limiting
client = HTTPClient(
    url="https://192.0.2.10",
    token="your-api-token",
    rate_limit=True,                    # Enable rate limiting
    rate_limit_max_requests=100,        # 100 requests per window
    rate_limit_window_seconds=60.0,     # 60 second window
    rate_limit_strategy="queue",        # Queue overflow requests
    rate_limit_queue_size=50,           # Max 50 queued
)

# Enable circuit breaker
client = HTTPClient(
    url="https://192.0.2.10",
    token="your-api-token",
    circuit_breaker=True,               # Enable circuit breaker
    circuit_breaker_threshold=5,        # Open after 5 failures
    circuit_breaker_timeout=60.0,       # Wait 60s before retry
    circuit_breaker_half_open_calls=3,  # Test with 3 calls
)

# Check stats
stats = client._rate_limiter.get_stats()
print(f"Allowed: {stats['allowed_requests']}")
print(f"Dropped: {stats['dropped_requests']}")
print(f"Efficiency: {stats['rate_limit_efficiency']:.1%}")

state = client.get_circuit_breaker_state()
print(f"Circuit state: {state['state']}")  # disabled, closed, open, half_open

See RATE_LIMITING_GUIDE.md for complete documentation.

Direct Usage (Advanced)

This package is typically not used directly. Instead, it provides the foundation for product-specific packages:

# Instead of using core directly...
# from hfortix_core import HTTPClient

# Use a product package that includes core:
from hfortix_fortios import FortiOSClient

🏗️ What's Included

Exception System

Comprehensive error handling with specific exceptions for all Fortinet API error codes:

from hfortix_core.exceptions import (
    APIError,
    ResourceNotFoundError,
    DuplicateEntryError,
    PermissionDeniedError,
    ValidationError
)

HTTP Client

• Sync and async HTTP client implementations
• Automatic retry with exponential backoff
• Connection pooling and HTTP/2 support
• Request/response logging
• Timeout handling

Utilities

• Data formatting functions
• Response parsing
• Parameter normalization
• Type conversions

🔗 Related Packages

This core package is used by:

• hfortix-fortios - FortiOS/FortiGate client
• hfortix - Meta-package for easy installation

📋 Requirements

• Python 3.10 or higher
• httpx >= 0.24.0
• pydantic >= 2.0.0

📄 License

MIT License - see LICENSE for details.

💬 Support

• 🐛 Report Issues
• 💬 Discussions
• 📖 Documentation

---

Author: Herman W. Jacobsen | LinkedIn | GitHub