Metadata-Version: 2.4
Name: hfortix-core
Version: 0.5.162
Summary: Core foundation for HFortix - Fully-typed Fortinet SDK with async support and type safety
Author-email: "Herman W. Jacobsen" <herman@wjacobsen.fo>
Maintainer-email: "Herman W. Jacobsen" <herman@wjacobsen.fo>
License: Proprietary
Project-URL: Homepage, https://github.com/hermanwjacobsen/hfortix
Project-URL: Repository, https://github.com/hermanwjacobsen/hfortix
Project-URL: Documentation, https://hfortix.readthedocs.io/
Project-URL: Bug Tracker, https://github.com/hermanwjacobsen/hfortix/issues
Project-URL: Changelog, https://github.com/hermanwjacobsen/hfortix/blob/main/CHANGELOG.md
Project-URL: Source Code, https://github.com/hermanwjacobsen/hfortix
Project-URL: Download, https://pypi.org/project/hfortix-core/#files
Keywords: fortinet,fortigate,fortios,api,sdk,typed,type-hints,pydantic,async,asyncio,automation,network-automation,network-administration,configuration-management,infrastructure-as-code,network-as-code
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Intended Audience :: System Administrators
Classifier: Intended Audience :: Information Technology
Classifier: License :: Other/Proprietary License
Classifier: Natural Language :: English
Classifier: Operating System :: OS Independent
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: System :: Networking
Classifier: Topic :: System :: Systems Administration
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Framework :: AsyncIO
Classifier: Framework :: Pydantic
Classifier: Typing :: Typed
Requires-Python: >=3.10
Description-Content-Type: text/markdown
Requires-Dist: httpx[http2]>=0.27.0
Requires-Dist: typing_extensions>=4.0.0

# HFortix Core

[![PyPI version](https://badge.fury.io/py/hfortix-core.svg)](https://pypi.org/project/hfortix-core/)
[![Python 3.10+](https://img.shields.io/badge/python-3.10+-blue.svg)](https://www.python.org/downloads/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)

> **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

```bash
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:

```python
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:**

```python
# 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:

```python
# 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:

```python
# 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:

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

```python
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:**

```python
{
    "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:

```python
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](../../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:

```python
# 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:

```python
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](https://github.com/hermanwjacobsen/hfortix-fortios) - FortiOS/FortiGate client
- [hfortix](https://github.com/hermanwjacobsen/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](LICENSE) for details.

## 💬 Support

- 🐛 [Report Issues](https://github.com/hermanwjacobsen/hfortix-fortios/issues)
- 💬 [Discussions](https://github.com/hermanwjacobsen/hfortix-fortios/discussions)
- 📖 [Documentation](https://github.com/hermanwjacobsen/hfortix)

---

**Author**: Herman W. Jacobsen | [LinkedIn](https://www.linkedin.com/in/hermanwjacobsen/) | [GitHub](https://github.com/hermanwjacobsen)
