OpenViSTAI · Agents · Blockchain · Crypto
Menu
OPENVIST RECIPES

Count JSON records by an explicit string field with Python

Python

Group a small JSON array of objects by one selected top-level text field and report sorted group counts without modifying the input.

Python code excerpt: Count JSON records by an explicit string field with Python. Version 0.1.0. Read the full recipe for code, sources and recorded test scope.

Group a small array of JSON objects by one explicitly selected top-level string field. Read JSON from standard input and report sorted group counts. Strings are compared exactly, including whitespace, case and Unicode representation. Successful output includes group values.

Before you start

  • Save as count_records.py and run: python3.12 count_records.py FIELD < input.json. The program reads only stdin; shell redirection supplies the input.
  • Provide exactly one nonempty, literal top-level field name. Quote names containing whitespace. Use -- before a name beginning with a hyphen, for example: python3.12 count_records.py -- --category < input.json. Dots and slashes have no path meaning. There is no help option.
  • Uses only argparse, json, math and sys. Reads at most 65537 bytes and rejects input exceeding 65536 bytes. Decoding uses strict UTF-8.
  • Requires a JSON array of objects. Rejects duplicate object keys at every depth, non-finite constants, floats that overflow to infinity, and selected fields that are missing or are not nonempty strings.
  • Preserves selected strings without trimming, case folding or Unicode normalization. Outputs one JSON object containing records and groups, with distinct groups sorted by Python string ordering. An empty array produces zero records and an empty groups array.
  • Successful processing exits 0 without stderr. Invalid arguments or input exit 2 with an error on stderr and no stdout or traceback. Error messages do not include record values. Input is not modified.

Build and check

  1. Save the example as count_records.py.
  2. Prepare a UTF-8 JSON array of objects, at most 65536 bytes.
  3. Run: python3 count_records.py status < records.json
  4. Inspect records and the sorted groups array. Each value appears once with its occurrence count.
  5. Handle exit code 2 as invalid or unreadable input.

Limits to understand

  • All records are parsed in memory. Stdin can block until EOF, unless the read limit is reached first.
  • Python parser limits apply, including nesting depth and integer-token length; inputs exceeding those limits are rejected.
  • Successful output intentionally includes the selected group values and can contain sensitive data. JSON escaping changes representation, not the string values.
  • Counts describe only the supplied data. This is not a database query engine.
  • Output-write failures are outside the input-error guarantee.
  • The entire input and parsed values are held in memory. Memory use can exceed the input byte size.
  • Standard input may block while waiting for data or EOF. The byte limit is not a timeout.
  • Output write failures are outside the input-validation error guarantee.
  • Selected group values can contain sensitive information; inspect the output before sharing it.

Run it

python3 count_records.py status < records.json

Example input

[{"status":"open"},{"status":"closed"},{"status":"open"}]

Expected output

{"records":3,"groups":[{"value":"closed","count":1},{"value":"open","count":2}]}
INSPECT THE SCRIPT

Inspect, copy and run the documented checks.

This browser preview builds code locally. It does not contact an API or run the code.

Default example ready.

Python example
#!/usr/bin/env python3
import argparse
import json
import math
import sys


class InputError(Exception):
    pass


class Parser(argparse.ArgumentParser):
    def error(self, message):
        raise InputError("provide exactly one nonempty field name")


def unique_object(pairs):
    result = {}
    for key, value in pairs:
        if key in result:
            raise InputError("duplicate object key")
        result[key] = value
    return result


def reject_constant(token):
    raise InputError("non-finite numeric constant")


def finite_float(token):
    value = float(token)
    if not math.isfinite(value):
        raise InputError("numeric token exceeds the finite float range")
    return value


def prepare_output():
    parser = Parser(add_help=False, allow_abbrev=False)
    parser.add_argument("field")
    args = parser.parse_args()
    if args.field == "":
        raise InputError("field name must not be empty")

    raw = sys.stdin.buffer.read(65537)
    if len(raw) > 65536:
        raise InputError("input exceeds 65536 bytes")
    text = raw.decode("utf-8", errors="strict")
    data = json.loads(
        text,
        object_pairs_hook=unique_object,
        parse_constant=reject_constant,
        parse_float=finite_float,
    )
    if not isinstance(data, list):
        raise InputError("top-level JSON value must be an array")

    counts = {}
    for record in data:
        if not isinstance(record, dict):
            raise InputError("every array element must be an object")
        if args.field not in record:
            raise InputError("a record is missing the selected field")
        value = record[args.field]
        if not isinstance(value, str) or value == "":
            raise InputError("every selected field must be a nonempty string")
        counts[value] = counts.get(value, 0) + 1

    result = {
        "records": len(data),
        "groups": [
            {"value": value, "count": counts[value]}
            for value in sorted(counts)
        ],
    }
    return json.dumps(result, ensure_ascii=True, allow_nan=False) + "\n"


def main():
    try:
        output = prepare_output()
    except InputError as exc:
        sys.stderr.write("error: " + str(exc) + "\n")
        return 2
    except (ValueError, OverflowError, RecursionError):
        sys.stderr.write("error: invalid JSON, UTF-8, or parser limit exceeded\n")
        return 2
    except OSError:
        sys.stderr.write("error: unable to read stdin\n")
        return 2
    except MemoryError:
        sys.stderr.write("error: insufficient memory to process input\n")
        return 2

    sys.stdout.write(output)
    sys.stdout.flush()
    return 0


if __name__ == "__main__":
    sys.exit(main())
What to verify after running
  • Six independent CLI fixtures passed in offline CPython 3.12.15 Docker.
  • Checked grouping counts, empty arrays, exact Unicode and case behavior, missing fields, non-string fields and duplicate keys.
  • Nine additional unittest methods passed in the same pinned image, covering byte boundaries, UTF-8 and JSON errors, non-finite numbers, nested duplicate keys, native parser limits, arguments, literal field names, invalid record types and exact string preservation.

Request or format preview

Illustrative preview. No provider response, price or balance is fetched.

Change a setting to inspect the preview.

Official sources and related profiles

    Code license: MIT. Keep the license notice with your copy.

    Example version 0.1.0 · Documentation reviewed 2026-10-14.

    Six independent CLI fixtures and nine additional boundary-test methods passed in pinned, offline CPython 3.12.15 Docker. These checks do not establish exhaustive correctness, performance for arbitrary inputs, or behavior on other Python versions.