Metadata-Version: 2.3
Name: sibi-flux
Version: 2026.1.4
Summary: Sibi Toolkit: A collection of tools for Data Analysis/Engineering.
Author: Luis Valverde
Author-email: Luis Valverde <lvalverdeb@gmail.com>
Requires-Dist: pandas>=2.3.3
Requires-Dist: pyarrow>=22.0.0
Requires-Dist: pydantic>=2.12.5
Requires-Dist: pydantic-settings>=2.12.0
Requires-Dist: dask>=2025.11.0
Requires-Dist: fsspec>=2025.10.0
Requires-Dist: s3fs>=2025.10.0
Requires-Dist: sqlalchemy>=2.0.44
Requires-Dist: psycopg2>=2.9.11
Requires-Dist: pymysql>=1.1.2
Requires-Dist: clickhouse-connect>=0.10.0
Requires-Dist: concurrent-log-handler>=0.9.28
Requires-Dist: rich>=14.2.0
Requires-Dist: filelock>=3.20.1
Requires-Dist: tqdm>=4.67.1
Requires-Dist: watchdog>=6.0.0
Requires-Dist: tornado==6.5.4
Requires-Dist: typer>=0.21.0
Requires-Dist: psutil>=6.1.1
Requires-Dist: httpx>=0.28.1
Requires-Dist: opentelemetry-api>=1.38.0
Requires-Dist: opentelemetry-exporter-otlp>=1.38.0
Requires-Dist: opentelemetry-sdk>=1.38.0
Requires-Dist: deep-translator>=1.11.4
Requires-Dist: pyyaml>=6.0.3
Requires-Dist: distributed>=2025.11.0
Requires-Dist: sibi-flux[distributed,geospatial,mcp] ; extra == 'complete'
Requires-Dist: osmnx>=2.0.7 ; extra == 'geospatial'
Requires-Dist: geopandas>=1.1.2 ; extra == 'geospatial'
Requires-Dist: geopy>=2.4.1 ; extra == 'geospatial'
Requires-Dist: folium>=0.20.0 ; extra == 'geospatial'
Requires-Dist: osmium>=4.2.0 ; extra == 'geospatial'
Requires-Dist: shapely>=2.0.0 ; extra == 'geospatial'
Requires-Dist: networkx>=3.6.1 ; extra == 'geospatial'
Requires-Dist: mcp>=1.1.2 ; extra == 'mcp'
Requires-Dist: fastapi>=0.127.0 ; extra == 'mcp'
Requires-Dist: uvicorn>=0.40.0 ; extra == 'mcp'
Requires-Dist: httpx>=0.28.1 ; extra == 'mcp'
Requires-Python: >=3.11
Provides-Extra: complete
Provides-Extra: geospatial
Provides-Extra: mcp
Description-Content-Type: text/markdown

# SibiFlux

**SibiFlux** is a production-grade resilient data engineering ecosystem designed to bridge the gap between local development, distributed computing, and agentic AI workflows. It provides a unified engine for hybrid data loading (batch + streaming), self-healing distributed operations, and native interfaces for AI agents via the Model Context Protocol (MCP).

```mermaid
graph TD
    subgraph "Agentic Interface (MCP)"
        Agent["AI Agent / Claude"] <--> Router["MCP Router"]
        Router <--> Resources["SibiFlux Resources"]
    end

    subgraph "Solutions Layer (Business Logic)"
        Logistics["Logistics Solutions"]
        Enrichment["Enrichment Pipelines"]
        Cubes["DataCubes"]
    end

    subgraph "SibiFlux Core Engine"
        DfHelper["DfHelper (Unified Loader)"]
        Cluster["Resilient Dask Cluster"]
        Managed["ManagedResource Lifecycle"]
    end

    Resources --> Cubes
    Logistics --> DfHelper
    Cubes --> DfHelper
    DfHelper --> Cluster
```

## Core Architecture

### 1. The Flux Engine (`sibi_flux`)
The foundational library providing resilient distributed primitives.
- **`DfHelper`**: A unified API for loading data from SQLAlchemy, Parquet, or HTTP sources into Pandas or Dask DataFrames.
- **`Dataset`**: A high-level abstraction for hybrid data loading, seamlessly merging historical (Parquet) and live (SQL) data sources.
- **`DfValidator`**: A robust schema enforcement tool that validates DataFrames against strict type maps and generates ClickHouse DDL.
- **`ArtifactOrchestrator`**: An async engine for managing concurrent artifact updates with retries, backoff, and worker isolation.
- **`ManagedResource`**: A rigorous lifecycle management system for async resources, ensuring clean shutdown, signal handling, and observability.
- **`Dask Cluster`**: A self-healing distributed runtime that detects worker failures, manages re-connection, and enforces "The Nuclear Option" for test isolation.

### 2. Agentic Interface (`mcp`)
Native support for the **Model Context Protocol (MCP)**, allowing AI agents to directly interact with SibiFlux data structures.
- **Expose DataCubes**: Automatically turn any `DataCube` into a queryable MCP Resource.
- **Tooling**: Register Python functions as tools callable by agents.

## Key Capabilities

### Hybrid Data Loading
SibiFlux implements a "Hot/Cold" architecture for seamless data access:
- **Historical Data**: Read efficiently from partitioned Parquet archives (S3/Local).
- **Live Data**: Query operational SQL databases for real-time changes.
- **Automatic Merge**: `DfHelper` and `Dataset` automatically stitch these sources together, handling schema evolution and deduplication.

### Resilient Distributed Compute
The SibiFlux Dask wrapper provides:
- **Auto-Healing**: Clients that automatically reconnect if the scheduler dies.
- **Safe Persistence**: Wrappers like `safe_persist` that retry operations on network jitter.
- **Smart Partitioning**: Automated repartitioning to prevent "small file problem" in Parquet outputs.

### Observability
Built-in integration with OpenTelemetry (OTel):
- structured logging with correlation IDs.
- distributed tracing across async boundaries.

## Quick Start

### Installation

```bash
# Base installation (Core Engine only)
pip install sibi-flux

# For Distributed Computing (Dask Cluster support)
pip install "sibi-flux[distributed]"

# For Geospatial capabilities (OSMnx, GeoPandas, etc.)
pip install "sibi-flux[geospatial]"

# For MCP Agentic Interface support
pip install "sibi-flux[mcp]"

# Complete installation (Simulates the "All-in-One" environment)
pip install "sibi-flux[complete]"
```

## API Examples

### 1. Data Loading (`Dataset`)

The `Dataset` class provides a high-level abstraction for hybrid loading.

```python
from sibi_flux import Dataset
from solutions.logistics.readers.products import ProductsParquetReader, ProductsSqlReader

class ProductsDataset(Dataset):
    historical_reader = ProductsParquetReader
    live_reader = ProductsSqlReader
    date_field = "created_at"

# Load hybrid data (merges Parquet + SQL)
ds = ProductsDataset(start_date="2023-01-01", end_date="2023-01-31")
df = await ds.aload()  # Returns a Dask DataFrame
```

### 2. Schema Validation (`DfValidator`)

Ensure your data meets strict type requirements and generate DDL for ClickHouse.

```python
from sibi_flux import DfValidator

# Define expected schema
SCHEMA = {
    "product_id": "Int64[pyarrow]",
    "price": "Float64[pyarrow]",
    "name": "string[pyarrow]"
}

validator = DfValidator(df)
validator.validate_schema(SCHEMA)
validator.standardize_data_quality()

# Generate ClickHouse DDL
ddl = validator.generate_clickhouse_ddl("products_table")
print(ddl)
```

### 3. Data Enrichment (`AsyncDfEnricher`)

Enrich a base DataFrame with data from other sources using `AttachmentSpec`.

```python
from sibi_flux.df_enricher import AsyncDfEnricher, AttachmentSpec

specs = [
    AttachmentSpec(
        key="customer_info",
        required_cols={"customer_id"},
        attachment_fn=fetch_customer_data,  # Async function returning DF
        left_on=["customer_id"],
        right_on=["id"],
        drop_cols=["id"]
    )
]

enricher = AsyncDfEnricher(base_df=orders_df, specs=specs)
enriched_df = await enricher.enrich()
```

### 4. Orchestration (`ArtifactOrchestrator`)

Manage concurrent updates of multiple artifacts with retries and worker isolation.

```python
from sibi_flux.orchestration import ArtifactOrchestrator

orchestrator = ArtifactOrchestrator(
    wrapped_classes={"daily": [ProductsDataset, OrdersDataset]},
    max_workers=3,
    retry_attempts=3
)

# Updates all 'daily' artifacts concurrently
results = await orchestrator.update_data("daily")
```

### 5. Distributed Compute (`Dask Cluster`)

Execute resilient operations that survive scheduler restarts.

```python
from sibi_flux.dask_cluster import safe_compute, get_persistent_client

client = get_persistent_client()

# Safe compute with auto-retry logic
result = safe_compute(df.groupby("category").sum())
```

### 6. Resource Management (`ManagedResource`)

Create lifecycle-safe components.

```python
from sibi_flux.core import ManagedResource

class MyResource(ManagedResource):
    async def _acleanup(self):
        await self.db_connection.close()
        self.logger.info("Cleaned up!")

async with MyResource() as res:
    await res.do_work()
# Automatically cleaned up here
```

### 7. Agentic Interface (`sibi_flux.mcp`)

Seamlessly bridge your data with AI Agents.

**Server Side (Expose Resources)**

```python
from sibi_flux.mcp import BaseMCPRouter
from products import ProductsDataset

# Create an MCP Router compatible with FastAPI
router = BaseMCPRouter(name="data-server")

# Automatically register a Dataset as an MCP Resource
# Agent can now read `sibi://ProductsDataset`
router.register_cube_resource(ProductsDataset)

# Register a custom tool
@router.tool()
def calculate_vat(amount: float) -> float:
    return amount * 0.2
```

**Client Side (Consume Resources)**

```python
from sibi_flux.mcp import GenericMcpClient

# Connect to the MCP Server
async with GenericMcpClient(url="http://localhost:8000/sse") as client:
    # Read the resource (returns JSON data from the Dataset)
    data = await client.read_resource("sibi://ProductsDataset")
    
    # Call a tool
    vat = await client.call_tool("calculate_vat", arguments={"amount": 100.0})
```

### 8. Datacube Generation (`gen_dc.py`)

Automate the creation of Datacube classes and Field Maps from your database schema.

**Configuration (`discovery_params.yaml`)**

Define your generation rules in a hierarchical configuration file:

```yaml
defaults:
  backend: sqlalchemy
  class_suffix: Dc

discovery:
  whitelist_file: whitelist.yaml
  rules_file: discovery_rules.yaml

generation:
  enable_field_maps: true
```

**Commands**

1.  **Discover**: Introspects the database and updates your whitelist/registry.
    ```bash
    uv run poe dc-discover
    ```
    *   **Whitelist**: Explicitly define tables to generate. Support `custom_name` to override class names.
    *   **Rules**: Regex-based patterns to match tables.

2.  **Sync**: Generates Python code (Datacubes and Field Maps) based on the registry.
    ```bash
    uv run poe dc-sync --force
    ```

**Key Features**
- **Custom Naming**: Add `custom_name: MyCube` to `whitelist.yaml` to override generated names.
- **Hierarchical Config**: Strict validation of generation parameters.
- **Field Maps**: Auto-generates type-safe mapping files for every table.
