📥 Ingest Data

Four shapes — NDJSON bulk, JSON upsert/skip, CSV, and document — all route through IngestClient and all trigger SmartIngest (76 canonical identifier types auto-detected on the way in).

ingest = IngestClient.from_client(client)

NDJSON bulk — the fast path

ingest.bulk("Person", [
    {"id": "alice", "name": "Alice Chen",   "email": "alice.chen@acmecorp.com",
     "phone": "+14155550100", "role": "CFO",      "company": "Acme Corp",   "risk": "HIGH"},
    {"id": "bob",   "name": "Bob Smith",    "email": "bob@shellco.io",
     "phone": "+14155550101", "role": "Director",  "company": "ShellCo Ltd", "risk": "HIGH"},
    {"id": "carla", "name": "Carla Nunez",  "email": "carla.nunez@acmecorp.com",
     "phone": "+34666123456", "role": "Accountant","company": "Acme Corp",   "risk": "LOW"},
    {"id": "david", "name": "David Kim",    "email": "d.kim@offshore.bn",
     "phone": "+822012345678","role": "Nominee",   "company": "Pacific Trust","risk": "MEDIUM"},
])

{"rows_queued": 4, "rows_rejected": 0, "task_id": "itsk_019fe254-3647-...", "connector": "direct", "errors": []}

Under the hood: SmartIngest scanned every field and auto-detected 4 phone numbers (E.164) and 4 emails (RFC 5322). They're now in the IdentityIndex — linkable across sources with no detection code on your side.

The money trail

ingest.bulk("Transaction", [
    {"id": "tx1", "from_account": "Acme Corp",         "to_account": "Pacific Trust 7742",
     "amount": 2300000, "currency": "USD", "date": "2026-01-15", "authorized_by": "Alice Chen"},
    {"id": "tx2", "from_account": "Pacific Trust 7742","to_account": "ShellCo Ltd",
     "amount": 850000,  "currency": "USD", "date": "2026-01-22", "authorized_by": "David Kim"},
    {"id": "tx3", "from_account": "ShellCo Ltd",       "to_account": "CASH",
     "amount": 120000,  "currency": "USD", "date": "2026-02-01", "authorized_by": "Bob Smith"},
    {"id": "tx4", "from_account": "Acme Corp",         "to_account": "Pacific Trust 7742",
     "amount": 500000,  "currency": "USD", "date": "2026-02-10", "authorized_by": "Alice Chen"},
])

Case documents (rich text for BM25)

ingest.bulk("CaseDoc", [
    {"id": "whistleblower", "title": "Whistleblower Complaint by Carla Nunez",
     "body": "I am writing to report suspected embezzlement by CFO Alice Chen. Over three "
             "months, Alice authorized four wire transfers totaling $2.8 million from Acme "
             "Corp to an offshore account at Pacific Trust Bank held by David Kim. The money "
             "was then moved to ShellCo Ltd, directed by Bob Smith."},
    {"id": "sar", "title": "Suspicious Activity Report - Pacific Trust Bank",
     "body": "Account 7742 held by David Kim received $2.8 million from Acme Corp. Funds "
             "were rapidly moved to ShellCo Ltd and partially withdrawn as cash. Pattern "
             "consistent with money laundering layering."},
    {"id": "news", "title": "Acme Corp CFO Under Scrutiny",
     "body": "Federal investigators examine whether CFO Alice Chen orchestrated a $2.8 "
             "million embezzlement through offshore accounts. A whistleblower complaint "
             "triggered the probe. Bob Smith of ShellCo denied involvement."},
])

JSON upsert / skip — conflict resolution

# Re-ingest with on_conflict='upsert' → update existing rows by id.
ingest.bulk("Person", [
    {"id": "alice", "risk": "CRITICAL", "sanctions_status": "under_review"},
], on_conflict="upsert")
 
# 'skip' keeps the existing row untouched if the id already exists.
ingest.bulk("Person", [
    {"id": "alice", "risk": "LOW"},
], on_conflict="skip")  # alice's risk stays CRITICAL

CSV ingest — bulk from a file

csv_text = """id,from_account,to_account,amount,currency,date
tx5,ShellCo Ltd,CASH,80000,USD,2026-02-05
tx6,Acme Corp,Pacific Trust 7742,300000,USD,2026-02-08
"""
ingest.bulk_csv("Transaction", csv_text)
# {"rows_queued": 2, "rows_rejected": 0, ...}

Document ingest — datagrep extractor protocol v1.1.0

POST /ingest/document accepts unstructured documents in the datagrep extractor protocol. It creates one IntelReport row (the manifest) + one IntelChunk row per chunk. The format is strict:

Line 1 of chunks_jsonl MUST be the envelope: {"type":"envelope","schema_version":"1.1.0"}
Each subsequent line is a chunk: {"type":"chunk","id":"ch_...","text":"...","sequence_index":N}
manifest_json MUST include schema_version, source (filename, sha256, media_type), run_config, and stats

# Build a valid dgrep extractor v1.1.0 payload
chunks_jsonl = (
    '{"type":"envelope","schema_version":"1.1.0"}\n'
    '{"type":"chunk","id":"ch_001","text":"SAR filed on Pacific Trust account 7742 held by David Kim.","sequence_index":0,"source_filename":"sar.pdf"}\n'
    '{"type":"chunk","id":"ch_002","text":"Alice Chen authorized 4 transfers totaling $2.8M disguised as consulting fees.","sequence_index":1,"source_filename":"sar.pdf"}'
)
 
manifest_json = json.dumps({
    "manifest_id": "mf_shadow_ledger_001",
    "schema_version": "1.1.0",
    "status": "complete",
    "created_at": "2026-08-08T16:00:00Z",
    "extractor_version": "1.2.0",
    "source": {
        "filename": "sar.pdf",
        "sha256": "aabbccdd1234",
        "media_type": "text/plain",
        "size_bytes": 2048,
        "ingested_at": "2026-08-08T16:00:00Z"
    },
    "run_config": {"parser": "pdf:pymupdf4llm", "ocr_enabled": False},
    "stats": {"total_chunks": 2, "total_tokens": 85, "mean_confidence": 0.95}
})
 
# Send it
r = httpx.post(f"{BASE}/ingest/document",
               json={"chunks_jsonl": chunks_jsonl, "manifest_json": manifest_json,
                     "purpose": "analytics"},
               headers=H, timeout=30)
print(r.json())

{"manifest_id": "mf_shadow_ledger_001", "chunks_ingested": 2, "rows_queued": 3, "warnings": []}

What lands in the store: 1 IntelReport row (the manifest metadata) + 2 IntelChunk rows (one per chunk, with text_body, sequence_index, source_filename, confidence). Both are governed — PURPOSE-tagged, provenance-tracked, ACL-filtered, and bi-temporal. Query them with SELECT * FROM IntelChunk WHERE report_id = '...'.

Full chunk field reference (from /specs)

Every chunk line supports these fields:

Field	Type	Required	Description
`type`	string	yes	Must be `"chunk"`
`id`	string	yes	`ch_` prefix + hex
`text`	string	yes	Plain-text body
`sequence_index`	integer	yes	0-based document order
`section_path`	string[]	no	Heading ancestors
`page_start` / `page_end`	integer	no	1-based page numbers
`token_count` / `char_count`	integer	no	Size metadata
`confidence`	float	no	0.0–1.0 extraction quality
`extraction_method`	string	no	e.g. `pdf:pymupdf4llm`
`source_media_type`	string	no	Original format
`source_filename`	string	no	Source file name
`source_sha256`	string	no	Content hash
`entities`	array	no	`[{entity_type, text, score, start, end}]`
`chunk_metadata_entity_types`	string[]	no	Pre-computed sorted labels

Streaming, CDR, OTLP — the long tail of ingest shapes

Method	Shape	Use it for
`ingest.bulk("T", rows, detect_packs="network,financial")`	NDJSON + detector override	per-call SmartIngest pack selection
`ingest.ingest_iter("T", generator, batch_size=500)`	streaming iterator	O(batch_size) memory for huge CSVs
`ingest.ingest_cdr(rows)`	CSV via `/ingest/cdr`	call-detail records (caller/callee/tower)
`ingest.otlp_traces(payload)` / `otlp_logs(...)` / `otlp_metrics(...)`	OTLP/JSON	OpenTelemetry ingest
`ingest.media_status(task_id)`	poll	multipart media upload progress

# Stream a million rows without holding them all in memory:
def row_gen():
    for i in range(1_000_000):
        yield {"id": f"r{i}", "amount": i}
 
total = ingest.ingest_iter("Transaction", row_gen(), batch_size=1000)
# → 1_000_000

Next: SQL Queries — total up the wires, group them by source, and see the $3.77 M trail.

📥 Ingest Data#

NDJSON bulk — the fast path#

The money trail#

Case documents (rich text for BM25)#

JSON upsert / skip — conflict resolution#

CSV ingest — bulk from a file#

Document ingest — datagrep extractor protocol v1.1.0#