Processing a Large Media Library

import os, json, urllib.request
from mixpeek import Mixpeek

client = Mixpeek(api_key=os.environ["MIXPEEK_API_KEY"], namespace="media-library")

BUCKET_ID = "bkt_your_library"
BATCH_ID = "bat_the_draft_you_just_created"
COLLECTION_ID = "col_enriched_media"
ARCHIVE_COLLECTION_ID = "col_2019_archive"
CONNECTION_ID = "con_your_s3_connection"

BASE = "https://api.mixpeek.com/v1"
HEADERS = {
    "Authorization": f"Bearer {os.environ['MIXPEEK_API_KEY']}",
    "X-Namespace": "media-library",
    "Content-Type": "application/json",
}


def api(method, path, body=None):
    """The facade covers buckets, collections, retrievers, documents,
    evaluations, tasks and namespaces. Batch cost, bucket syncs and collection
    lifecycle have no resource on it yet, so those three go over HTTP."""
    req = urllib.request.Request(
        f"{BASE}{path}",
        method=method,
        data=json.dumps(body).encode() if body is not None else None,
        headers=HEADERS,
    )
    with urllib.request.urlopen(req) as r:
        return json.load(r)


# 1. Price the run before you spend anything. estimate-cost consumes no credits
#    and changes nothing about the batch. already_extracted_count is the number
#    you care about: under 'force' or 'replace' those objects go back through a
#    GPU and estimated_repay_credits is what that costs.
est = api("POST", f"/buckets/{BUCKET_ID}/batches/{BATCH_ID}/estimate-cost")
print(est["object_count"], "objects,", est["already_extracted_count"], "already done")
print(est["estimated_credits"], "credits")
if est.get("estimated_usd") is not None:
    print("approx USD:", round(est["estimated_usd"], 2))
for w in est.get("warnings", []):
    print("warning:", w)

# 2. Run it. dedup_strategy defaults to 'skip' and is scoped to
#    (bucket, collection), so re-triggering after the bucket grows pays only for
#    the new objects. Reach for 'replace' when the extractor itself changed and
#    the old documents are wrong; 'force' allows duplicates and is rarely what
#    you want.
client.collections.trigger(COLLECTION_ID, dedup_strategy="skip")

# 3. Stop triggering by hand. A sync watches the bucket's storage connection and
#    submits what lands. skip_duplicates is on by default and does the same job
#    as dedup_strategy for objects arriving continuously.
sync = api("POST", f"/buckets/{BUCKET_ID}/syncs", {
    "connection_id": CONNECTION_ID,
    "source_path": "incoming/",
    "sync_mode": "continuous",
    "polling_interval_seconds": 300,
    "batch_size": 50,
    "skip_duplicates": True,
})

# 4. Keep the hot set small. 'cold' evicts a collection's vectors from the vector
#    store and they stay searchable from object storage; 'active' rehydrates.
#    Search latency is set by what is resident, so an archive that nobody queries
#    should not be.
api("PATCH", f"/collections/{ARCHIVE_COLLECTION_ID}/lifecycle",
    {"lifecycle_state": "cold"})