origpy 0.1.3

Origin-aware values for Python

Origpy

Origpy is a small Python library for values that remember where they came from. Wrap a value with track(...), transform it with normal Python functions, and inspect its origin later without dragging in a framework.

It is useful when a script, notebook, service, or data cleanup job needs a lightweight answer to: "why does this value have this value?"

Install

pip install origpy

Origpy supports Python 3.11 and newer. It has no runtime dependencies outside the standard library.

Quick Start

from origpy import track

port = track(5432, source_type="config", source="settings.json")
dsn = port.map(lambda value: f"postgres://localhost:{value}/app")

print(dsn.value)
# postgres://localhost:5432/app

print(dsn.lineage())
# - transform:__main__.<lambda> fp=sha256:...
#   - config:settings.json fp=sha256:...

Track is the only wrapper you need to know. It exposes:

• item.value
• item.origin
• item.map(fn)
• item.combine(other, fn)
• item.tag(*tags)
• item.redact()
• item.lineage()

Why This Exists

Python code often builds important values from files, environment variables, API responses, and small transformations. By the time a value reaches the part of the program that uses it, the trail is usually gone.

Origpy keeps that trail next to the value. It does not try to be a database lineage system, a config framework, or a policy engine. It is deliberately just a value wrapper with safe metadata and a few practical helpers.

Config Provenance

This is the shape Origpy is best at: a value assembled from a file plus a small runtime override.

from pathlib import Path

from origpy import from_env, read_json

base = read_json("service.json", base_dir=Path("/app/config"))

endpoint = base.map(lambda data: data["service"]["endpoint"], source="service.endpoint")
timeout = from_env(
    "SERVICE_TIMEOUT",
    default=base.value["service"]["timeout_seconds"],
    sensitive=False,
    redacted=False,
)

public_config = endpoint.combine(
    timeout,
    lambda url, seconds: {"endpoint": url, "timeout_seconds": int(seconds)},
    source="runtime config",
)

print(public_config.value)
print(public_config.lineage())

Environment values are sensitive and redacted by default. Opt out only for values you are comfortable showing, such as a timeout or feature flag.

For a complete runnable version, see examples/config_provenance.py.

Traced Logic

@traced is useful when the interesting operation is already a normal function. Tracked inputs are unwrapped before the function runs, and their origins become parents of the result.

from origpy import track, traced

@traced(label="invoice total")
def calculate_invoice(subtotal_cents: int, tax_rate: float, discount_cents: int = 0) -> dict[str, int]:
    taxable = max(subtotal_cents - discount_cents, 0)
    tax_cents = round(taxable * tax_rate)
    return {"total_cents": taxable + tax_cents, "tax_cents": tax_cents}

subtotal = track(12_000, source_type="cart", source="cart:8421")
discount = track(1_500, source_type="promotion", source="spring-credit")

invoice = calculate_invoice(subtotal, 0.0825, discount_cents=discount)
print(invoice.value)
print(invoice.lineage())

See examples/traced_logic.py for the full example.

API Response Cleanup

from origpy import track

payload = track(
    {"id": "ord_1001", "amount": "42.50", "currency": "usd", "status": "PAID"},
    source_type="api",
    source="GET /v1/orders/ord_1001",
    tags=("orders",),
)

normalized = payload.map(
    lambda order: {
        "order_id": order["id"],
        "amount_cents": int(float(order["amount"]) * 100),
        "currency": order["currency"].upper(),
    },
    source="normalize_order",
)

total = normalized.map(lambda order: order["amount_cents"], source="amount_cents")
currency = normalized.map(lambda order: order["currency"], source="currency")
summary = total.combine(currency, lambda cents, code: f"{code} {cents / 100:.2f}")

print(summary.value)
# USD 42.50

The lineage is handy when a cleaned field looks wrong and you need to see which response and transform produced it. A fuller version lives in examples/api_response_cleanup.py.

Redaction

Redaction is explicit. It hides values and origin details from repr, lineage rendering, JSON serialization, and the CLI.

from origpy import from_env

credential = from_env("PAYMENT_API_TOKEN", default="<not-set>")

print(credential)
# Track(value=<redacted>, origin=Origin(source_type='env', source='<redacted>', ...))

The raw value still exists at .value; Origpy is not a secret vault. The point is to avoid accidental leakage while debugging or serializing metadata.

Safe Serialization

Origpy uses a small explicit JSON format. There is no pickle, no eval, and no code execution during deserialization.

from origpy import from_json, to_json, track

report = track({"job": "daily-settlement", "records": 128}, source_type="job", source="settlement:daily")
payload = to_json(report)
loaded = from_json(payload)

assert loaded.value == report.value
assert loaded.origin == report.origin

Redacted values stay redacted:

from origpy import from_env, to_json

safe_payload = to_json(from_env("PAYMENT_API_TOKEN", default="<not-set>"))
assert "PAYMENT_API_TOKEN" not in safe_payload

Serialization is intentionally strict: duplicate JSON keys, non-string object keys, NaN, infinity, reserved Origpy keys in user data, oversized payloads, and deeply nested structures are rejected.

See examples/serialization_roundtrip.py.

CLI Inspection

The CLI is intentionally tiny. It inspects an Origpy JSON blob without printing raw tracked values.

origpy inspect tracked.json

Example output:

Track(value=<dict len=2>, origin=Origin(source_type='job', source='settlement:daily'))
- job:settlement:daily fp=sha256:...

For a redacted blob, the source and value are hidden. See examples/cli_inspection.py.

When To Use It

Use Origpy for:

• config provenance in scripts and small services
• debugging transformed API or JSON payloads
• notebook and batch-job reproducibility notes
• lightweight audit trails around derived values

Reach for something else when you need:

• a distributed lineage database
• access control or secret storage
• a workflow engine
• automatic tracking of every Python operation

Security Notes

Origpy is defensive by default:

• no pickle, eval, or exec
• strict JSON deserialization
• duplicate JSON keys are rejected
• file and CLI reads are bounded
• base_dir prevents path escapes in file helpers
• redacted parent provenance stays hidden
• lineage and unwrapping are cycle-safe
• repr and CLI output avoid raw values

The one thing Origpy cannot protect you from is deliberately printing .value. If a tracked value is sensitive, treat .value as sensitive too.

Runnable Examples

The source repository's examples/ directory contains short scripts that are also covered by the test suite:

• config_provenance.py
• api_response_cleanup.py
• traced_logic.py
• serialization_roundtrip.py
• cli_inspection.py

Run one with:

PYTHONPATH=src python examples/config_provenance.py

On Windows PowerShell:

$env:PYTHONPATH = "src"
python examples/config_provenance.py

Development

pip install -e .
python -m unittest

Build a release locally with:

python -m build

License

origpy is published by thundersl under the MIT license.