Metadata-Version: 2.4
Name: satisfactoscript
Version: 1.4.1
Summary: Declarative data engineering framework for Spark and Databricks Lakehouse.
Author-email: julhouba <houbartjulien80@gmail.com>
License: MIT
Project-URL: Homepage, https://github.com/julhouba/satisfactoscript
Project-URL: Documentation, https://julhouba.github.io/satisfactoscript
Project-URL: Repository, https://github.com/julhouba/satisfactoscript
Project-URL: Changelog, https://github.com/julhouba/satisfactoscript/blob/main/CHANGELOG.md
Project-URL: Bug Tracker, https://github.com/julhouba/satisfactoscript/issues
Keywords: databricks,spark,pyspark,delta-lake,data-engineering,lakehouse,declarative,yaml,etl,pipeline,bronze-silver-gold,semantic-layer
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Operating System :: OS Independent
Classifier: Development Status :: 5 - Production/Stable
Classifier: Intended Audience :: Developers
Classifier: Intended Audience :: Science/Research
Classifier: Topic :: Database
Classifier: Topic :: Scientific/Engineering :: Information Analysis
Classifier: License :: OSI Approved :: MIT License
Requires-Python: >=3.9
Description-Content-Type: text/markdown
Requires-Dist: pyyaml>=6.0
Requires-Dist: python-dotenv>=1.0.0
Provides-Extra: spark
Requires-Dist: pyspark>=3.3.0; extra == "spark"
Requires-Dist: delta-spark>=2.0.0; extra == "spark"
Provides-Extra: snowflake
Requires-Dist: snowflake-snowpark-python>=1.0.0; extra == "snowflake"
Provides-Extra: bigquery
Requires-Dist: google-cloud-bigquery>=3.0.0; extra == "bigquery"
Requires-Dist: google-cloud-bigquery-storage>=2.0.0; extra == "bigquery"
Provides-Extra: llm-openai
Requires-Dist: openai>=1.0.0; extra == "llm-openai"
Provides-Extra: llm-anthropic
Requires-Dist: anthropic>=0.30.0; extra == "llm-anthropic"
Provides-Extra: llm-google
Requires-Dist: google-generativeai>=0.5.0; extra == "llm-google"
Provides-Extra: semantic-pdf
Requires-Dist: fpdf2>=2.7.0; extra == "semantic-pdf"
Requires-Dist: matplotlib>=3.7.0; extra == "semantic-pdf"
Provides-Extra: serving
Requires-Dist: mlflow<3,>=2.12.0; extra == "serving"
Requires-Dist: openai>=1.0.0; extra == "serving"
Provides-Extra: semantic-full
Requires-Dist: openai>=1.0.0; extra == "semantic-full"
Requires-Dist: anthropic>=0.30.0; extra == "semantic-full"
Requires-Dist: fpdf2>=2.7.0; extra == "semantic-full"
Requires-Dist: matplotlib>=3.7.0; extra == "semantic-full"
Provides-Extra: dev
Requires-Dist: pytest>=7.0; extra == "dev"
Requires-Dist: pytest-cov>=4.0; extra == "dev"
Requires-Dist: ruff>=0.1.0; extra == "dev"
Requires-Dist: pytest-mock>=3.10.0; extra == "dev"

# SatisfactoScript Framework (v1.4.0)

> **An Enterprise-Ready, Declarative Data Engineering Framework for Spark and Databricks Lakehouse.**

SatisfactoScript transforms complex PySpark pipelines into standardized, readable, and maintainable declarative contracts. By strictly decoupling the **What** (YAML schemas) from the **How** (Python business rules), it enables robust Bronze → Silver → Gold pipelines on Databricks and local Spark, optimized for Power BI Direct Query.

---

## Key Capabilities

- **YAML Pipeline Schemas** — define sources, joins, transformations, and quality checks in readable YAML files. No more 1,000-line PySpark notebooks.
- **External Declarative Sources** — read CSV, Parquet, JSON, Avro, ORC, Delta, or Text files from ADLS Gen2, DBFS, S3-compatible mounts, or local storage directly from YAML — zero Python required.
- **Self-documenting operators** — `region:equals:EMEA`, `amount:greater_than_equal:100`, `status:in:ACTIVE,PENDING`. No need to memorize abbreviations.
- **Smart Sandbox** — in interactive mode, source tables are auto-resolved to your personal sandbox schema. Missing tables are transparently cloned from the main schema.
- **Business Logic Isolation** — register pure Python/PySpark rules with `@RuleRegistry.register_rule()`.
- **Semantic Layer** — auto-generate semantic YAML models from your Gold tables via LLM, then query them in natural language with `GenBIAgent`.
- **Local Spark Development Mode** — run the full framework locally with `local[*]` PySpark + Delta Lake. No Databricks cluster required.
- **Environment Aware** — auto-detects Dev / QA / Prod Databricks catalogs at runtime with per-user sandbox isolation.
- **Direct Query Optimized** — pre-calculate OBT, YoY shifts, and distinct counts in the Gold layer to keep Power BI DAX ultra-light.

---

## Architecture

```
Bronze (Raw)  →  Silver (Standardized)  →  Gold (Semantic / OBT)
                          │
        ┌─────────────────┴────────────────────┐
        │          SatisfactoScript             │
        │  ├─ 1. Declarative Schema (dict)      │
        │  ├─ 2. Rule Registry (Python logic)   │
        │  ├─ 3. Delta I/O & Z-Order            │
        │  └─ 4. Semantic Layer + LLM Agent     │
        └──────────────────────────────────────┘
                          │
                 Power BI (Direct Query)
```

---

## Installation

```bash
pip install satisfactoscript
pip install "satisfactoscript[spark]"        # Databricks + local PySpark/Delta

# Optional — LLM providers for the Semantic Layer
pip install satisfactoscript[llm-anthropic]   # Claude (Anthropic)
pip install satisfactoscript[llm-openai]      # GPT (OpenAI)
pip install satisfactoscript[llm-google]      # Gemini (Google)

# Optional — PDF export for session history
pip install satisfactoscript[semantic-pdf]
```

---

## Local Development Setup

Run the same Spark-oriented framework on your laptop without a Databricks cluster. PySpark runs in `local[*]` mode, Delta Lake is enabled, and Apache Derby serves as the embedded metastore (no installation required).

### 1. Create `config.yaml` at your project root

```yaml
default_env: LOCAL
priority_check: [DEV, QA, PROD, LOCAL]

environments:
  LOCAL:
    catalog: null          # null = no Unity Catalog, triggers local mode
    is_production: false

  DEV:
    catalog: "my_dev_catalog"
    is_production: false

  PROD:
    catalog: "my_prod_catalog"
    is_production: true
```

When `catalog` is `null`, the engine skips all Databricks catalog checks and boots in local mode. When Databricks credentials are present (`DATABRICKS_HOST`, `DATABRICKS_TOKEN`, `DATABRICKS_CLUSTER_ID`), the engine tries Databricks Connect first and falls back to local automatically.

### 2. (Optional) `.env` for Databricks credentials

```ini
# Only needed for Databricks Connect (remote cluster from IDE)
DATABRICKS_HOST=https://your-workspace.azuredatabricks.net
DATABRICKS_TOKEN=dapiXXXXXX
DATABRICKS_CLUSTER_ID=0123-456789-abcdef

# LLM provider for the Semantic Layer
ANTHROPIC_API_KEY=sk-ant-...
```

### 3. Boot the engine

```python
from satisfactoscript import SatisfactoEngine

engine = SatisfactoEngine()  # auto-discovers config.yaml upwards from cwd
# → boots local[*] Spark + Delta Lake + Derby metastore
# → env=LOCAL, no catalog prefix, sandbox suffix = _<your_os_user>
```

Session detection priority:
1. **Active Databricks session** — running inside a notebook or cluster
2. **Databricks Connect v2** — IDE + remote cluster via `.env` credentials
3. **Local PySpark + Delta Lake** — fully offline, zero configuration

---

## Quick Start: Building a Pipeline

### Option A — YAML file (recommended)

Store schemas as YAML files in your project for reuse and version control.

```yaml
# schemas/gold/fact_transactions.yaml
tables:
  - name: "{{ catalog }}.silver.transactions"
    alias: tx
    filter:
      - "region:equals:EMEA"
      - "customer_id:is_not_null"
    quality_checks:
      drop_duplicates_on: [transaction_id]
      drop_nulls_in: [amount, transaction_date]

business_rules:
  - flag_high_value

select_final:
  - [transaction_id, id]
  - [transaction_date, date, [cast:date]]
  - [amount, amount_eur, [cast:double, round:2]]
  - [is_high_value, is_high_value]
```

```python
from satisfactoscript import SatisfactoEngine, RuleRegistry, load_schema
from pyspark.sql import functions as F

engine = SatisfactoEngine()

@RuleRegistry.register_rule()
def flag_high_value(df):
    return df.withColumn("is_high_value", F.when(F.col("amount") >= 1000, 1).otherwise(0))

# Load schema from file — {{ catalog }} is replaced with engine's active catalog
schema = load_schema("schemas/gold/fact_transactions.yaml", params=engine.default_params)

# Preview before running (no Spark execution)
engine.describe_schema(schema)

# Run and write to Delta
engine.run_process_to_table(schema_dict=schema, target_layer="gold", target_table_name="fact_transactions")
```

### Option B — Inline YAML string

For quick iterations or notebook-local schemas.

```python
from satisfactoscript import SatisfactoEngine, parse_schema

engine = SatisfactoEngine()

schema = parse_schema("""
tables:
  - name: "{{ catalog }}.silver.transactions"
    alias: tx
    filter:
      - "region:equals:EMEA"
select_final:
  - [transaction_id, id]
  - [amount, amount_eur, [cast:double]]
""", params=engine.default_params)

engine.run_process_to_table(schema_dict=schema, target_layer="gold", target_table_name="fact_transactions")
```

### Execution patterns

| Method | Use case |
|---|---|
| `run_process_to_table(schema, layer, table)` | Process a schema and write to a single Delta table |
| `run_process_and_split(schema, split_values, layer, base_name, col)` | Split result into one table per value (e.g. one table per region) |
| `run_union_sources_to_table(schema, partitions, src_layer, tgt_layer, table, bases, alias, dedup_after_union=True)` | Union partitioned source tables, process, write |
| `optimize_table(layer, table, zorder_cols)` | Run Delta OPTIMIZE with optional ZORDER BY |
| `describe_schema(schema)` | Dry-run summary: sources, joins, columns — no Spark execution |

---

## Semantic Layer

The Semantic Layer lets you auto-generate structured YAML models from your Gold tables using an LLM, then query them in natural language.

### Step 1 — Build a semantic model from a Gold table

`SemanticBuilder` inspects the table schema, optionally reads a Jupyter notebook and a business glossary, calls the LLM, validates the output (up to 3 attempts), and registers the model in `semantic_catalog.yaml`.

```python
from satisfactoscript import SatisfactoEngine
from satisfactoscript.semantic.builder import SemanticBuilder
from satisfactoscript.semantic.llm_provider import get_llm_provider

engine = SatisfactoEngine()
builder = SemanticBuilder(
    llm_provider=get_llm_provider(),   # auto-detects from env vars
    output_dir="semantic_models",
)

builder.build(
    model_name="kpi_orders",
    split_value="erp",
    table="gold.fact_orders",
    layer="gold",
    source_notebook="notebooks/fact_orders.ipynb",  # optional — adds business context
    glossary_path="glossaries/orders.json",          # optional — injects domain terms
    description="Order KPIs from ERP (SAP source)",
    tags=["orders", "revenue", "erp"],
)
# → writes semantic_models/kpi_orders.erp.yaml
# → updates semantic_catalog.yaml
```

The generated YAML describes dimensions (with SQL expressions + types) and metrics (with SQL + aggregation type). It is fully human-readable and editable after generation.

### Step 2 — Load the Semantic Engine

```python
from satisfactoscript.semantic.semantic import SemanticEngine

sem = SemanticEngine(engine, models_dir="semantic_models")
# → loads semantic_catalog.yaml only at startup (lightweight)

# Browse available models
sem.list_models()
sem.list_models(tags=["revenue"], summary=True)

# Inspect a specific model
sem.get_model_summary("kpi_orders.erp")
```

### Step 3 — Query in natural language with GenBIAgent

```python
from satisfactoscript.agentic.agent import GenBIAgent

agent = GenBIAgent(semantic_engine=sem, llm_provider=get_llm_provider())

# Ask a business question
response = agent.ask("What is the total revenue by region for last quarter?")

if response.success:
    response.result.data.show()       # PySpark DataFrame
elif response.needs_clarification:
    print(response.clarification_message)

# Export the session as PDF
agent.history.to_pdf("session_export.pdf")
```

### LLM provider auto-detection

`get_llm_provider()` selects the provider based on environment variables:

| Variable | Provider |
|---|---|
| `ANTHROPIC_API_KEY` | Claude (Anthropic) |
| `OPENAI_API_KEY` | GPT (OpenAI) |
| `GOOGLE_API_KEY` | Gemini (Google) |
| `LLM_PROVIDER=anthropic\|openai\|google` | Force a specific provider |

---

## External Declarative Sources

Read external files (CSV, Parquet, JSON, Avro, ORC, Delta, Text) from any blob storage or local filesystem directly from your YAML schema — no Python required. All standard pipeline features (filter, quality checks, dev_limit, joins) apply identically to external sources.

```yaml
tables:
  - name: raw_orders
    alias: orders
    source:
      type: csv
      path: "abfss://container@account.dfs.core.windows.net/bronze/orders/*.csv"
      options:
        header: "true"
        inferSchema: "true"
    filter:
      - "status:is_not_null"
    quality_checks:
      drop_nulls_in: [order_id, amount]
    dev_limit: 5000

  - name: "{{ catalog }}.silver.products"   # regular Delta table — mix freely
    alias: products
```

Supported `type` values: `csv`, `parquet`, `json`, `avro`, `orc`, `delta`, `text`.  
Path supports `{{ param }}` injection — use `load_schema(..., params={"base_path": "..."})` to parameterize.

> **Note:** External sources are a Spark/Databricks feature. Legacy non-Spark backends are not part of the stabilized product path.

---

## Adding Business Rules

Rules are decoupled from execution notebooks. Define them in a centralized `rules.py` and import it before running the engine.

```python
from pyspark.sql import functions as F
from satisfactoscript import RuleRegistry

@RuleRegistry.register_rule()
def enrich_transaction_data(df):
    return (
        df
        .withColumn(
            "is_high_value",
            F.when(F.col("amount") >= 1000, 1).otherwise(0)
        )
        .withColumn(
            "clean_status",
            F.when(F.lower(F.col("status")).isin(["completed", "done"]), "Paid")
             .otherwise("Pending")
        )
        .fillna({"amount": 0.0})
    )
```

---

## YAML Schema Reference

### Filter operators

All operators use full English names. SQL abbreviations (`eq`, `gte`, etc.) are accepted as aliases.

| Operator | Example | Notes |
|---|---|---|
| `equals` | `"status:equals:ACTIVE"` | alias: `eq` |
| `not_equals` | `"status:not_equals:CANCELLED"` | alias: `ne` |
| `greater_than` | `"amount:greater_than:100"` | alias: `gt` |
| `less_than` | `"age:less_than:18"` | alias: `lt` |
| `greater_than_equal` | `"score:greater_than_equal:90"` | alias: `gte` |
| `less_than_equal` | `"qty:less_than_equal:5"` | alias: `lte` |
| `in` | `"status:in:ACTIVE,PENDING"` | comma or `;` separated |
| `not_in` | `"region:not_in:FR,DE"` | |
| `contains` | `"label:contains:promo"` | |
| `not_contains` | `"label:not_contains:test"` | |
| `starts_with` | `"ref:starts_with:ORD"` | |
| `ends_with` | `"email:ends_with:@corp.com"` | |
| `is_null` | `"discount:is_null"` | no value |
| `is_not_null` | `"customer_id:is_not_null"` | no value |
| `like` | `"name:like:J%"` | SQL LIKE pattern |
| `not_like` | `"name:not_like:test%"` | |
| `sql` | `"sql:amount > threshold"` | raw SQL escape hatch |

For values containing commas, use the dict form: `{column: city, operator: in, value: ["New York, NY", "Paris"]}`.

### `select_final` operations

Operations are applied left-to-right on the source column.

| Operation | Example | Result |
|---|---|---|
| `cast:type` | `cast:date`, `cast:double` | Type casting |
| `upper` / `lower` | `upper` | String case |
| `trim` | `trim` | Strip whitespace |
| `round:N` | `round:2` | Round to N decimals |
| `abs` | `abs` | Absolute value |
| `length` | `length` | String length |
| `to_date:fmt` | `to_date:yyyy-MM-dd` | Parse string to date |
| `nvl:val` | `nvl:0` | Replace null with value |
| `coalesce:val` | `coalesce:0` | Same as `nvl:` |
| `lit:val` | `lit:ERP` | Constant value |
| `expr:sql` | `expr:year(order_date)` | Arbitrary SQL expression |
| `split:sep,idx` | `split:-,1` | Split string, take index |
| `substring:start,len` | `substring:1,4` | Substring |
| `when:op:val` | `when:equals:DONE` | Condition (use with `then:` / `else:`) |

**Shorthand for constant columns:**
```yaml
select_final:
  - [literal:ERP, source_system]          # adds source_system = 'ERP'
  - [literal:0.0, discount, [cast:double]]
```

**OR filter groups:**
```yaml
filter_groups:
  - ["region:equals:EMEA", "status:is_not_null"]   # EMEA AND not null
  - ["region:equals:APAC"]                          # OR APAC
```

**Keep all columns + add computed ones:**
```yaml
keep_all_columns: true
add_columns:
  - [amount, amount_rounded, [round:2]]
  - [literal:ERP, source_system]
```

**Quality checks:**
```yaml
quality_checks:
  drop_nulls_in: [customer_id, order_date]
  drop_duplicates_on: [order_id, sku_id]
```

**Compact join syntax:**
```yaml
join:
  - table_from: [orders, customer_id]
    table_to: [customers, id]
    type: left
```

**Parameter injection:**
```yaml
tables:
  - name: "{{ catalog }}.silver.orders"
    filter:
      - "region:equals:{{ region }}"
```
```python
schema = load_schema("schemas/fact_orders.yaml", params={**engine.default_params, "region": "FR"})
```

**Dev sampling (ignored in job/prod):**
```yaml
dev_limit: 10000    # schema-level
tables:
  - name: silver.orders
    dev_limit: 5000  # table-level override
```

---

## Smart Sandbox

In interactive (non-job, non-prod) mode, source tables are transparently resolved to your personal sandbox schema (`schema_XXXX` where `XXXX` is derived from your username).

| Situation | Behavior |
|---|---|
| Table exists in `silver_XXXX` | Loaded directly — transparent |
| Table missing in `silver_XXXX` | Logged warning + shallow clone from `silver` + load |
| Schema `silver_XXXX` doesn't exist | Schema created + table cloned + load |
| Table missing in main schema | `ValueError` raised |

Configure behavior in `config.yaml`:
```yaml
sandbox:
  missing_table: copy   # copy (default) | error
```

> **Note:** Add `.satisfacto_user` to your `.gitignore`. This file is auto-generated to cache your sandbox suffix.

---

## Developer Tools

```python
# Force a specific environment (bypass auto-detection)
engine = SatisfactoEngine(force_env="LOCAL")

# Preview a schema without executing Spark
engine.describe_schema(schema)

# List registered rules and loaders
RuleRegistry.list_rules()
RuleRegistry.list_loaders()
```

---

## Configuration Reference (`config.yaml`)

```yaml
default_env: LOCAL                      # Fallback if no Databricks catalog is reachable
priority_check: [DEV, QA, PROD, LOCAL]  # Detection order

environments:
  LOCAL:
    catalog: null                       # null = local mode (no Unity Catalog)
    is_production: false

  DEV:
    catalog: "my_dev_catalog"
    is_production: false

  QA:
    catalog: "my_qa_catalog"
    is_production: false

  PROD:
    catalog: "my_prod_catalog"
    is_production: true

sandbox:
  missing_table: copy                   # copy (default) | error
```

The engine also reads the `semantic_views_schema` key (optional) to resolve the target schema for semantic views.
