malha 0.7.1
Sistema Operacional Semântico - Distributed Data Kernel with P2P Replication
License: MIT
Requires Python: >=3.13

# Malha

**Sistema Operacional Semântico** - Kernel de Dados Distribuído com Replicação P2P e Federação Zero-ETL

[![PyPI version](https://badge.fury.io/py/malha.svg)](https://badge.fury.io/py/malha)
[![Coverage](https://img.shields.io/badge/coverage-86%25-green.svg)](https://github.com/kevinqz/malha)
[![Python 3.13+](https://img.shields.io/badge/python-3.13+-blue.svg)](https://www.python.org/downloads/)

## Overview

Malha é um kernel de dados unificado que combina SQL, Grafo e Analytics em uma única abstração:

- **Bitemporalidade Estrita (SCD Type 2)** com Pessimistic Locking
- **Replicação P2P** via Synapse (gRPC mesh networking)
- **Zero-ETL Federation** - Query Postgres, Parquet, S3 diretamente
- **Consistência Dual** (SQL Grafo) via Outbox Pattern
- **Zero-Copy Analytics** (DuckDB + Arrow)
- **Identidade Imutável** (Registro RIDs)

## Quick Start

### Installation

```bash
pip install malha

# Com suporte a replicação distribuída
pip install malha[synapse]
```

### Uso Básico

```python
import asyncio
from malha import connect
from registro import DomainResource, register

# Define seu modelo de domínio
class User(DomainResource):
    name: str
    email: str

register("User", User)

async def main():
    # Context manager garante cleanup automático
    async with connect() as kernel:
        # Criar e salvar
        user = User(name="Alice", email="alice@example.com")
        saved = await kernel.save_versioned(user)
        print(f"Saved: {saved.rid}")
        
        # Atualizar (cria nova versão, mantém histórico)
        saved.name = "Alice Smith"
        updated = await kernel.save_versioned(saved)
        print(f"Version: {updated.version}")

asyncio.run(main())
```

### Zero-ETL Federation (v0.6.0+)

Query dados externos sem pipelines de ETL:

```python
async with connect() as kernel:
    # Query PostgreSQL diretamente
    customers = await kernel.from_postgres(
        "dbname=legacy user=reader host=db.example.com",
        "customers",
        "status = 'active'"
    )
    
    # Query Parquet no S3
    events = await kernel.from_parquet(
        "s3://bucket/events.parquet",
        "year = 2024"
    )
    
    # Query CSV local
    products = await kernel.from_csv("/data/products.csv", "price > 100")
    
    # Query SQLite externo
    orders = await kernel.from_sqlite("/data/legacy.db", "orders")
```

## Architecture

```
┌─────────────────────────────────────────────────────────────┐
│              UnifiedDataManager (Kernel)                    │
├─────────────────────────────────────────────────────────────┤
│  ┌─────────────┐  ┌──────────┐  ┌──────────────────┐       │
│  │ SQL Driver  │  │  Graph   │  │  Analytics       │       │
│  │ (SQLModel)  │  │  (Kùzu)  │  │  (DuckDB+Helix)  │       │
│  └─────────────┘  └──────────┘  └──────────────────┘       │
│         │               │                 │                 │
│         └───────────────┴─────────────────┘                 │
│                         │                                   │
│                    Outbox Pattern                           │
│                         │                                   │
│         ┌───────────────┴─────────────────┐                 │
│         │                                 │                 │
│    Graph Sync                      Synapse (P2P)            │
│    (Local)                         (Distributed)            │
└─────────────────────────────────────────────────────────────┘
                          │
          ┌───────────────┼───────────────┐
          │               │               │
      PostgreSQL      Parquet/S3       CSV/SQLite
      (Federation)    (Federation)     (Federation)
```

## Key Features

### 1. Context Manager Support

```python
# Cleanup automático garantido
async with connect() as kernel:
    await kernel.save_versioned(user)
# kernel.close() é chamado automaticamente
```

### 2. Federation Shortcuts

```python
# Ao invés de dicts verbosos:
await kernel.query_federated({"type": "postgres", "dsn": "...", "table": "..."}, "...")

# Use shortcuts elegantes:
await kernel.from_postgres(dsn, table, filter)
await kernel.from_parquet(path, filter)
await kernel.from_csv(path, filter)
await kernel.from_sqlite(path, table, filter)
```

### 3. Pessimistic Locking (SCD Type 2)

```python
# Atomicidade garantida em ambientes concorrentes
await kernel.save_versioned(resource)
# → SELECT FOR UPDATE (lock)
# → Close old version
# → Insert new version
# → Release lock
```

### 4. P2P Replication (Synapse)

```python
from malha.drivers.synapse import SynapseDriver

synapse = SynapseDriver(
    kernel_ref=None,
    node_id="node-1",
    port=50051,
    peers=["192.168.1.11:50051", "192.168.1.12:50051"]
)

async with connect(replication_driver=synapse) as kernel:
    # Writes são automaticamente replicados!
    await kernel.save_versioned(user)
```

### 5. Observability

```python
from malha import KernelMonitor, KernelInstrumentation

async with connect(enable_monitoring=True) as kernel:
    # Métricas são coletadas automaticamente
    summary = kernel.monitor.get_summary()
    print(f"Operations: {summary['total_operations']}")
```

## Development

```bash
# Instalar dependências
uv sync

# Rodar testes
uv run pytest

# Com cobertura
uv run pytest --cov=malha --cov-report=html

# Lint
uv run ruff check malha/ tests/
```

## API Reference

### Core

| Method | Description |
|--------|-------------|
| `connect()` | Create kernel instance (supports `async with`) |
| `kernel.save_versioned(obj)` | Save with SCD Type 2 versioning |
| `kernel.delete_versioned(obj)` | Soft delete (close active version) |
| `kernel.get(Model, rid=...)` | Retrieve by RID with time travel |

### Federation

| Method | Description |
|--------|-------------|
| `kernel.from_postgres(dsn, table, filter)` | Query PostgreSQL |
| `kernel.from_parquet(path, filter)` | Query Parquet (local/S3) |
| `kernel.from_csv(path, filter)` | Query CSV |
| `kernel.from_sqlite(path, table, filter)` | Query SQLite |
| `kernel.query_federated(config, filter)` | Generic federation |

### Signals

| Signal | Emitted When |
|--------|--------------|
| `post_save` | After successful save |
| `post_delete` | After successful delete |
| `post_ingest` | After data ingestion |

## License

MIT