Metadata-Version: 2.4
Name: pycrowley
Version: 0.2.0
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Operating System :: MacOS
Classifier: Operating System :: POSIX :: Linux
Classifier: Programming Language :: Rust
Classifier: Programming Language :: Python :: 3 :: Only
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: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: Implementation :: CPython
Classifier: Topic :: File Formats :: JSON
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: Text Processing
Classifier: Typing :: Typed
Requires-Dist: numpy ; extra == 'dataframe'
Requires-Dist: pandas ; extra == 'dataframe'
Requires-Dist: fsspec ; extra == 'remote'
Provides-Extra: dataframe
Provides-Extra: remote
License-File: LICENSE.md
License-File: THIRD-PARTY-NOTICES.md
Summary: High-throughput streaming JSON query engine
Keywords: json,ndjson,jsonl,streaming,query,simd,ijson,parser,dataframe
Author-email: "Nicolas R. Posner" <nicolasposner@gmail.com>, Micah Kepe <micahkepe@gmail.com>
License-Expression: MIT
Requires-Python: >=3.9
Description-Content-Type: text/markdown; charset=UTF-8; variant=GFM

# `crowley`: ripgrep for json

`crowley` is a tool for querying, tabulating, and profiling streaming JSON 
with O(buffer) memory, whatever the input size. You can:
 - Answer many queries in a single pass
 - Gather multiple outputs (value streams, aggregations, summaries) per query
 - Turn arbitrarily nested, schemaless JSON into the tabular form of your
   choice bounded only by your output size
 - Query many heterogenous data sources in parallel
 - Run the same tool on a local machine or a cloud instance, on a fraction
   of a core or on many cores, on local, streamed, or compressed data, on
   boundedly-sized data or continuous web socket output.
 - All with minimal memory usage and at 2+ GB/s on a single core: see 
   detailed benchmarks in [Performance](#performance).

Currently supports NEON, AVX-2, and AVX-512 instruction sets. While capable of
falling back to a fully scalar implementation in the absence of supported
instruction sets, it is considerably slower and not recommended for
production use.

## Installation

```bash
pip install pycrowley
```

The distribution is named `pycrowley` (PyPI reserves the bare name), but
the package you import is plain `crowley`:

```python
import crowley
```

Optional extras: `pip install 'pycrowley[remote]'` adds `fsspec` for
`s3://`/`https://…` URL sources (add the scheme's backend too, e.g.
`s3fs`); `'pycrowley[dataframe]'` adds numpy + pandas for `to_pandas()`
and real numeric dtypes. `to_polars()`/`to_arrow()`/`write_parquet()` use
`polars`/`pyarrow` if you have them installed; nothing else is required.

Wheels cover CPython ≥ 3.9 on macOS (arm64, x86_64) and Linux (x86_64,
aarch64, manylinux2014). No Windows wheel yet.

## Quick Start

```python
import crowley

# iterate lazily (à la ijson)
for event in crowley.items("events.json", "events[*]"):
    ...     # your logic goes here 


# extract a multi-column table in a single pass 
# works with arbitrarily-nested JSON, not just NDJSON!
df = crowley.table(
    "events.json", rows="events[*]",
    columns={"id": "id", "repo": "repo.name", "at": "created_at"},
).to_pandas()


# set up a store elsewhere, e.g. a list
unique_values = []
# append streamed values into the list *after* the run completes
crowley.Query("events.json", 
              "events[*].repo.name", 
              crowley.ValuesUnique(into=unique_values))

# or, set numeric=True to coerce matched values to ``float`` and append
# a single NumericBuffer to ``into`` instead. 
import numpy as np
out = []
crowley.Query("events.json",
              "events[*].created_at",
              crowley.Values(into=out, numeric=True))
timestamps = np.frombuffer(out[0].buffer, dtype=np.float64)
# if ``into`` is omitted from Values or ValuesUnique, they are equivalent
# to ``Count`` and ``CountUnique``, respectively


# run a query over an unbounded socket+pipe stream, progressively printing 
# the count of matching elements until a condition is met
def cb(bytes_scanned: int, results: list[crowley.QueryResult]):
    region_dict = results[0].value_counts()
    print(region_dict)      # e.g. { europe: 123, asia: 456,  ... }
    return region_dict.get("north_america", 0) > 1000
    # when we reach a certain condition, terminate gracefully
    # by returning a truthy value
    # otherwise, it will run until the input fails or the process aborts

sock = socket.create_connection(...)
# take the reader and make into a file-like object
q = crowley.Query(
    crowley.Source(sock.makefile("rb"), progress=cb, compression="none"), 
    "requests[*].region", 
    crowley.ValueCounts())
# final tally after scan finishes
q.result(0, 0).value_counts()


# make multiple queries and output multiple statistics in a single scan
q = crowley.Query("events.json", {
    "events[*].type": [crowley.ValueCounts()],
    "events[*].payload.size": [crowley.Sum(), crowley.Max()],
})
# grab the result of (source 0, query 0)
# and extract the value_counts result output (a dict)
result_dict = q.result(0, 0).value_counts()
# grab the query by name rather than index,
# and extract the sum result output (a numeric)
result_sum = q.result(0, "events[*].payload.size").sum()


# perform a query in parallel across many sources
# you can pass a glob, a list, a list with globs, a mixed
# list of files, globs, urls, inline JSON, and more!
q = crowley.Query("data/bug_reports/*.json",
                  "reports[*].severity",
                  crowley.ValueCounts())
# view the individual sources and results after globbing and deduplication
for s in q.sources: print(s, q.result(s, 0).value_counts()) 
# get the result from a particular source by name
q.result("data/bug_reports/july_2025.json", 0).value_counts()
```

## Sources

Every entry point takes the same source forms:

- **A path**: `"data.json"`. Will stream-decode natively when the extension
  is `.gz`, `.zst`, or `.zstd`.
- **A glob**: `"logs/*.json"`, expanded with shell semantics and deduplicated
  against literal paths. NOTE: `*` doesn't match hidden dot-files.
- **A URL**: `"s3://bucket/key.json.gz"`, `"https://…"` opened through
  `fsspec` (optional dependency, plus the scheme's backend such as `s3fs`).
  As with paths, compression is inferred from the extension.
- **A file-like object**: anything with `read(size)`, such as an open file, 
  an HTTP response body, a boto3 `StreamingBody`, or an fsspec file. `crowley`
  automatically handles closing file-like objects *which it opened*. Any 
  file-like object which the user opened and passed to `crowley` is **not** 
  closed automatically, and must be handled by the caller.
- **Inline JSON**: a `str` beginning with `{`/`[`, or `bytes`/`bytearray`.
- **A configured source**: `crowley.Source(src, ...)` wraps any of the above
  to attach options that belong to one source rather than to the run:
  - `progress=` / `progress_interval=`: a callback for this one source.
    See [progress](#aggregation-query) below.
  - `name=`: a label for `result(name, ...)`, equivalent to the
    `{name: source}` dict form.
  - `compression="gzip" | "zstd" | "none"`: overrides extension inference,
    necessary to decode a compressed *file-like object*, which has
    no extension to infer from. Wrapping a glob applies the override to 
    every match.

A `Source` carrying a callback or a name must resolve to exactly one
file: a glob fanning out under one name would make the name ambiguous, and
one callback across several files would make its byte counts meaningless.

`Query` accepts a list of many sources of mixed types, scanned in parallel 
on a worker pool, each with its own results. Unbounded streams 
(sockets, pipes) mix in like any other source. Memory use remains bounded, 
and each stream can carry its own `progress=` callback to watch it and stop 
it (though accumulated results constructed from the input will grow unless 
stored elsewhere).

**NDJSON** and **JSONL** do not need separate modes: the scanner reads
concatenated root values natively. You may query with paths relative to each 
root, or anchor a table with `rows=""` (the empty query, matching each root 
value). NOTE: Scalar values at root (outside an array) are not supported.

## The query language

Queries are regular expressions over paths:

| Operator     | Example              | Meaning                                  |
|--------------|----------------------|------------------------------------------|
| Sequence     | `foo.bar.baz`        | field `foo`, then `bar`, then `baz`      |
| Field        | `foo` / `"foo bar"`  | exact field (quote for spaces/metachars) |
| Array index  | `[0]`, `[1:3]`, `[3:]` | index or slice (exclusive end)         |
| Wildcards    | `*` / `[*]`          | any one field / any one index            |
| Disjunction  | `foo \| bar`         | match either `foo` or `bar`              |
| Optional     | `foo?.bar`           | with or without the `foo` step           |
| Repetition   | `foo*`               | zero or more repeats of the step         |
| Grouping     | `foo.(bar\|baz).qux` | nest freely                              |
| Kleene star  | `**`                 | Match zero or more field accesses        |

Recursive descent can be constructed from the algebra: `(* | [*])*.name` matches 
`name` at any depth. The empty string `""` is the empty path, the root value
itself (each root, for NDJSON).

Quoted fields use JSON escapes, but a field name containing a character JSON
itself must escape (e.g. a double quote, backslash, or control character) is
rejected at parse: the engine matches keys by their raw document bytes,
where such a character never appears unescaped, so the query cannot match. 

Reach such keys structurally instead: a `*` wildcard, or `kvitems` on the 
enclosing object (whose yielded keys are properly unescaped).

## Iteration: `items` and `kvitems`

```python
for user in crowley.items( # lazily read a single source
    "data.json", 
    "users[*]", 
    accumulators=[crowley.ValuesUnique(), crowley.Types()]): # carry accumulators along
    ...
for key, val in crowley.kvitems("data.json", "config"):    # members of an object
    ...
```

Values are yielded as the source is read; abandoning the iterator abandons
the scan. `limit=n` stops after n matches per query and stops reading.
Multiple queries interleave through one iterator, and can be distinguished using
`with_index=True` to get `(query_index, value)`. `it.stop(query_idx)` silences
one query mid-flight (already-buffered values still drain).

The scan ends only when all queries are stopped or satisfied.

In addition to this lazy iteration, you may use `accumulators=[...]` to
also use `crowley`'s regular accumulators, and read them with `it.results()`.

Note that if `it.results()` is called before exhaustion **the scan ends** and
the results up to that point are reported alongside `bytes_covered`.

## Aggregation: `Query`

```python
q = crowley.Query(sources, queries, accumulators, limit=None, mode="auto",
                  progress=None, progress_interval=None)
q.result(source, query)           # -> QueryResult; index or name, either slot
```

`queries` is a string, a list (sharing one `accumulators` list), or a
`{query: [accumulators]}` dict. `sources` may be a `{name: source}` dict;
`result()` then takes the name (unnamed paths answer to their own spelling,
and any query answers to its text). `q.sources` / `q.queries` /
`q.accumulators` list the valid addresses, and a wrong name raises
`KeyError` naming them.

Accumulators, each read back from `QueryResult` via its accessor (an
accessor whose accumulator wasn't requested returns `None`):

| Accumulator | Read back with | Result |
|---|---|---|
| `Count()` | `.count()` | number of matched values |
| `CountUnique()` | `.count_unique()` | number of *distinct* matched values |
| `ValueCounts()` | `.value_counts()` | `{value: count}` dict |
| `Exists()` | `.exists()` | whether anything matched at all |
| `Min()` | `.min()` | minimum of numeric matches |
| `Max()` | `.max()` | maximum of numeric matches |
| `Sum()` | `.sum()` | sum of numeric matches |
| `Mean()` | `.mean()` | mean of numeric matches |
| `Product()` | `.product()` | product of numeric matches |
| `Mode()` | `.mode()` | most frequent value(s) as `[value, count]` pairs, ties included |
| `Types()` | `.types()` | set of JSON type names seen at matched positions |
| `Values(into=…)` | your `into` list + `.count()` | every matched value, streamed |
| `ValuesUnique(into=…)` | your `into` list + `.count_unique()` | distinct matched values, streamed |

On the streaming pair, `numeric=True` fills a zero-copy float64
`NumericBuffer` instead of Python objects: use
`numpy.frombuffer(buffer, dtype=np.float64)` to bring it into Python
efficiently.

- `limit=n` bounds matches **per query**. When all queries are satisfied, the 
  scan stops and `bytes_covered` records how far it got.
- `progress=cb` takes any callable and calls it as
  `cb(bytes_scanned, [QueryResult, …])` roughly every `progress_interval` bytes
  (default 8MB; it can't fire more often than once per internal scan buffer, 
  1–4MB). Return True to stop the scan gracefully (results so far become 
  final, with `bytes_covered` recording how far it got); raise to abort it. 
  This allows us to deal with unbounded streams, and also allows progress bars 
  to hook in. See the example with tqdm below:

```python
from tqdm import tqdm

with tqdm(total=os.path.getsize(path), unit="B", unit_scale=True) as bar:
  def cb(bytes_scanned, results):
      bar.n = bytes_scanned
      bar.refresh()
  q = crowley.Query(path, "events[*].id", [crowley.Count()], progress=cb)
```
- **Multiple sources and callbacks**: when using multiple sources, attach each 
  callback to its source with `crowley.Source(src, progress=cb)`. The run-level
  `progress=` kwarg can only take a single source because a single callback over
  interleaved sources would have no meaningful byte count. A truthy return
  stops *only that source* and finalizes its results while other sources keep
  scanning, while an exception aborts the whole run. Callbacks run one at a
  time on the constructor thread in no defined order. 

```python
# watch two unbounded streams while a backfill file scans alongside
q = crowley.Query(
    [
        crowley.Source(us_sock.makefile("rb"), name="us", progress=us_cb),
        crowley.Source(eu_sock.makefile("rb"), name="eu", progress=eu_cb,
                    progress_interval=1_000_000),
        "backfill.ndjson"
    ],
    "requests[*].region", [crowley.ValueCounts()])
q.result("us", 0).value_counts()      # final tally per stream
```

- Scans release the GIL and are Ctrl-C interruptable to within ~50ms.

## Tables

```python
# create a table with the declared columns from a JSON stream
t = crowley.table(
    "events.ndjson", rows="",
    columns={"id": "id", "repo": "repo.name", "stars": "payload.stars",
             "at": "created_at"},
    # optional type declaration: nullable int32 and non-nullable timestamp
    dtypes={"stars": "int32?", "at": "timestamp"},
)
# turn a table into a dataframe...
t.to_pandas()
t.to_polars()
# or an in-memory pyarrow.Table (write_parquet writes parquet files directly)...
t.to_arrow()
# or a dictionary...
t.to_dict()
# or extract one column (numeric columns come back as NumPy arrays,
# others as plain lists)
t.column("id")
```

`rows` is the path to the container that contains a single row of data; 
`columns` maps output names to paths **relative to** the row (or is a list 
of paths used as names). Anchoring with `rows` ensures that a record missing
a field gets a null *in its own row*, so column lengths remain consistent and
the position of missing data isn't misattributed. A column matching more than 
once per row raises a `ValueError` unless the 
`on_multi="first" | "last" | "list"` kwarg is set. Fields which are not 
requested are never materialized.

Unless declared explicitly, data types are inferred per column 
(`int64`, `float64`, `bool`, `str`, nullable variants, `object` for a mix):
`int8/16/32/64`, `float32/64`, `bool`, `str`, `timestamp`, `uuid` (yielding
real `uuid.UUID`s), `decimal(p,s)`. Where a type is declared, `crowley` will 
throw an error rather than coerce. A *missing* field is a null under any 
declaration, but an explicit JSON `null` is refused by a plain declaration; 
append `?` (`"int32?"`, `"decimal(10,2)?"` etc) to declare the column 
nullable and accept explicit nulls as nulls. `to_arrow()` exports over the 
Arrow C Data Interface with zero copies for numeric columns (`crowley` 
itself carries no arrow dependency, though consuming the result requires 
`pyarrow` installed).

For sources bigger than memory, stream chunks instead of holding the table:

```python
for chunk in crowley.table_batches(src, rows="", columns=cols, batch_rows=65536):
    consume(chunk)        # each chunk is a Table; nothing accumulates
```

Or write straight to disk through pyarrow's writers (installed separately):

```python
crowley.write_parquet(src, "out.parquet", rows="", columns=cols, dtypes=d)
crowley.write_csv(src, "out.csv", rows="", columns=cols)
```

Both stream record batches, so peak memory is determined by batch size,
not overall file/stream size.

## Reshaping: `write_ndjson` and `ndjson_lines`

```python
# normalize: nested JSON in, NDJSON out
crowley.write_ndjson("dump.json", "events.ndjson", rows="events[*]")

# slim + compress: keep desired fields and debloat
crowley.write_ndjson("dump.json", "slim.ndjson.zst", rows="events[*]",
                     fields={"id": "id", "repo": "repo.name", "at": "created_at"})

# sink-agnostic: lazy bytes lines for S3 multipart, sockets, anything
for line in crowley.ndjson_lines("dump.json", "events[*]", fields=["id"]):
    upload(line)
```

One line per `rows`-matched record, streaming, memory bounded regardless of
source size. Without `fields=`, each record re-emits whole, preserving
member order. With `fields=`, each record becomes a flat object: missing fields
are omitted, explicit nulls are kept. `on_multi`/`limit` behave the same as
with tables. Output paths compress by extension (`.gz`, `.zst`, `.zstd`).

With the exception of newlines, the re-emitted values are byte-faithful to the 
original. Compare to `jq`, which routes a 20-digit id through a double,
corrupting it, while `crowley` preserves it exactly.

Benchmarking re-emission on the 502MB nested JSON corpus:

| Tool | Runtime (s) | Peak RSS (MB) |
|---|---:|---:|
| `jq -c` | 15.1 | 2300 |
| `ijson` (hand-rolled loop) | 4.4 | 40 |
| **`crowley`** | **0.65** | **34** |

`crowley` completes the same re-emission task 23× faster than `jq` at
1/66th the memory, and completes a projection task at 0.35s/102MB.

## Streaming group-by: `group_by`

```python
t = crowley.group_by(
    "events.ndjson", rows="",
    by={"repo": "repo.name"},                 # 1+ key paths (str/list/dict)
    agg={
        "events": "count",                    # rows per group; no path
        "stars":  ("sum", "payload.stars"),
        "actors": ("count_unique", "actor.login"),
        "seen":   ("first", "created_at"),
    },
)   # -> Table: key columns + agg columns, one row per group
```

SQL's `GROUP BY` over the streamed rows in one pass: `rows` anchors what a
record is (exactly as for tables), keys and agg paths are relative to it.

Supported operations: `count`, `sum`, `min`, `max`, `mean`, `count_unique`, `first`,
`last`. 

Groups come out in first-seen document order; a record missing a key field 
lands in the null group rather than being dropped. Missing and null values 
are skipped by every aggregate (as SQL), but a non-null non-number under a 
numeric op is an error. All-integer columns use `int64`, with overflow
errors rather than turning into a float. Even a single fractional value 
promotes the entire column to `float64`.

The result is an ordinary `Table`, with the same `to_pandas()`/`to_polars()`/
`to_arrow()`/`to_dict()` methods available. Peak RSS is defined by O(groups),
not O(rows). To guard against excessive memory use, `max_groups` defaults to
1 million, raising an error if this is exceeded. When large group counts are
intended, raise this value or pass `None`. `count_unique` additionally holds
each group's distinct values.

## Profiling: `describe`

```python
print(crowley.describe("unknown.json", limit_bytes=50_000_000))
```

Profile structures without passing a query. This returns every path, how many 
values, which JSON types, approximate distinct counts (`None` past the  
`distinct_cap` kwarg, which tells us there's more than we care to count), and 
nesting depth. A quick pass to examine or confirm the structure of a file.

## `ijson` compatibility

```python
import crowley.ijson as ijson
for record in ijson.items(f, "item"):      # ijson prefix syntax, unchanged
    ...
```

For ease of migration from `ijson`, we provide a drop-in replacement for 
`ijson`'s `items` and `kvitems` methods. `ijson` prefix strings are translated 
into their equivalent `crowley` query, including `item` for array elements and 
`""` for the root.

Numbers arrive as `Decimal` by default, as `ijson`'s do. To
restore floats, pass `use_float=True`.

If you use `ijson.parse()` instead, explore the native `crowley.Query` API 
above as a more efficient and expressive way to perform the same work.

## Errors and partial results

Failures return a descriptive type, from which results can be salvaged:

- `crowley.Incomplete`: the source ended before the document did
  (truncation, a dropped connection). `.recover()` returns the per-query
  results for the bytes that arrived. (A Ctrl-C is not an `Incomplete`; it
  surfaces as an ordinary `KeyboardInterrupt`.)
- `crowley.Corrupted`: the JSON is not valid. `.recover_corrupted()` returns the 
  per-query results accumulated before failure. This output is NOT guaranteed 
  to be correct.

Both carry a `bytes_covered` property describing how far they got before 
failing. A `QueryResult` salvaged from either error has the `is_partial`
property set to `True`.

Note that a scan that stopped early on purpose (such as by satisfying all 
`limit`s) is not partial, as nothing failed.

## Ordering: completion order

Streamed values (`items`, `kvitems`, `Values(into=…)`) arrive in
**completion order**: a container is yielded at its closing brace, and with
multiple queries, values are interleaved, grouped by scan buffer. Any single
query's stream is in document order end to end; cross-query arrival order is
not meaningful. 

Use `with_index=True` rather than zipping streams positionally (or use `table`, 
whose cells land by row identity). Accumulators and tables are order-insensitive
and unaffected. This is necessary to perform multi-query scans in a single pass 
with bounded memory.

## Performance

While `crowley` can be used in a wide variety of environments for many tasks,
it excels in running multi-GB scans with O(buffer) peak memory use.

**Throughput.** On a 502MB GHArchive corpus (M3 Max, warm cache), a
single-query scan runs at **3.5 GB/s** in the default two-thread pipelined
mode and **2.6 GB/s** on one thread (`mode="batched"`).

**Small machine meets big file.** Extracting a five-column table out of a 
502MB NDJSON file, running on AWS Lambda (128MB, arm64, Python 3.13, 
us-east-1) taking the best of 4 warm runs from each tool. All tools returned 
identical results (when they succeeded).

| tool | 128MB Lambda | peak memory | $ / 1k runs |
|---|---|---|---|
| **`crowley`** | **29.3s** | **102MB** | **$0.049** |
| msgspec (hand-chunked) | 43.5s | 102MB | $0.072 |
| polars | ✗ OOM | — | $0.069 (at 512MB) |
| pyarrow | ✗ OOM | — | $0.110 (at 512MB) |
| pandas | ✗ OOM | — | ✗ OOM at 512MB too |
| ijson | ✗ timeout (600s) | — | $0.197 (at 512MB) |

At this scale, whole-file engines simply fail. `ijson`, our baseline comparison
for streaming engines, can handle the space constraints but times out. The 
only other tool that finishes, `msgspec`, needs to have its input split into
chunks by hand, and `crowley` outcompetes it 1.5x in both time and cost.

This is on NDJSON. On the equivalent *nested* document, `crowley`'s numbers
are unchanged — while hand-chunking (msgspec's lifeline) no longer works
without line framing, and of the rest only `ijson` can process a nested
document in bounded memory at all.

`crowley` is outcompeted by `polars` when RAM and cores are abundant. On a
26MB NDJSON file at 1769MB of Lambda memory, `polars` finishes twice as quickly
as `crowley`. `crowley` is preferred when data originates in nested form,
input size is large or unbounded, and memory/CPU is constrained.

## Execution modes and hardware

The `mode=` kwarg defaults to `"auto"`, which detects local hardware and
chooses the execution mode for highest throughput. It reads cgroup quotas as 
well as Lambda memory size, so cloud containers get sensible defaults without
explicit configuration.

You may also explicitly set the execution mode to `"pipelined"`, which uses
two threads per source for maximum throughput, or `"batched"`, which uses a
single thread. `"batched"` is superior when fewer than 2 cores are available, 
including for sub-vCPU deployments, or when you wish to optimize for cost.

The scanner's SIMD kernel is chosen at import: NEON on aarch64, AVX-512/AVX2
on x86-64. The active kernel can be examined at runtime with
`crowley._active_kernel()`.

If no supported instruction set is detected, `crowley` will raise a warning and
proceed with a scalar fallback.

SVE2 and SSE4.2 kernels are still in development.

