Metadata-Version: 2.4
Name: smartinno
Version: 1.2.25
Summary: Message Bus infrastructure for multi-cloud systems.
Author: Safari Pro Engineering Ltd
License: MIT
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Operating System :: OS Independent
Requires-Python: >=3.10
Description-Content-Type: text/markdown
Requires-Dist: redis>=5.0.0
Requires-Dist: python-dotenv>=1.0.0
Requires-Dist: pulsar-client>=3.0.0
Requires-Dist: temporalio>=1.0.0
Requires-Dist: requests>=2.31.0
Requires-Dist: pydantic>=2.0.0
Requires-Dist: cryptography>=41.0.0
Provides-Extra: otel
Requires-Dist: opentelemetry-api>=1.20.0; extra == "otel"
Dynamic: author
Dynamic: classifier
Dynamic: description
Dynamic: description-content-type
Dynamic: license
Dynamic: provides-extra
Dynamic: requires-dist
Dynamic: requires-python
Dynamic: summary

# Unified Redis & Pulsar Python SDK (v1.2.0)

> 📖 **Developer Guide & Manual**:  
> 👉 [**Complete Developer Guide & Architecture Manual**](../DEVELOPER_GUIDE.md) — Comprehensive guide with logic flows, encryption mechanics, replay attack defense, and live cluster test results.  
> 👉 [**End-to-End Cryptographic Lifecycle & Hotel Booking Scenario**](END_TO_END_LIFECYCLE_SCENARIO.md) — Full production flow from mobile payment to Temporal Saga execution.

A unified, robust Python wrapper for interacting with multiple Redis architectures and Apache Pulsar clusters seamlessly. This SDK uses the Factory Design Pattern to abstract away the complexity of connecting to different infrastructure setups. It natively enforces data standardization (JSON enveloping), multi-tenant HKDF key derivation, transparent AES-256-GCM encryption at rest and in transit, Dead-Letter Queue (DLQ) poison pill quarantine across all Redis data structures and Pulsar messaging topics, and dynamic, uniform Redis Cluster slotting and key distribution across all 16,384 hash slots.

## Core Features (v1.2.0)

- **Redis Cluster Dynamic Slotting & Uniform Key Distribution**:
  - **Eliminates Fixed Single-Slot Bottleneck**: Previously, hardcoded `{service_name}` hash tags collapsed 100% of all cache keys, hashes, lists, sets, and streams onto a single hash slot (slot 3731), overloading a single master node while leaving all other cluster masters idle.
  - **`ClusterSlottingStrategy` Enum**:
    - `DISTRIBUTED` (Default): Full key CRC16 hashing across all 16,384 slots (`service:action:key`), uniformly distributing reads, writes, and memory across all available cluster nodes.
    - `ENTITY_TAGGED`: Automatically wraps the primary entity identifier in curly braces (`service:action:{entity}:sub`), co-locating related records for the same entity while distributing different entities across nodes.
    - `VIRTUAL_SHARDS`: Controlled hashing across $N$ virtual shard buckets (`{service_shard_X}`), preventing hotspots while managing shard counts.
    - `LEGACY_FIXED`: Backward-compatible mode retaining the fixed `{service_name}` hash tag.
  - **Preserves Explicit Hash Tags**: Any caller-supplied hash tag `{...}` (e.g. `{order_100}:items`) is always preserved for intentional slot co-location, transactional pipelines, or multi-key operations.
  - **Cluster Slot Diagnostics & Inspection**: Built-in `get_key_slot(key)` and `get_slot_distribution(keys)` utilities calculate CRC16 slots, detect skew (`is_skewed`), report distribution ratios, and map slots to physical nodes via `get_node_for_key(key)`.
- **Dead-Letter Queue (DLQ) & Poison Pill Quarantine**:
  - `DeadLetterQueueHandler` automatically quarantines malformed, tampered, or repeatedly failing messages to a dedicated DLQ (`{topic}:dlq` or `{topic}-dlq`) with full audit diagnostics (`dlq_id`, `original_topic`, `error`, `stack_trace`, `attempts`, `reason`, `failed_at`).
  - Tampered HMAC signatures (`HMAC_TAMPER_ALERT`) and replay attacks (`NONCE_REPLAY_ATTACK`) are immediately quarantined without executing worker code.
  - Redis Streams track retry attempts and acknowledge from PEL upon reaching `max_retries`. Pulsar integrates `negative_acknowledge()` redeliveries before DLQ routing.
- **PulsarSmartRouter Delayed Delivery Fix & Continuous Streaming**:
  - Fixed delayed delivery (`delete_after`) to route through `PulsarProducerAdapter.send(..., deliver_at=...)`, preserving the cryptographic envelope with AES-256-GCM encryption and 3-day Redis nonces.
  - Enhanced `listen_to` and added `listen_stream()` with transparent AES-256-GCM payload unwrapping and DLQ poison pill quarantine.
- **Unified Cryptographic Envelope Parity Across Apache Pulsar & Redis**:
  - Every message sent to Apache Pulsar (`smart_send`, `send`, `PulsarProducerAdapter.send/send_async`, `publish_to`) is strictly wrapped into the standardized JSON cryptographic envelope (`request_id`, `nonce`, `group`, `event`, `type`, `service`, `tenant_name`, `payload`, `timestamp`, `signature`).
  - Automatic topic tenant extraction (`persistent://{tenant}/{namespace}/{topic}`) guarantees complete tenant isolation.
  - Automatic broker `partition_key` mapping to `request_id` for deterministic Key_Shared distribution.
  - Transparent consumer unwrapping and decryption (`receive_unwrapped`, `smart_listen`, `@client.on`).
- **Transparent AES-256-GCM Encryption At Rest Across All Redis Data Structures**:
  - **Strings**: `set`, `get`, `mset`, `mget`, `setnx`, `setex`, `getset`, `getdel`.
  - **Hashes**: `hset`, `hmset`, `hget`, `hmget`, `hgetall`, `hvals`, `hsetnx`, `hdel`, `hexists`, `hlen`, `hkeys`.
  - **Lists**: `lpush`, `rpush`, `lpop`, `rpop`, `lrange`, `lindex`, `llen`, `lrem`, `ltrim`.
  - **Sets**: `sadd`, `smembers`, `spop`, `srandmember`, `srem`, `sismember`, `scard`.
  - **Sorted Sets (ZSets)**: `zadd`, `zrange`, `zrevrange`, `zrangebyscore`, `zrevrangebyscore`, `zscore`, `zcard`, `zcount`, `zrank`, `zrevrank`, `zrem`.
  - **Streams**: `xadd`, `xread`, `xreadgroup`, `xautoclaim`, `xclaim`.
  - **PubSub**: `publish`, `subscribe`.
- **HKDF (RFC 5869) Multi-Tenant Key Derivation**: Keys derived per-tenant using mTLS certificates and master entropy. Cross-tenant reads fail tag authentication.
- **Dual-Mode Nonce Replay Prevention**:
  - **Pre-Registered Mode**: 3-day TTL registration + atomic Lua GETDEL verification & immediate deletion.
  - **Sliding-Window Mode**: 300s window drift validation + post-ingest `SET NX EX 300`.
  - **Strict Mode**: `require_registered_nonce=True` drops unregistered nonces.
- **Timing-Safe HMAC-SHA256 Signatures**: Protects against tampering of amounts, IDs, and tenants.
- **Distributed Multi-Tier Idempotency**: Concurrency locking (`STARTED`), cached responses (`COMPLETED`), and retry unlock (`FAILED`).
- **Resilient Message Bus**: Single-authoritative pipeline via Apache Pulsar with Redis Streams fallback.
- **Circuit Breaker Resiliency**: Guarded by `@_circuit_protected` with CLOSED/OPEN/HALF_OPEN transitions.
- **Verified Production Test Suite**: 58/58 automated unit tests + 100% passing live 6-node Redis cluster integration test suite.



---

## Supported Architectures & Use Cases

| Architecture | `UseCase` Enum | Underlying Driver | Best For |
|---|---|---|---|
| **Intelligent Router** | `AUTO` | `IntelligentRouterAdapter` | Hands-off optimal routing based on command type. |
| **Native Redis Cluster** | `HIGH_CONCURRENCY` | `redis.cluster.RedisCluster` | Horizontal scaling, massive reads/writes. |
| **Sentinel + HAProxy** | `HEAVY_TRANSACTIONS` | `redis.Redis` (HAProxy) | Pub/Sub, robust transactions. |
| **Hybrid (Predixy Proxy)** | `MICROSERVICES` | `redis.Redis` (Standard) | Easy multi-tenant connections, stateless clients. |

---

## Usage Guide

### 1. Basic Import
```python
from smartinno.config import RedisConfig, UseCase
from smartinno.factory import RedisClientFactory
```

### 2. Configuration & Enterprise Settings
Initialize your configuration and pass `service_name` to define your application's namespace. You can also pass enterprise features like retries and connection limits.

```python
cluster_config = RedisConfig(
    host="redis-node-1", 
    port=6379, 
    password="your_password",
    startup_nodes=[{"host": "redis-node-1", "port": 6379}],
    
    # Core SDK configs
    service_name="my_billing_app",
    
    # Enterprise configs
    ssl=False,
    max_connections=100,
    socket_timeout=5.0,
    retry_on_timeout=True,
    retry_backoff=True, # Enables Exponential Backoff
    retry_backoff_retries=3
)
```

### 3. Native Cluster (`HIGH_CONCURRENCY`)
```python
from smartinno import RedisConfig, UseCase, RedisClientFactory, ClusterSlottingStrategy

cluster_config = RedisConfig(
    host="redis-node-1", 
    port=6379, 
    service_name="safari_pro",
    cluster_slotting_strategy=ClusterSlottingStrategy.DISTRIBUTED # Default: uniform multi-slot distribution
)

client = RedisClientFactory.get_client(cluster_config, UseCase.HIGH_CONCURRENCY)

# Automatically formats key to 'safari_pro:cache:user_101'
# Distributed across all 16,384 cluster slots (CRC16 full-key hash)
client.set("user_101", {"name": "Alice"})

# Check exact Redis Cluster CRC16 hash slot (0 to 16,383)
slot = client.get_key_slot("user_101")
print(f"Key slot: {slot}")

# Inspect batch distribution metrics across all cluster nodes
dist = client.get_slot_distribution(["user_1", "user_2", "user_3", "user_4"])
print(f"Unique slots: {dist['unique_slots']}, Skewed: {dist['is_skewed']}")

# Transparently unwraps the envelope and decrypts AES-256-GCM
print(client.get("user_101"))
```

#### Cluster Slotting Strategies (`ClusterSlottingStrategy`)
| Strategy | Key Format Pattern | Distribution Behavior | Best For |
|---|---|---|---|
| `DISTRIBUTED` (Default) | `{service}:{action}:{key}` | Full key CRC16 hashed across all 16,384 slots | General caching, balanced memory & network across all cluster nodes |
| `ENTITY_TAGGED` | `{service}:{action}:{{entity}}:{sub}` | Co-locates sub-keys for the same entity on 1 node | Multi-key pipelines or transactions per entity (e.g. `{user:101}`) |
| `VIRTUAL_SHARDS` | `{{{service}_shard_X}}:{action}:{key}` | Partitions keys across $N$ virtual buckets ($0 \dots N-1$) | Controlled horizontal partition bounds without slot skew |
| `LEGACY_FIXED` | `{{{service}}}:{action}:{key}` | Pins all keys to single slot (slot 3731 for `safari_pro`) | Strict backward compatibility |

> **Note on Custom Hash Tags**: Any caller-supplied hash tag `{...}` (e.g. `client.set("{order_99}:items", ...)`) is **always preserved verbatim** across all strategies, allowing deliberate slot co-location when needed.


### 4. Hybrid Proxy (`MICROSERVICES`)
Connects to the Native Cluster through a Predixy proxy. The client patches pipeline transactions automatically because Predixy handles clustering efficiently.

```python
hybrid_config = RedisConfig(
    host="localhost", 
    port=6381, 
    password="your_password",
    service_name="my_billing_app"
)

client = RedisClientFactory.get_client(hybrid_config, UseCase.MICROSERVICES)

# Pipelines are patched automatically!
pipe = client.pipeline(transaction=True)
pipe.set("a", 1)
pipe.set("b", 2)
pipe.execute()
```
> **Note on Predixy Pipelines:** Predixy Proxy does not support cross-node `MULTI/EXEC` efficiently via python pipelines. If you pass `transaction=True`, the Hybrid adapter catches it, prints a notice, and gracefully overrides it to `transaction=False`.

### 5. Intelligent Router (`AUTO`)
The absolute easiest way to consume Redis. Provide the configurations for all environments to the factory, request `UseCase.AUTO`, and the router will send your queries to the optimal architecture.

```python
router_client = RedisClientFactory.get_client(
    config=global_config, 
    use_case=UseCase.AUTO
)

# Routes to Native Cluster
router_client.set("cache_key", "fast_data")

# Routes to Sentinel
router_client.publish("global_events", "system_ready")

# Routes to Hybrid Proxy
router_client.xadd("audit_trail", {"action": "login"})
```

---

## Running the Tests

To verify that the SDK routes correctly to all three architectures:

1. **Inside Docker (Recommended)**: Use `run_tests_in_docker.ps1` to run the tests securely inside the same docker network as the Redis instances.
2. **Local Windows**: Run `python test_sdk.py`. (Note: Native Cluster tests may fail gracefully if it requires docker DNS resolution).
# Redis SDK Developer Guide

This developer guide provides step-by-step instructions on how to use the Safari Pro Redis SDK, including how to connect to Redis, read data (both pending and new) from streams, and utilize the built-in Communication Service for sending SMS, emails, and in-app notifications.

## 1. Connecting to Redis

The SDK abstracts the underlying Redis architecture (Cluster, Sentinel, or Hybrid/Predixy) using the `RedisClientFactory` and the `UseCase` enum.

```python
from smartinno.config import RedisConfig, UseCase
from smartinno.factory import RedisClientFactory

# Initialize the configuration with your service namespace
config = RedisConfig(
    host="redis-node-1",
    port=6379,
    password="your_password",
    startup_nodes=[{"host": "redis-node-1", "port": 6379}],
    service_name="my_service"
)

# Request a client tailored for high concurrency (e.g., Native Cluster)
client = RedisClientFactory.get_client(config, UseCase.HIGH_CONCURRENCY)
```

## 2. Reading Data from Redis Streams

Redis Streams is the backbone of the SDK's messaging architecture. You can use the unified `xread` method to consume data from streams. The SDK handles JSON envelop unwrapping automatically.

### Reading Pending Messages
To read all existing/pending messages in a stream from the beginning, pass `0` (or `"0-0"`) as the stream ID.

```python
# Read pending messages from the beginning (ID 0)
messages = client.xread({"communication:events": 0}, count=10)

if messages:
    for stream_name, events in messages:
        print(f"Found {len(events)} pending messages in {stream_name}")
        for msg_id, msg_data in events:
            print(f"Message ID: {msg_id}, Data: {msg_data}")
```

### Reading New Messages
To listen for new messages, pass `$` as the stream ID and use the `block` parameter (in milliseconds). The client will block and wait until a new message arrives.

```python
# Block and wait for up to 5000ms for new messages
new_messages = client.xread({"communication:events": "$"}, count=1, block=5000)

if new_messages:
    for stream_name, events in new_messages:
        for msg_id, msg_data in events:
            print(f"New event received: {msg_id}")
```

> [!TIP]
> The Intelligent Router (`UseCase.AUTO`) will automatically route stream commands like `XADD` and `XREAD` to the Hybrid architecture!

---

## 3. Using the Communication Service

The `CommunicationService` is a robust module integrated into the SDK that queues messages onto Redis Streams for external communication workers to process.

```python
from smartinno.communication_service import communication_service
```

### Sending SMS

You can send SMS messages to one or multiple recipients.

```python
success = communication_service.send_sms(
    recipients=["+254700123456", "+254711654321"],
    message="Your Safari Pro booking is confirmed!",
    template="booking_confirmation",
    metadata={"booking_id": "BK-001"}
)
```

### Sending Emails

The email functionality supports plain text, HTML, and templates.

```python
email_result = communication_service.send_email(
    recipients=["guest@example.com"],
    subject="Booking Confirmation",
    body="Your safari booking BK-001 has been confirmed.",
    html_content="<h1>Booking Confirmed</h1>",
    template_name="booking_email",
    context_data={"guest_name": "Alice"},
    metadata={"booking_id": "BK-001"}
)

print(email_result["message_id"]) # Message ID tracking
```

### Sending In-App Notifications

Queue an in-app notification that will be pushed to the user's notification center in real-time.

```python
inapp_result = communication_service.send_inapp_notification(
    user_id="user_123",
    title="Booking Confirmed",
    message="Your booking #BK-001 has been confirmed",
    notification_type="booking_confirmed",
    action_url="/bookings/BK-001",
    priority="high"
)
```

### Multi-Channel Notifications (All-in-One)

Send to multiple channels simultaneously using `send_notification()`.

```python
results = communication_service.send_notification(
    notification_type="payment_received",
    recipients={
        "email": ["finance@safari.co"],
        "sms": ["+254700999888"],
        "user_id": "user_456",
    },
    content={
        "email_subject": "Payment Received",
        "email_body": "Payment of $500 received for booking BK-002.",
        "sms_message": "Payment of $500 received. Ref: BK-002",
        "inapp_title": "Payment Received",
        "inapp_message": "We received your $500 payment for booking BK-002",
    },
    metadata={"booking_id": "BK-002", "amount": "500.00", "priority": "high"}
)

print(results) # {"email": {...}, "sms": True, "inapp": True}
```
