Streaming & incremental

import goldenmatch as gm

matches = gm.match_one(
    {"name": "John Smith", "zip": "10001"},
    existing_df,
    matchkey,
)
# Returns list of (row_id, score) tuples

import goldenmatch as gm

processor = gm.StreamProcessor(existing_df, config)

# Process a single record immediately
matches = processor.process_record({"name": "John Smith", "zip": "10001"})

# Micro-batch mode: process a list of records at once
results = processor.process_batch(incoming_records)

processor = gm.StreamProcessor(df, config)
result = processor.process_record(new_record)
# result includes matches and updated cluster assignments

processor = gm.StreamProcessor(df, config)
batch_results = processor.process_batch(batch)

import goldenmatch as gm

# Add a record to a cluster (join or merge clusters)
gm.add_to_cluster(record_id, matches, clusters)

# Remove a record from its cluster and re-cluster
gm.unmerge_record(record_id, clusters)

# Shatter a cluster into singletons
gm.unmerge_cluster(cluster_id, clusters)

goldenmatch incremental base.csv --new-records new_records.csv --config config.yaml

from goldenmatch.core.ann_blocker import ANNBlocker

blocker = ANNBlocker(top_k=20)
blocker.add_to_index(embedding)        # Add a single embedding
neighbors = blocker.query_one(embedding)  # Find nearest neighbors

goldenmatch watch --table customers --connection-string "$DATABASE_URL" --config config.yaml --interval 30

# First run: full scan
goldenmatch sync --table customers --connection-string "$DATABASE_URL" --config config.yaml

# Subsequent runs: incremental (only new records)
goldenmatch sync --table customers --connection-string "$DATABASE_URL"

import goldenmatch as gm

def poll_new_records():
    # Return the latest batch of records as a list of dicts.
    return fetch_new_rows()

# Returns a StreamProcessor; call .stop() to halt the loop.
processor = gm.run_stream(existing_df, config, poll_new_records, poll_interval=30)

New Record
    |
    v
match_one() --> (row_id, score) pairs
    |
    v
add_to_cluster() --> updated cluster dict
    |
    v
build_golden_record() --> updated canonical record

Postgres table (new rows)
    |
    v
watch/sync --> load new records
    |
    v
Hybrid blocking (SQL + ANN)
    |
    v
Score + cluster + golden
    |
    v
Write back to gm_clusters, gm_golden_records