siphon-dsl 0.11.0

Minimal DSL for API data extraction

Siphon

A minimal DSL for extracting data from JSON APIs.

Like a siphon draws liquid from a container, Siphon draws the data you need from nested JSON structures—just define the paths, and let it flow.

Install

pip install siphon-dsl

Or with uv:

uv add siphon-dsl

Quick Start

from siphon import process

data = {
    "data": {
        "id": "prod_123",
        "items": [
            {"id": 1, "status": "active", "name": "Widget"},
            {"id": 2, "status": "inactive", "name": "Gadget"},
            {"id": 3, "status": "active", "name": "Thing"},
        ],
    }
}

spec = {
    "extract": {
        "id": "$.data.id",
        "all_active": {
            "path": "$.data.items[*]",
            "where": {"status": "active"},
            "select": {"item_id": "id", "item_name": "name"},
            "collect": True,
        },
    }
}

result = process(spec, data)

Output:

{
  "id": "prod_123",
  "all_active": [
    {"item_id": 1, "item_name": "Widget"},
    {"item_id": 3, "item_name": "Thing"}
  ]
}

Features

Feature	Syntax	Description
Simple paths	$.data.id	Extract nested values
Array iteration	$.items[*].name	Traverse arrays
List indexing	items[0].id / names[-1]	Pick a specific list position (supports negative)
Filtering	where: {status: "active"}	Filter by field values
Ancestor filtering	where: {parentId: 123}	Filter by parent-level properties
Projection	select: {new: "old"}	Rename and reshape fields
Coalesce	select: {price: "sale || list"}	First non-null path wins (SQL-style ||)
Collect	collect: true	Return all matches (default: first only)
Reduce	reduce: "min_time"	Aggregate array values to a single result

Spec Format

Simple extraction

extract:
  id: "$.data.id"
  name: "$.data.name"

Extended extraction

extract:
  active_items:
    path: "$.data.items[*]"
    where: {status: "active"}
    select: {item_id: "id", item_name: "name"}
    collect: true

List indexing ([N])

Any path segment can drill into a list at a specific position with [N]. Negative indices count from the end; out-of-bounds or non-list values resolve to None. Indexing composes with ||:

extract:
  rows:
    path: "$.items[*]"
    select:
      # Try the first tier, fall back to a flat price field
      amount: "tieredPrices[0].amount || price.amount"
      last_name: "names[-1]"
    collect: true

[*] and [N] are distinct — [*] iterates, [N] picks one element.

Coalesce in select

A select path string may contain || to declare a fallback chain. Segments are tried left-to-right; the first one resolving to a non-None value wins; otherwise the field is None. Any number of segments is supported.

extract:
  products:
    path: "$.products[*]"
    select:
      id: id
      price: "sale_price || list_price"   # fall back to list_price if sale_price is null
      label: "customLabel || title || ticketCategory"
    collect: true

Reduce (aggregation)

extract:
  earliest_slot:
    path: "$.items[*].from_datetime"
    reduce: min_time          # earliest time-of-day across all items

  latest_slot:
    path: "$.items[*].to_datetime"
    reduce: max_time          # latest time-of-day across all items

  total_price:
    path: "$.items[*].price"
    reduce: sum

  unique_categories:
    path: "$.items[*].category"
    reduce: distinct

Available operators:

Operator	Description
min_time / max_time	Earliest/latest time-of-day — accepts ISO 8601 datetime or plain HH:MM / HH:MM:SS (ignores date + timezone)
min_date / max_date	Earliest/latest calendar date (ignores time)
min_datetime / max_datetime	Earliest/latest full datetime (timezone-normalised)
min_int / max_int	Minimum/maximum numeric value
sum	Sum of numeric values
count	Count of non-null values (returns 0 for empty)
first / last	First or last value in traversal order
concat	Join values as a string (default separator ", ")
distinct	Deduplicated list, preserving first-seen order

For concat with a custom separator use the dict form:

"reduce": {"op": "concat", "sep": " | "}

Fetch from API

from siphon import fetch_and_process

spec = {
    "request": {"path": "/products"},
    "extract": {
        "id": "$.data.id",
        "names": {"path": "$.data.items[*].name", "collect": True},
    },
}

result = fetch_and_process(spec, "https://api.example.com")

Requires requests:

pip install siphon-dsl[http]

Typed Specs (Pydantic)

from siphon.typed import process_spec, ExtractSpec, FieldSpec

spec = ExtractSpec(
    extract={
        "id": "$.data.id",
        "active_items": FieldSpec(
            path="$.data.items[*]",
            where={"status": "active"},
            select={"item_id": "id", "name": "name"},
            collect=True,
        ),
    }
)

result = process_spec(spec, data)

Requires pydantic:

pip install siphon-dsl[typed]

Why Siphon?

• Minimal — ~100 lines of code, no dependencies
• Declarative — specs are data, not code
• Composable — combine paths, filters, and projections

Spec History

See specs/ for version history and full documentation. Latest: v0.11 — paths support [N] list indexing (positive and negative).

License

MIT