Filter JSON Lines by a typed value

Before you begin

Python 3.12 or newer

What this script changesCreates a new result and refuses overwrite; input is unchanged. Nonstandard NaN/Infinity JSON is rejected; nested documents can exceed the interpreter recursion limit.

Set up your example

Save the complete code as jsonl-filter-field.py in a working folder.
Save the provided sample files in that same folder; run commands from that folder.

Sample files in this download:

types.json
sample.json
people.json
other.json
sample.jsonl
flat.json
sample.xml

Download the ZIP, extract it, and open a terminal inside the extracted folder. The source file is jsonl-filter-field.py.

Run the script

python "jsonl-filter-field.py" --input "sample.jsonl" --field "score" --equals-json 8 --output "filtered.jsonl"

If your system uses python3 instead of python, use that command for Python examples. Keep paths containing spaces in quotes. Required inputs and optional settings are described in the code and setup steps.

What to expect

Keep records matching a top-level field and exact JSON value type. Only Ada's score-8 object remains.

The complete script

#!/usr/bin/env python3
'Filter JSON Lines by a typed value'
import argparse
import json
from pathlib import Path

def write_json(path, value):
    data = json.dumps(value, indent=2, ensure_ascii=False, allow_nan=False) + "\n"
    with Path(path).open("x", encoding="utf-8", newline="") as handle:
        handle.write(data)

import random
from collections import Counter
import xml.etree.ElementTree as ET

def read_text(path):
    if Path(path).stat().st_size > 10*1024*1024:
        raise ValueError("Input exceeds 10 MiB.")
    return Path(path).read_text(encoding="utf-8-sig")
def reject(value):
    raise ValueError("Nonstandard JSON constant: "+value)
def load(path):
    return json.loads(read_text(path),parse_constant=reject)
def lines(path):
    return [json.loads(s,parse_constant=reject) for s in read_text(path).splitlines() if s.strip()]
def array(value):
    if not isinstance(value,list):
        raise ValueError("Expected an array root.")
    return value
def objects(value):
    value=array(value)
    if not all(isinstance(r,dict) for r in value):
        raise ValueError("Every array item must be an object.")
    return value
def walk(value,path=()):
    yield path,value
    if isinstance(value,dict):
        for k,v in value.items(): yield from walk(v,path+(k,))
    elif isinstance(value,list):
        for i,v in enumerate(value): yield from walk(v,path+(i,))
def pointer(value,path):
    if not path: return value
    if not path.startswith("/"): raise ValueError("Use a /-prefixed JSON Pointer.")
    for part in path[1:].split("/"):
        key=part.replace("~1","/").replace("~0","~")
        if isinstance(value,list):
            if not key.isdigit() or int(key)>=len(value): raise ValueError("Bad array index.")
            value=value[int(key)]
        elif isinstance(value,dict): value=value[key]
        else: raise ValueError("Pointer crosses a scalar.")
    return value
def xml(path):
    text=read_text(path)
    if "<!DOCTYPE" in text.upper() or "<!ENTITY" in text.upper():
        raise ValueError("DTD/entity declarations are refused.")
    try: return ET.fromstring(text)
    except ET.ParseError as exc: raise ValueError(str(exc)) from exc
def save(path,text):
    with Path(path).open("x",encoding="utf-8",newline="") as handle: handle.write(text)

def main():
    parser = argparse.ArgumentParser(description='Filter JSON Lines by a typed value')
    parser.add_argument("--input", required=True)
    parser.add_argument("--output", required=True)
    parser.add_argument("--field",required=True)
    parser.add_argument("--equals-json",required=True)
    args = parser.parse_args()
    expected=json.loads(args.equals_json,parse_constant=reject)
    selected=[r for r in objects(lines(args.input)) if args.field in r and type(r[args.field]) is type(expected) and r[args.field]==expected]
    save(args.output,"".join(json.dumps(r,ensure_ascii=False,allow_nan=False)+"\n" for r in selected))

if __name__ == "__main__":
    try:
        main()
    except (OSError, ValueError, KeyError, TypeError, ArithmeticError, IndexError, LookupError) as exc:
        raise SystemExit(str(exc))

Open source as plain text · Download source and sample files

How it works

Local UTF-8 input is limited to 10 MiB and held in memory.
The comparison distinguishes numeric 8 from the text 8 and boolean true from integer 1.
Only Ada's score-8 object remains.

Make it your own

--equals-json

Use --equals-json 5 to select Bo.

Adapt the processing

Edit main's selection condition to call pointer for nested-field matching.

Change one setting at a time, run it again on sample data, and compare the output. The customization guide walks through this process.

If something goes wrong

No records match.

Check the field's JSON type as well as its value.

The destination already exists.

Choose a new --output path. The script deliberately refuses to replace an existing result.

Reference & next steps

These official references explain the underlying tools. The Scriptage example and guide are original instructional content. See verification notes for what has been checked and the limits of those checks.

Copies a template for your own notes. It does not send anything.