Metadata-Version: 2.4
Name: payloaded
Version: 0.2.2
Summary: Transform flat tabular data into nested, hierarchical API payloads with grouping and reconciliation.
Project-URL: Homepage, https://github.com/svr-s/payloaded
Project-URL: Repository, https://github.com/svr-s/payloaded
Project-URL: Issues, https://github.com/svr-s/payloaded/issues
Author-email: Vatsa <228140210+svr-s@users.noreply.github.com>
License: MIT
License-File: LICENSE
Keywords: api,batching,data-transformation,json,pandas,payload
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
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: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: Utilities
Requires-Python: >=3.9
Requires-Dist: pandas>=1.5.0
Provides-Extra: all
Requires-Dist: pandas>=1.5.0; extra == 'all'
Requires-Dist: polars>=0.20.0; extra == 'all'
Provides-Extra: dev
Requires-Dist: build>=1.0.0; extra == 'dev'
Requires-Dist: polars>=0.20.0; extra == 'dev'
Requires-Dist: pytest>=7.0.0; extra == 'dev'
Requires-Dist: twine>=4.0.0; extra == 'dev'
Provides-Extra: polars
Requires-Dist: polars>=0.20.0; extra == 'polars'
Description-Content-Type: text/markdown

# payloaded

[![PyPI version](https://img.shields.io/pypi/v/payloaded.svg)](https://pypi.org/project/payloaded/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)

**`payloaded`** (`import payloaded as pld`) transforms flat tabular data (pandas or polars DataFrames) into nested, hierarchical JSON payloads for REST/GraphQL APIs with conditional routing, multi-level batch chunking, and mathematical row reconciliation.

---

## Key Features

- **Conditional Routing & Negation Rules**: Route records to distinct payload structures using positive matches (`['New', 'Update']`) or negation rules (`['~Terminated']`).
- **Condition Value Partitioning**: Records matching a rule are partitioned by their actual status values so disparate states are never mixed in the same payload.
- **Consistent Canonical Configuration**: Single, uniform configuration layout for both conditional and unconditional jobs.
- **Column Name & Index Flexibility**: Map source columns using either column name strings (whitespace-tolerant) or 0-based integer column indices.
- **Multi-Level Hierarchy & Grouping**: Group rows across parent, child, and grandchild entities (`root`, `orders`, `orders.line_items`) using single or composite keys.
- **Cascading Batch Limits**: Enforce maximum item limits at any level of the hierarchy (e.g. max 50 orders, max 100 line items).
- **Audit & Zero-Loss Reconciliation**: Generates an auditable DataFrame containing payload indices, rules, partitioned values, row counts, running totals, and Postman-ready JSON payloads.

---

## Installation

```bash
pip install payloaded
```

---

## Canonical Configuration Schema

`payloaded` uses a single, consistent configuration format:

```python
config = {
    "condition_source_key": "Status",  # Column name, 0-based column index, or "" if unconditional
    "conditions": [
        {
            "condition_rule": ["~Terminated"],  # Rule list, or [] if unconditional
            "wrapper_envelope": { ... },         # Optional API envelope containing "{payload_template}"
            "payload_template": [ ... ],         # Payload JSON structure skeleton
            "entities": [ ... ]                  # Hierarchy, grouping, and field mappings
        }
    ]
}
```

---

## Syntax Standards: Where to Use Curly Braces `{}`

To keep configurations clean, predictable, and avoid syntax errors, `payloaded` maintains a strict distinction between **Declarations** (plain strings) and **Dynamic Substitutions** (curly braces `{}`):

* **Declarations & Identifiers (NO `{}`)**: Schema keys, column pointers, and hierarchy paths must be plain text:
  - `"payload_key": "order_id"` *(declares the placeholder identifier)*
  - `"source_key": "Order_ID"` *(points to the source column name or 0-based column index)*
  - `"condition_source_key": "Status"` *(points to routing column name or index)*
  - `"path": "orders.line_items"` *(structural tree coordinate)*
  - `"group_by": ["order_id"]` *(grouping key identifiers)*

* **Dynamic Placeholders & References (MUST USE `{}`)**:
  - `payload_template`: `"order_id": "{order_id}"` *(signals a placeholder to be hydrated)*
  - `formula`: `"lower({Location}) & '-' & {Batch_Number}"` *(signals columns to be evaluated)*

### Why Formulas Require Braces `{col}`
Enclosing column names in `{}` within formulas is essential:
1. **Prevents Function Name Collisions**: If your DataFrame has a column literally named `strip`, `date`, `int`, or `round`, writing `strip({strip})` explicitly distinguishes the function `strip()` from the data column `{strip}`.
2. **Supports Spaces and Symbols**: Allows referencing columns with spaces like `{Batch Number}` or `{Order-ID}` without syntax errors.

> **Note**: `payloaded` strictly adheres to these definitions. Misplaced braces (e.g. putting `{}` in `payload_key` or `source_key`) are not silently altered and will be treated as literal text.

---

## Quickstart

### 1. Conditional Routing Example

```python
import pandas as pd
import payloaded as pld

# Flat tabular dataset with status column
df_source = pd.DataFrame([
    {"Batch_ID": "B01", "Status": "New", "Order_ID": "ORD_1", "SKU": "A1", "Qty": 2},
    {"Batch_ID": "B01", "Status": "New", "Order_ID": "ORD_2", "SKU": "A2", "Qty": 1},
    {"Batch_ID": "B01", "Status": "Update", "Order_ID": "ORD_3", "SKU": "B1", "Qty": 5},
    {"Batch_ID": "B01", "Status": "Terminated", "Order_ID": "ORD_4", "SKU": "C1", "Qty": 0},
])

# Canonical configuration with conditional routing & negation
config = {
    "condition_source_key": "Status",
    "conditions": [
        {
            # Matches all active statuses (~Terminated)
            # Automatically partitions 'New' and 'Update' into separate payloads
            "condition_rule": ["~Terminated"],
            "payload_template": [
                {
                    "batch_id": "{batch_id}",
                    "orders": [
                        {
                            "order_id": "{order_id}",
                            "items": [{"sku": "{sku}", "qty": "{qty}"}]
                        }
                    ]
                }
            ],
            "entities": [
                {
                    "path": "root",
                    "repeat_limit": 10,
                    "group_by": ["batch_id"],
                    "mappings": [{"payload_key": "batch_id", "source_key": "Batch_ID"}],
                },
                {
                    "path": "orders",
                    "repeat_limit": 50,
                    "group_by": ["order_id"],
                    "mappings": [{"payload_key": "order_id", "source_key": "Order_ID"}],
                },
                {
                    "path": "orders.items",
                    "repeat_limit": 100,
                    "mappings": [
                        {"payload_key": "sku", "source_key": "SKU"},
                        {"payload_key": "qty", "formula": "int(coalesce({Qty}, 0))"},
                    ],
                },
            ],
        },
        {
            # Specialized payload structure for terminations
            "condition_rule": ["Terminated"],
            "payload_template": [
                {
                    "batch_id": "{batch_id}",
                    "cancellations": [{"order_id": "{order_id}"}]
                }
            ],
            "entities": [
                {
                    "path": "root",
                    "group_by": ["batch_id"],
                    "mappings": [{"payload_key": "batch_id", "source_key": "Batch_ID"}],
                },
                {
                    "path": "cancellations",
                    "group_by": ["order_id"],
                    "mappings": [{"payload_key": "order_id", "source_key": "Order_ID"}],
                },
            ],
        },
    ]
}

# Generate auditable payloads DataFrame and meta summary
df_payloads, meta_df = pld.build_payloads(
    source=df_source,
    config=config,
    output_format="json_string"  # or "dict"
)

# Inspect high-level batch metrics and condition coverage
print(meta_df)

# Verify zero data loss with mathematical reconciliation
audit_report = pld.reconcile(df_payloads, expected_rows=len(df_source))
print(audit_report.summary())
```

---

### 2. Unconditional Example

When all records follow the same structure unconditionally, set `condition_source_key: ""` and `condition_rule: []`:

```python
config = {
    "condition_source_key": "",
    "conditions": [
        {
            "condition_rule": [],
            "payload_template": [{"id": "{id}", "name": "{name}"}],
            "entities": [
                {
                    "path": "root",
                    "mappings": [
                        {"payload_key": "id", "source_key": "ID"},
                        {"payload_key": "name", "source_key": "Name"},
                    ],
                }
            ],
        }
    ],
}
```

---

## Formula & Expression Engine

`payloaded` includes a sandboxed, zero-dependency AST formula engine (no `eval()`) that allows you to transform, concatenate, slice, and generate values directly in your configuration:

```json
{
  "payload_key": "tracking_code",
  "formula": "lower(strip({Location}))[1:5] & '-' & replace(strip({Batch_Number}), ':', '')"
}
```

### 1. Concatenation & String Operations

You can concatenate multiple columns, literals, and transforms using either:
- **`&` (Ampersand)**: Excel / PowerQuery style concatenation (recommended)
- **`+` (Plus)**: Python / JavaScript style string concatenation

```json
{
  "payload_key": "full_name",
  "formula": "{first_name} & ' ' & {last_name}"
}
```

```json
{
  "payload_key": "tracking_code",
  "formula": "upper(strip({country})) & '-' & lpad({id}, 6, '0') & '-' & replace({batch_num}, ':', '')"
}
```

---

### 2. Built-in Function Reference

| Function | Description | Example |
|---|---|---|
| `upper(val)` / `lower(val)` | Case transformations | `upper({status})` |
| `strip(val)` / `trim(val)` | Whitespace trimming | `strip({code})` |
| `replace(val, old, new)` | Substring replacement | `replace({phone}, '-', '')` |
| `slice(val, start, end)` or `[start:end]` | Native slicing | `{sku}[0:4]` |
| `lpad(val, len, char)` / `rpad(...)` | String padding | `lpad({id}, 5, '0')` $\rightarrow$ `'00042'` |
| `coalesce(a, b, ...)` | First non-empty value | `coalesce({alt_phone}, {phone}, 'N/A')` |
| `date_format(val, in_fmt, out_fmt)` | Date format conversion | `date_format({dt}, '%Y-%m-%d', '%d/%m/%Y')` |
| `now(format)` | UTC timestamp generator | `now('%Y-%m-%dT%H:%M:%SZ')` |
| `uuid()` | Generates unique UUID v4 | `uuid()` |
| `int(val)` / `float(val)` / `str(val)` | Type casting | `int({qty})` |
| `json(val)` | Parses string into native JSON object/array | `json({raw_orders})` |

---

## Static Wrapper Envelopes

Enterprise APIs (e.g., ADP, Workday, Salesforce) commonly wrap batch payload arrays inside boilerplate envelopes with headers and service metadata. 

Use **`wrapper_envelope`** directly inside any condition with the `"{payload_template}"` (or `"{payload}"`) placeholder. The engine will chunk and reconcile the business data inside `payload_template`, then seamlessly inject the batch into the envelope:

```json
{
  "conditions": [
    {
      "wrapper_envelope": {
        "batchHeader": {
          "batchId": "{batch_id}",
          "sourceSystem": "ERP_INGEST",
          "timestamp": "{created_at}"
        },
        "payload": {
          "orders": "{payload_template}"
        }
      },
      "payload_template": [
        {
          "orderId": "{order_id}",
          "lineItems": [ ... ]
        }
      ],
      "entities": [ ... ]
    }
  ]
}
```

Any `{...}` placeholders in the wrapper (such as `{batch_id}` and `{created_at}`) are automatically populated from the batch's representative record.

---

## Sequence Counters & Scoping

Child entities in API payloads (e.g. invoice lines, order items) frequently require auto-incrementing line numbers that don't exist in source tabular data. `payloaded` provides stateful sequence generators with explicit scoping:

```python
sequence(start=1, step=1, scope="parent")   # Default: Resets per parent entity
sequence(start=100, step=1, scope="global") # Never resets: counts continuously across all payloads
```

### Scoping Behavior

| Scope | Under `line_items` | Under `orders` | Across Split Chunks |
|---|---|---|---|
| **`parent`** *(default)* | Resets to `start` for **each Order** | Resets to `start` for **each Batch** | **Continues across chunks** (e.g. Order 101 chunk 1 has lines 1–40; chunk 2 continues with 41–80; next Order 102 resets to 1) |
| **`global`** | Counts continuously across all rows | Counts continuously across all rows | Never resets (e.g. 100, 101, 102...) |

#### Example: Line Item Numbering
```json
{
  "path": "orders.line_items",
  "mappings": [
    {
      "payload_key": "line_number",
      "formula": "sequence(start=1, scope='parent')"
    },
    {
      "payload_key": "sku",
      "source_key": "SKU"
    }
  ]
}
```

---

## Blank & Optional Field Handling (`omit_if_blank`)

By default, missing or `NaN` values in your source data resolve to JSON `null`:
```json
{"sku": "A1", "discount_code": null, "notes": null}
```

If an upstream API schema rejects `null` or empty fields for optional keys, use **`omit_if_blank: true`**.

### Behavior
- **Default**: `false` (optional field, does not need to be specified).
- **When `true`**: Evaluates whether the rendered value is `None`, `NaN`, empty string `""`, or whitespace-only `"   "`. If blank, the key is **completely omitted** from the hydrated JSON object.
- **Valid 0 / False preservation**: Numeric `0`, `0.0`, and boolean `false` are considered valid data and are **never** omitted.
- **Inheritance**: Can be configured at the **field level** or at the **entity level** (where all fields in that entity inherit the setting unless overridden).

#### Example: Field-Level & Entity-Level Omission
```json
{
  "path": "orders.line_items",
  "omit_if_blank": false,
  "mappings": [
    {
      "payload_key": "sku",
      "source_key": "SKU"
    },
    {
      "payload_key": "promo_code",
      "source_key": "Promo_Code",
      "omit_if_blank": true
    },
    {
      "payload_key": "gift_message",
      "source_key": "Gift_Message",
      "omit_if_blank": true
    }
  ]
}
```
If a row has `"SKU": "A1"` and empty cells for `Promo_Code` and `Gift_Message`, the output payload becomes:
```json
{
  "sku": "A1"
}
```
*(No `promo_code` or `gift_message` keys are present)*

---

## Wide-to-Long Wildcard Unpivot (`*`)

Tabular exports often store repeating items horizontally across columns (wide format) rather than normalized across rows:
```csv
location,batchnumber,orderid,ordername,ordervalue,orderid2,ordername2,ordervalue2,orderid3,ordervalue3
```

`payloaded` supports automatic wide-to-nested unpivoting using **wildcard asterisks (`*`)** in `source_key`:

```json
{
  "path": "orders",
  "mappings": [
    {
      "payload_key": "id",
      "source_key": "orderid*"
    },
    {
      "payload_key": "name",
      "source_key": "ordername*",
      "omit_if_blank": true
    },
    {
      "payload_key": "value",
      "source_key": "ordervalue*"
    }
  ]
}
```

### Key Advantages
1. **Dynamic Token Discovery**: No need to hardcode `orderid1`, `orderid2`, ..., `orderid42`. It scans the DataFrame headers and discovers all present suffixes/tokens dynamically (e.g. `""`, `"2"`, `"3"`, ..., `"42"`).
2. **Flexible Wildcard Placement**: Works for prefix, suffix, or infixed patterns (e.g. `orderid*`, `item_*_sku`, `line_*`).
3. **Resilient to Missing/Jagged Columns**: If an export generated `orderid3` and `ordervalue3` but completely omitted the column `ordername3`, `payloaded` will **not crash**. It safely resolves missing columns to `null` (or omits them if `omit_if_blank: true`).
4. **Empty Slot Pruning**: If an entity slot is completely empty for a row (e.g. only 2 orders present in a 5-slot file), blank entries are discarded automatically.
5. **Universal Placement**: Can be used on **any entity level** across your payload hierarchy (child items, grandchildren, or root items).

---

## Cell Explode & Normalization (`source_column`)

When data arrives with embedded/semi-structured objects packed directly inside a single column cell (e.g. JSON strings or Python dicts/lists), use `source_column` on the entity config to automatically explode and normalize them into child items:

```json
{
  "path": "orders",
  "source_column": "orders_raw",
  "mappings": [
    {
      "payload_key": "id",
      "source_key": "orderid*"
    },
    {
      "payload_key": "amount",
      "source_key": "ordervalue"
    },
    {
      "payload_key": "name",
      "source_key": "ordername",
      "omit_if_blank": true
    }
  ]
}
```

### Supported Cell Structures
1. **List of Objects**: `[{"orderid": "ord1", "ordervalue": 100}, {"orderid": "ord2", "ordervalue": 200}]` $\rightarrow$ Unpacks directly into child records.
2. **Sparse Dictionaries**: Combined with `"omit_if_blank": true`, missing keys in individual elements are cleanly omitted without failing.
3. **Inconsistent Keys with Wildcards**: Wildcard source keys (e.g. `orderid*`) match fields inside the cell objects (e.g. matching `orderid` in element 1 and `orderid2` in element 2).
4. **Key-Value Maps**: `{"ord1": "100", "ord2": "200"}` $\rightarrow$ Map dynamically via virtual keys `source_key: "__key__"` and `source_key: "__value__"`.
5. **Raw JSON Passthrough**: If you want to retain the raw parsed JSON structure without decomposing it into child entities, use the `json({column})` formula function.

---

## Dual DataFrame Support: Polars & Pandas

`payloaded` dynamically adapts to your data engineering stack. It accepts either **`pandas.DataFrame`** or **`polars.DataFrame`** as the `source` input and automatically returns the **exact same DataFrame type**:

```python
import polars as pl
import payloaded as pld

# 1. Read high-performance Polars DataFrame
pldf = pl.read_parquet("orders.parquet")

# 2. Generates payloads and returns a polars.DataFrame automatically
polars_payloads = pld.build_payloads(source=pldf, config=config, output_format="dict")
print(type(polars_payloads))  # <class 'polars.dataframe.frame.DataFrame'>
```

- **Zero Heavy Forced Overhead**: In serverless environments (like AWS Lambda), Polars-only users don't need pyarrow or heavy native libraries.
- **Consistent Output**: The output table always has the exact same 7 canonical columns and schema whether returned as Polars or Pandas.

---

## Chunking & Memory-Constrained Streaming (`chunksize`)

When processing massive tabular datasets or executing inside constrained memory budgets (e.g., 256MB AWS Lambda functions), `payloaded` provides two built-in chunking strategies:

### 1. In-Memory Batching (`chunksize=N`)
Processes the source data in slices under the hood to cap peak memory, while seamlessly maintaining the cumulative `running_total` across all batches and returning a single, unified DataFrame:

```python
# Processes in slices of 50,000 rows
df_payloads, meta_df = pld.build_payloads(
    source=df, 
    config=config, 
    chunksize=50000
)
```

### 2. Generator Streaming Mode (`stream=True`)
Yields mini-DataFrames matching the source type (`pandas` or `polars`) one chunk at a time, allowing you to stream payloads directly to external APIs without accumulating the full output table in memory:

```python
for chunk_df in pld.build_payloads(source=df, config=config, chunksize=10000, stream=True):
    # chunk_df is a mini-DataFrame of 7 columns ready to dispatch
    for row in chunk_df.iter_rows(named=True):  # or chunk_df.iterrows()
        post_to_api(row["payload"])
```

- **Group-Boundary Awareness**: When top-level grouping keys (`group_by`) are configured on the root entity, `payloaded` automatically avoids severing a parent group across chunk boundaries.

---

## Metadata Summary DataFrame (`meta_df`)

To inspect batch sizes, payload counts, and row coverage across distinct conditions without writing custom groupby logic, `build_payloads` automatically returns `(payloads_df, meta_df)` as a 2-element tuple:

```python
# 1. Returned directly from build_payloads
payloads_df, meta_df = pld.build_payloads(df, config=config)

# 2. Or computed from an existing payloads DataFrame
meta_df = pld.summarize(payloads_df)
```

### Structure of `meta_df`

| Column | Type | Description |
|---|---|---|
| `condition_rule` | `str` | Condition rule name (or `ALL` in total row) |
| `condition_value` | `str` | Partitioned value (e.g. `Create`, `Update`, `TOTAL`) |
| `payload_count` | `int` | Number of distinct payloads produced |
| `total_rows` | `int` | Exact sum of source rows packed into those payloads |
| `avg_rows_per_payload` | `float` | Average rows packed per payload batch |
| `min_rows` | `int` | Minimum payload batch size |
| `max_rows` | `int` | Maximum payload batch size |
| `node_totals` | `dict` | Total unique entity counts per node/path without double-counting split payloads |

*Returns either a `pandas.DataFrame` or `polars.DataFrame` matching your input type automatically.*

---

## Output DataFrame Structure

The resulting DataFrame contains 8 canonical columns for end-to-end traceability, entity sizing, and Postman/API dispatch:

| Column | Type | Description |
|---|---|---|
| `index` | `int` | Sequential 1-based payload identifier |
| `condition_rule` | `str` | Matching condition rule expression (e.g. `~Terminated`) |
| `condition_value` | `str` | Actual partitioned condition value (e.g. `New`, `Update`) |
| `rows_in_payload` | `int` | Exact count of source tabular rows packed into this payload |
| `running_total` | `int` | Cumulative row count within the current source DataFrame |
| `node_counts` | `dict` | Exact counts of entities / child items present inside this individual payload |
| `payload` | `str` / `dict` | Complete, nested JSON payload string or Python dictionary |
| `source_filename` | `str` | Origin source identifier for multi-input traceability |

---

## Built-In Testing Framework (`payloaded.testing`)

For data engineering pipelines and CI/CD validation, `payloaded` provides testing assertions out of the box (`from payloaded import testing` or `from payloaded.testing import ...`):

```python
from payloaded.testing import (
    assert_payload_schema,
    assert_reconciled,
    assert_node_counts,
    assert_meta_balanced,
)

# 1. Assert canonical 8-column schema integrity
assert_payload_schema(df_payloads)

# 2. Assert zero data loss with exact source row count
assert_reconciled(df_payloads, expected_rows=len(df_source))

# 3. Assert entity node counts across all payloads or per payload
assert_node_counts(
    df_payloads,
    expected_totals={"orders": 100, "orders.items": 450},
)

# 4. Assert metadata summary balance and summary totals
assert_meta_balanced(
    meta_df,
    expected_payload_count=10,
    expected_total_rows=len(df_source),
)
```

---

## License

This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.


