Export nested customer order summaries

Before you begin

Python 3.12 or later.
This report expects the sample customers/products/orders/order_items schema; adapt its SELECT for another schema.

What this script changesReads source with mode=ro and creates one new JSON file, refusing overwrite. Exports names and email addresses; review privacy before sharing.

Set up your example

Save as sqlite-customer-order-json.py.
Open a terminal in its folder.
Run python "sqlite-customer-order-json.py" --help and prepare the listed input files.
Save the supplied sample.sql and create_sample_db.py together in the same folder.
For the included educational fixture, run python create_sample_db.py --output sample.db. It refuses to overwrite an existing file.
The fixture intentionally contains duplicate customer emails, one NULL email, and one orphan order so audit examples have visible findings.

Sample files in this download:

sample.sql
create_sample_db.py

Download the ZIP, extract it, and open a terminal inside the extracted folder. The source file is sqlite-customer-order-json.py.

Run the script

python "sqlite-customer-order-json.py" --output customer-orders.json

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

Ada has two nested orders; Dee has an empty orders array.

The complete script

#!/usr/bin/env python3
"""Create a new nested JSON document grouping each customer with their sample-schema orders."""
import argparse
import sys
import json
from pathlib import Path
import sqlite3
import time
from contextlib import closing

def identifier(value):
    return '"'+value.replace('"','""')+'"'

def connect(path):
    path=path.resolve()
    if not path.is_file():raise ValueError('Database must already exist')
    connection=sqlite3.connect(path.as_uri()+'?mode=ro',uri=True,timeout=5)
    connection.row_factory=sqlite3.Row
    connection.execute('PRAGMA query_only=ON')
    deadline=time.monotonic()+20
    connection.set_progress_handler(lambda: int(time.monotonic()>deadline),10000)
    return connection

def table(connection,name):
    if not connection.execute("SELECT 1 FROM sqlite_schema WHERE type='table' AND name=?",(name,)).fetchone():raise ValueError('Unknown table: '+name)
    return identifier(name)

def column(connection,name,field):
    names=[row['name'] for row in connection.execute('PRAGMA table_info('+table(connection,name)+')')]
    if field not in names:raise ValueError('Unknown column: '+field)
    return identifier(field)

def json_default(value):
    if isinstance(value,bytes):return {'blob_hex':value.hex()}
    raise TypeError(type(value).__name__)

def emit(rows):
    print(json.dumps([dict(row) for row in rows],indent=2,default=json_default))

def main():
    parser = argparse.ArgumentParser(description='Create a new nested JSON document grouping each customer with their sample-schema orders.')
    parser.add_argument("--database",type=Path,default=Path("sample.db"))
    parser.add_argument('--output',type=Path,default=Path('customer-orders.json'))
    args = parser.parse_args()
    try:
        with closing(connect(args.database)) as connection:
            customers=[dict(r) for r in connection.execute('SELECT id,name,email,region FROM customers ORDER BY id LIMIT 1000')]
            for customer in customers:
                customer['orders']=[dict(r) for r in connection.execute('SELECT id,ordered_at,total,status FROM orders WHERE customer_id=? ORDER BY ordered_at,id LIMIT 1000',(customer['id'],))]
            with args.output.open('x',encoding='utf-8') as out:json.dump(customers,out,indent=2)
            print(f'Created {args.output}; limited to 1000 customers and 1000 orders per customer.')
    except sqlite3.Error as exc:
        raise RuntimeError(str(exc))

if __name__ == '__main__':
    try:
        main()
    except (OSError, ValueError, RuntimeError) as exc:
        print(f'Error: {exc}', file=sys.stderr)
        sys.exit(1)

Open source as plain text · Download source and sample files

How it works

Create a new nested JSON document grouping each customer with their sample-schema orders.
Arguments are explicit and expected input errors produce a message with a nonzero exit code.
Results reflect your selected local data or the specific endpoint you choose; the example illustrates the result shape.

Make it your own

Adapt the task

Change the explicit customer and order SELECT fields for your schema; select a fresh --output each time.

Adjust presentation

Edit final print statements or json.dumps indentation to adjust the report layout. Keep input validation and exclusive-create file modes intact.

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

An input or dependency is unavailable.

This report expects the sample customers/products/orders/order_items schema; adapt its SELECT for another schema.

Actual results differ from the example.

Inputs and system state determine values. Review each finding in context; a report is not proof of system health or a complete security assessment.

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.