RelataDB SDK Cookbook#

pip install relata-sdk httpx
relata serve  # starts on localhost:9090

from relata import (
    RelataClient, IngestClient, Memory, McpClient,
    AuditClient, Namespace, QueryResult,
)

client = RelataClient(
    "http://localhost:9090",
    bearer_token="perftoken",
    purpose="analytics",
)

client.health()
# HealthResponse(status="ok", profile="free", node_id="node-7f3a...")
 
client.status()
# StatusResponse(profile="free", role="coordinator",
#                query_quota=QueryQuota(cost_remaining=1000000, ...))
 
client.ready()
# ReadyReport(is_ready=True, checks={
#   "storage": True, "wal": True, "ingest_queue": True, ...})
 
client.version()
# VersionInfo(version="2.0.0", commit="a1b2c3d", build_profile="release")
 
client.stats()
# Stats(records=11, states=11, snapshot_rows=11, log_leaves=4, tokens=0)
 
client.list_modules()
# {"modules": ["aml", "maritime", "sanctions", "sigma"]}

client.register_type(
    "Person",
    description="A natural person under investigation",
    owner="fraud-team",
    properties={
        "id":      {"type": "text"},
        "name":    {"type": "text"},
        "email":   {"type": "text"},
        "phone":   {"type": "text"},
        "role":    {"type": "text"},
        "company": {"type": "text"},
        "risk":    {"type": "text"},
    },
)
# {"created": True, "name": "Person"}

client.list_types()
# {"types": [{"name": "Person", "rows": 0}, {"name": "Transaction", "rows": 0}, ...]}
 
client.type_detail("Person")
# {"name": "Person", "owner": "fraud-team", "rows": 4,
#  "properties": {"id": {...}, "name": {...}, ...}}

# Add a `sanctions_status` column to every existing Person row.
client.schema_alter("Person", "add", "sanctions_status", col_type="text")
 
# Rename `risk` → `risk_band`.
client.schema_alter("Person", "rename", "risk", new_column="risk_band")
 
# Retype a column.
client.schema_alter("Person", "retype", "phone", col_type="text")

client.register_edge_type("Person", "Transaction", "AUTHORIZED")
# {"from_type": "Person", "to_type": "Transaction", "label": "AUTHORIZED"}
 
client.list_edge_types()
# {"edges": [{"from_type": "Person", "to_type": "Transaction",
#             "label": "AUTHORIZED"}, ...]}

client.deregister_type("ExperimentalType")
# {"deleted": True, "name": "ExperimentalType"}

ingest = IngestClient.from_client(client)

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": []}

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"},
])

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."},
])

# 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_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, ...}

chunks = '{"chunk_id":"c1","text":"SAR filed on Pacific Trust account 7742"}\n' \
         '{"chunk_id":"c2","text":"Alice Chen authorized 4 transfers totaling $2.8M"}'
manifest = '{"source":"sar.pdf","extractor":"dgrep-v1","chunks":2}'
 
client.ingest_document(chunks, manifest)
# IngestDocumentResponse(report_id="rep_019fe2...", chunks_ingested=2,
#                        warnings=[], queue_depth=0)

result = client.query("SELECT name, role, company FROM Person WHERE risk = 'HIGH'")
for row in result:
    print(row["name"], row["role"])

{"data": [
  {"name": "Alice Chen", "role": "CFO",      "company": "Acme Corp"},
  {"name": "Bob Smith",  "role": "Director", "company": "ShellCo Ltd"}
]}

client.query("SELECT SUM(amount) FROM Transaction")
# {"data": [{"SUM(amount)": 3770000}]}
 
client.query(
    "SELECT from_account, COUNT(*), SUM(amount) "
    "FROM Transaction GROUP BY from_account"
)

{"data": [
  {"from_account": "Acme Corp",           "COUNT(*)": 2, "SUM(amount)": 2800000},
  {"from_account": "Pacific Trust 7742",  "COUNT(*)": 1, "SUM(amount)": 850000},
  {"from_account": "ShellCo Ltd",         "COUNT(*)": 1, "SUM(amount)": 120000}
]}

result = client.query_params(
    "SELECT name, role FROM Person WHERE risk = $1 AND company = $2",
    ["HIGH", "Acme Corp"],
)
# ?-placeholders are rewritten to $1, $2, … automatically:
client.query_params("SELECT * FROM Person WHERE id = ?", ["alice"])

result = (
    client.select("name", "risk")
         .from_("Person")
         .where("risk = 'HIGH'")
         .order_by("name")
         .limit(10)
         .execute()
)

tbl = client.query_arrow("SELECT * FROM Transaction LIMIT 1000")
df = tbl.to_pandas()   # requires pyarrow

client.multi_search({
    "queries": [
        {"query": "alice",       "type": "Person",      "limit": 5},
        {"query": "embezzlement","type": "CaseDoc",     "limit": 5},
        {"query": "pacific trust","type": "Transaction","limit": 5},
    ],
})
# {"results": [<SearchResponse>, <SearchResponse>, ...],
#  "processing_time_ms": 7.2}

client.graphql("""
  query {
    Person(where: { risk: { _eq: "HIGH" } }, limit: 10) {
      id name role company
    }
  }
""")

client.sparql("""
  PREFIX rel: <https://relata.io/ns#>
  SELECT ?s ?o WHERE { ?s rel:authorizedBy ?o } LIMIT 5
""")

res = client.search(
    "alice chen", "Person",
    limit=10,
    facets=["company", "risk"],
    highlight=True,
    filters={"company": "Acme Corp"},
    matching_strategy="all",
)
for hit in res.hits:
    print(hit.score, hit.fields["name"])
# 8.42 Alice Chen

{
  "hits": [{"score": 8.42, "fields": {"name": "Alice Chen", ...}}],
  "total": 1,
  "estimated_total_hits": 1,
  "facets": {"company": {"Acme Corp": 1}, "risk": {"HIGH": 1}},
  "processing_time_ms": 2.1
}

client.query(
    "HYBRID_SEARCH FROM CaseDoc "
    "QUERY 'embezzlement offshore transfers' LIMIT 3"
)

{"rows": 3, "data": [
  {"title": "Whistleblower Complaint by Carla Nunez", "_score": 12.84},
  {"title": "Acme Corp CFO Under Scrutiny",           "_score": 9.21},
  {"title": "Suspicious Activity Report",             "_score": 7.55}
]}

# Pure BM25 (keyword precision, no semantic fuzziness):
client.query(
    "HYBRID_SEARCH FROM CaseDoc QUERY 'ShellCo shell company' "
    "LIMIT 3 WEIGHTS 0.0 1.0 0.0"
)
 
# Balanced BM25 + vector via the search() door (set metric or weights to
# trigger the hybrid channel — #2672):
client.search(
    "embezzlement offshore", "CaseDoc",
    metric="cosine", weights=[0.0, 0.5, 0.5],
)

client.detect_identities(
    "Contact Alice at alice.chen@acmecorp.com or +14155550100. "
    "Wire to Pacific Trust account 7742, IBAN GB29NWBK60161331926819."
)
# {"data": [
#   {"kind": "email",  "value": "alice.chen@acmecorp.com"},
#   {"kind": "phone",  "value": "+14155550100"},
#   {"kind": "iban",   "value": "GB29NWBK60161331926819"}, ...]}

# The SQL operator — one query, no detection code:
client.query("LOOKUP_IDENTITY '+14155550100'")
# {"rows": 1, "data": [{"id": "alice", "name": "Alice Chen",
#                       "phone": "+14155550100", "risk": "HIGH"}]}
 
# The typed SDK helper — resolve to the full identity cluster:
client.identity_cluster("+14155550100")
# {"data": [{"id": "alice", "match_kind": "phone",
#            "cluster": [{"kind":"email","value":"alice.chen@acmecorp.com"},
#                         {"kind":"phone","value":"+14155550100"}, ...]}]}
 
client.resolve_ids("+14155550100", mode="canonical")
# Returns the single best canonical match.

client.same_identity("+14155550100", "alice.chen@acmecorp.com")
# True  — both resolve to Alice.
 
client.same_identity("+14155550100", "bob@shellco.io")
# False

# Merge two records that are actually the same person:
client.fuse_identities("alice", "a.chen.duplicate")
# Writes an IdentityLink with link_type='fused'; returns the merged cluster.
 
# Undo a mistaken fuse:
client.split_identities("alice", "a.chen.duplicate")

client.create_link("AUTHORIZED", "alice", "Person", "tx1", "Transaction")
client.create_link("AUTHORIZED", "david", "Person", "tx2", "Transaction")
client.create_link("AUTHORIZED", "bob",   "Person", "tx3", "Transaction")
client.create_link("AUTHORIZED", "alice", "Person", "tx4", "Transaction")
# {"link_name": "AUTHORIZED", "source_id": "alice", ...}

client.graph_shortest_path("alice", "tx3")
# {"path": ["alice", "tx1", "Pacific Trust 7742", "tx2", "ShellCo Ltd", "tx3"],
#  "hops": 5}
 
client.graph_dijkstra("Transaction", "tx1", "tx3")  # weighted variant

client.graph_traverse("alice", direction="out", max_depth=3, limit=50)
# {"nodes": [{"id": "tx1"}, {"id": "tx4"}, ...],
#  "edges": [{"from": "alice", "to": "tx1", "label": "AUTHORIZED"}, ...]}

client.graph_pagerank("Person", damping=0.85, max_iter=100)
# {"data": [{"id": "alice", "score": 0.42},
#           {"id": "bob",   "score": 0.21}, ...]}
 
client.graph_community("Person")
# {"communities": [{"id": 0, "members": ["alice", "bob", "david"]}, ...]}
 
client.graph_scc("Person")                # strongly connected components
client.graph_triangle_count("Transaction") # graph density / cohesion

client.graph_link_predict("Person")
# Predicts missing relationships: [{"from": "alice", "to": "david", "score": 0.78}, ...]
 
client.graph_node_similarity("Person", "alice")
# {"data": [{"id": "bob", "score": 0.91}, {"id": "david", "score": 0.64}]}

mcp = McpClient.from_client(client)
 
mcp.investigate_entity("Person", "alice")
# {"profile": {...}, "timeline": [...], "connections": [...],
#  "risk": {"score": 0.92, "factors": ["sanctions_proximity", ...]}}

client.sanctions_screen("Alice Chen")
# {"data": []}  — no hits yet
 
mcp.screen_sanctions("Alice Chen", threshold=0.85)
# {"rows": 0}

client.beneficial_ownership_chain("Pacific Trust 7742", max_depth=6)
# {"chain": [{"account": "Pacific Trust 7742", "holder": "David Kim"},
#            {"holder": "David Kim", "nominee_for": "Bob Smith"}, ...]}

client.crypto_trace("0x742d35Cc6634C0532925a3b844Bc9e7595f0bEb")
# {"hops": [{"from": "0x742d...", "to": "0x9ab1...", "amount": 12.5}, ...]}
 
client.wire_reconstruction("Pacific Trust 7742", tolerance_pct=5.0)
# {"chain": [{"account": "...", "in": 2800000, "out": 850000}, ...]}
 
client.hawala_trace("alice", max_hops=5)
# {"network": [{"node": "alice", "linked_to": ["hawaladar_1", ...]}, ...]}

client.geofence("POINT(-97.74 30.27)", target_type="MovementEvent")
# {"data": [{"id": "m1", "lat": 30.27, "lon": -97.74}, ...]}
 
client.crime_pattern_cluster("downtown_austin")
# {"clusters": [{"centroid": [...], "events": 23}, ...]}

client.burner_detect(max_age_days=30, max_calls=3)
# {"burners": [{"phone": "+14155550199", "age_days": 12, "calls": 2}, ...]}
 
client.convoy_detect(radius_m=200, time_tol_secs=300, min_points=3)
# {"convoys": [{"members": ["veh_1", "veh_2", "veh_3"], "window": [...]}]}

client.vessel_track(538005644, window_secs=86400)        # MMSI → AIS track
client.dark_fleet_detect(max_gap_hours=24)               # AIS gaps ("going dark")
client.vessel_to_vessel_transfer(proximity_nm=0.5,
                                  time_window_minutes=120)
# {"transfers": [{"vessel_a": 538..., "vessel_b": 636..., ...}]}

mem = Memory("http://localhost:9090",
             purpose="agent-notes",
             bearer_token="perftoken",
             session_id="shadow-ledger")

mid1 = mem.add("Alice Chen authorized $2.3M wire to Pacific Trust account 7742 on Jan 15.",
               confidence=0.95, memory_class="episodic")
mid2 = mem.add("Bob Smith directs ShellCo Ltd, received $850K from the offshore account.")
mid3 = mem.add("Carla Nunez blew the whistle — four fraudulent transfers, $2.8M total.")
mid4 = mem.add("Classic laundering: placement → layering → integration.",
               memory_class="procedural")
# mid1 = "019fe254-3647-77fc-..."

ids = mem.add_batch([
    "The wire transfer matches a known fraud pattern: rapid offshore movement.",
    {"content": "Customer #1234 has no prior history with this beneficiary.",
     "confidence": 0.8, "memory_class": "semantic"},
    "Beneficiary account opened 2 days before the transfer request.",
])
# ids = ["019fe255-...", "019fe256-...", "019fe257-..."]

hits = mem.search("How much money was transferred?", top_k=2)

mem.search("Who is the whistleblower?")
# [{"content": "Carla Nunez blew the whistle...", "score": 1.0, ...}]
 
mem.search("What is ShellCo?")
# [{"content": "Bob Smith directs ShellCo Ltd...", "score": 1.0, ...}]

envelope = mem.search_detailed(
    "money transfer",
    top_k=5,
    min_confidence=0.5,            # CONFIDENCE
    recency_half_life_secs=86400,  # RECENCY
    budget_tokens=2048,            # BUDGET
    stability_days=30.0,           # FORGETTING_CURVE
    cancel_threshold=0.2,          # CANCEL_WHEN
)
# envelope = {"rows": [...],
#             "recall_cost_tokens": 412,   # BUDGET running total
#             "cancelled": False}          # CANCEL_WHEN short-circuit

mem.associate(mid1, mid2, relation="same_investigation")
# {"from_id": mid1, "to_id": mid2, "relation": "same_investigation"}

mem.episodes(session_id="shadow-ledger")
# [{"id": "ep_1", "session_id": "shadow-ledger",
#   "summary": "Operation Shadow Ledger investigation", ...}]

mem.justify(mid1)
# {"found": True,
#  "provenance": {"prov_hex": "a3f8b2c1...",
#                  "source": "memory:remember",
#                  "timestamp": "2026-08-08T16:20:14Z"}}

new_id = mem.update(mid1, "UPDATED: Alice authorized $2.3M — confirmed by 2 sources.")
# Old belief is superseded, not deleted. Bi-temporal history preserves it.
 
mem.resolve(new_id)   # follow the supersession chain to canonical head
# {"id": new_id, "content": "UPDATED: ...", "supersedes": [mid1]}
 
mem.summarise([mid1, mid2, mid3], summary_content="Three findings on Shadow Ledger.")
# {"id": "summ_...", "content": "Three findings on Shadow Ledger."}
 
mem.forget(mid4)      # governed retention-policy retract (not a hard delete)
# {"memory_item_id": mid4, "policy": "soft_delete",
#  "forget_at_ns": 1789234560000000000}

mcp = McpClient.from_client(client)
 
mcp.initialize()    # handshake
tools = mcp.list_tools()
# tools = [{"name": "query_knowledge", ...}, {"name": "recall", ...}, ...]
# len(tools) == 69

claude mcp add relata http://localhost:9090/mcp \
  --header "Authorization: Bearer perftoken"

mcp.query_knowledge("SELECT name, risk FROM Person WHERE risk = 'HIGH'",
                    purpose="analytics")
 
mcp.search_knowledge("embezzlement offshore", purpose="analytics", top_k=5)
 
mcp.explain_policy("SELECT * FROM Person", purpose="analytics")
# Shows the ACL / org-isolation policy that would apply, without executing.
 
mcp.suggest_extensions()
# Lists extension packs and their data availability.

mcp.list_entity_types()
mcp.get_entities("Person", filters={"risk": "HIGH"}, limit=10)
mcp.search_entities("alice", entity_types=["Person"])
mcp.get_domain_summary("financial")
# {"counts": {"Person": 4, "Transaction": 4, ...}, "freshness": {...}}

mcp.get_entity_profile("alice", purpose="analytics")
mcp.get_timeline("alice", purpose="analytics",
                 since_ns=1736899200000000000)   # since 2025-01-15
mcp.find_connections("alice", purpose="analytics", limit=50)
mcp.get_relationships(subject="alice", predicate="AUTHORIZED",
                      purpose="analytics")
mcp.investigate_entity("Person", "alice")
mcp.find_threats("ProcessEvent")
mcp.add_case_note("case-2026-001",
                  "Alice authorized 4 transfers to Pacific Trust 7742.",
                  author="investigator-1")
mcp.get_case_summary("case-2026-001", purpose="analytics")

mcp.lookup_identity("+14155550100", purpose="analytics")
mcp.resolve_entity_identity("alice", purpose="analytics")

mid = mcp.remember("Alice authorized $2.3M wire on Jan 15.", purpose="agent-notes")
mcp.remember_batch([{"content": "ShellCo received $850K."},
                    {"content": "Carla is the whistleblower."}],
                   purpose="agent-notes")
mcp.recall("whistleblower", purpose="agent-notes", top_k=3)
mcp.recognize(mid, purpose="agent-notes")
mcp.justify(mid, purpose="agent-notes")
mcp.consolidate(mid, "UPDATED: confirmed by 2 sources.", purpose="agent-notes")
mcp.forget(mid, retain_days=90, purpose="agent-notes")
mcp.remember_procedure("fraud-agent", "SAR_FILING",
                       "1. Screen. 2. Document. 3. File SAR within 30 days.",
                       purpose="agent-notes")
mcp.recall_procedure("fraud-agent", name="SAR_FILING", purpose="agent-notes")
mcp.associate(mid, mid2, relation="same_case", purpose="agent-notes")
mcp.resolve(mid, purpose="agent-notes")
mcp.summarise(ids=[mid, mid2], purpose="agent-notes")
mcp.episodes_in("shadow-ledger", purpose="agent-notes")

mcp.nl_query("show all high risk persons", purpose="analytics")
# {"dialect": "sql", "sql": "SELECT * FROM Person WHERE risk = 'HIGH'",
#  "rows": [...]}
 
mcp.nl_query("who is alice connected to",
             purpose="analytics", max_sub_questions=2)
# {"dialect": "cypher", "decomposed": True,
#  "sub_results": [{"dialect": "sql", ...}, {"dialect": "cypher", ...}]}

mcp.import_sigma("""
title: Suspicious Large Wire Transfer
status: experimental
logsource:
  product: relata
  service: Transaction
detection:
  selection:
    amount: 1000000
  condition: selection
level: high
""", purpose="security")
# {"rule_id": "019fe24e-e333-...", "name": "Suspicious Large Wire Transfer",
#  "status": "active"}
 
mcp.list_rules()
mcp.create_rule("high_value_wire",
                "SELECT * FROM Transaction WHERE amount > 1000000",
                severity="high", purpose="security")
mcp.list_jobs()
mcp.schedule_job("high_value_wire")
mcp.job_status()

mcp.list_workflows()
mcp.run_workflow("sar_filing_workflow")
mcp.workflow_status("<run_id>")

mcp.detect_communities("Person", purpose="analytics")
mcp.rank_key_nodes("Person", metric="pagerank", purpose="analytics")
mcp.hub_authority("Person", purpose="analytics")
mcp.find_scc("Person", purpose="analytics")
mcp.predict_links("Person", from_id="alice", purpose="analytics")
mcp.paths_between("alice", "tx3", max_hops=4, purpose="analytics")
mcp.list_link_types()

mcp.trace_crypto("0x742d...", max_hops=5, purpose="analytics")
mcp.beneficial_ownership("alice", max_depth=6, purpose="analytics")
mcp.reconstruct_wire("Pacific Trust 7742", tolerance_pct=5.0)
mcp.trace_hawala("alice", max_hops=5)
mcp.screen_sanctions("Alice Chen", purpose="compliance_review")
mcp.geofence(30.27, -97.74, radius_m=1000, purpose="analytics")

mcp.rag_store_answer("Who authorized the transfers?", "Alice Chen.",
                     source_ids=["tx1", "tx4"], purpose="rag")
mcp.rag_store_elements([{"type": "fact", "text": "ShellCo is a front."}],
                       purpose="rag")
mcp.ingest_document(chunks_jsonl=chunks, manifest_json=manifest,
                    purpose="rag")
mcp.hybrid_search("CaseDoc", "embezzlement", top_k=5, purpose="analytics")
mcp.similar_multimodal("MediaEmbedding", "img_42",
                       modality="image", purpose="investigation")
mcp.search_video_frames("frame_1", top_k=20, purpose="security_incident")
mcp.ingest_media("MediaImage", bytes_b64="...", modality="image")

mcp.server_health()
mcp.job_status()
mcp.metrics()
mcp.aggregate_stats("Transaction", agg="SUM", column="amount")
mcp.get_audit_trail(principal_filter="investigator-1", limit=100)

client.face_search(
    "gallery-1", [0.12, -0.34, 0.56, ...],   # 128-dim probe embedding
    k=5, threshold=0.6, purpose="investigation",
)
# QueryResult: columns entity_id, gallery_id, score, modality (ranked desc)
 
client.match_pdq(
    "corpus-1", "a1b2c3d4e5f67890...",        # PDQ hex hash
    threshold=0.9, purpose="investigation",
)
# QueryResult: entity_id, media_type, corpus_id, score, matcher
 
client.similar_image(
    "media-42", threshold=0.6, index="ncmec",
    purpose="investigation",
)
# QueryResult: near-duplicate images, ranked desc
 
client.dns_tunnel_detect("ws-finance-01", purpose="security")
# {"data": [{"domain": "exfil.bad.tld", "entropy": 7.8, "count": 1242}]}

client.erase_subject("alice", reason="gdpr-art17-request")
# {"subject": "alice", "shredded": True,
#  "rows_affected": 4, "vectors_removed": 1, "blobs_removed": 0,
#  "receipt": {"signed": "..."}}

# Stage some writes, review them, then commit or discard.
client.session_diff("session-shadow-ledger")
# {"changes": [{"type": "Person", "id": "alice", "op": "upsert", ...}]}
 
client.session_commit("session-shadow-ledger")
# {"committed": True, "rows": 4, "commit_hash": "f3a2c8b1..."}
 
# Or throw them away:
client.session_discard("session-shadow-ledger")
# {"discarded": True}

client.export_data("Transaction", format="json")
# {"type": "Transaction", "format": "json",
#  "rows": [...], "exported_at": "2026-08-08T..."}
 
client.export_data("Person", format="csv")

client.create_schema_branch("hypothesis-bribery", from_branch="main")
# {"created": True, "branch": "hypothesis-bribery", "source": "main"}
 
# Ingest alternative theories into the branch, run queries, compare.
# Then either merge or delete:
client.delete_schema_branch("hypothesis-bribery")
# {"deleted": True}

docs = client.namespace("CaseDoc")
 
# Schemaless upsert (POST /ingest/auto — auto-creates the type if absent).
docs.write([
    {"id": "memo1", "title": "Bribery hypothesis",
     "body": "Pacific Trust may be a kickback conduit.", "status": "draft"},
], schema={"title": "text", "body": "text"})
 
# Typed ranked search — text compiles to BM25; filters narrow server-side.
res = docs.query(
    text="bribery kickback",
    match_column="title",
    filters=[{"field": "status", "op": "eq", "value": "draft"}],
    limit=5,
)
for row in res:
    print(row["title"])
 
docs.get("memo1")           # point lookup
docs.delete_all()           # governed bi-temporal tombstone every row
docs.branch_from("CaseDoc") # T10 copy-on-write namespace branch

client.cluster_nodes()
# [ClusterNode(id="node-7f3a", role="coordinator", addr="10.0.0.1:9090",
#              partitions=[0,1,2], state="healthy"),
#  ClusterNode(id="node-9b2c", role="reader",      addr="10.0.0.2:9090", ...)]
 
client.cluster_topology()
# {"nodes": [...], "partitions": [{"id": 0, "primary": "node-7f3a",
#                                  "replicas": ["node-9b2c"]}], "roles": {...}}
 
client.cluster_rebalance()
# {"rebalanced": True, "partitions_moved": 3, "duration_ms": 1240}
 
client.cluster_drain("node-9b2c")   # evacuate for maintenance
# {"drained": True, "node": "node-9b2c", "partitions_relocated": 3}

for event in client.observe_stream():
    print(event["kind"], event["level"], event["message"])
# query INFO  SELECT name FROM Person (rows=4, latency_ms=2.1)
# ingest INFO bulk Person rows=4
# ...

client.register_webhook(
    "https://hooks.slack.com/services/...",
    event_types=["ingest.completed", "alert.triggered"],
)
# {"id": "wh_019fe2...", "url": "https://...", "event_types": [...]}
 
client.list_webhooks()
# {"webhooks": [{"id": "wh_019fe2...", ...}]}
 
client.delete_webhook("wh_019fe2...")
# {"deleted": True}

tbl = client.query_flight(
    "SELECT * FROM Transaction LIMIT 1000", purpose="analytics",
)
df = tbl.to_pandas()
# Requires RELATA_FLIGHT_ENABLE=true (port 8815); pyarrow only, no grpcio.

result = client.query_grpc("SELECT * FROM Person LIMIT 1000")
# Same QueryResult shape as query(); requires grpcio
# (pip install relata-sdk[grpc]).
 
result = client.query_grpc_stream("SELECT * FROM Transaction")
# Server-streaming variant — same shape, frame-by-frame.

from relata import AuditClient
 
audit = AuditClient.from_client(client)
 
audit.count()
# AuditCountResponse(count=47, entries=47,
#                    chain_valid=True, chain_head="f3a2c8b1...")

# Paginated entries with filters:
audit.entries(principal="investigator-1",
              purpose="analytics",
              limit=5)
# {"entries": [{"ts_ns": ..., "principal": "investigator-1",
#               "action": "query", "sql": "SELECT ...",
#               "decision": "allow", "request_id": "req_..."}, ...],
#  "next_cursor": "...", "chain_valid": True}
 
audit.find_by_request_id("req_abc123")
# Single entry, or None

pdf_bytes = audit.export_pdf("case-2026-001", template="default")
Path("shadow-ledger.pdf").write_bytes(pdf_bytes)
 
receipt = audit.sign_receipt({
    "case_id": "case-2026-001",
    "subject": "alice",
    "action": "gdpr_erasure",
})
# {"signed": True, "signature": "...", "signed_at": "..."}

# Valid time: what was true on Jan 20?
client.query("SELECT COUNT(*) FROM Person AS OF '2026-01-20T00:00:00Z'")
 
# System time: what did we KNOW on Jan 20?
client.query("SELECT COUNT(*) FROM Person AS OF SYSTEM TIME '2026-01-20T00:00:00Z'")

result = client.query("SELECT name, company FROM Person WITH PROVENANCE")
# Each row includes prov_hex, source, commit_hash, timestamp.

result = (
    client.select("Person")
         .where("risk = 'HIGH'")
         .as_of("2026-01-20")
         .with_provenance()
         .limit(20)
         .execute()
)