Catalog

pip install zstandard lmdb

# Your catalog URL from voodoo.center/dashboard/api
CATALOG_URL="https://downloads.voodoo.center/{client_id}.lmdb.zst"
# Stream the download straight through the decompressor — nothing is buffered
# and the compressed archive is never written to disk.
curl -sS "$CATALOG_URL" | zstd -d -o catalog.lmdb

import json
import struct
import lmdb
import requests
import zstandard
# Your catalog URL from the dashboard (voodoo.center/dashboard/api)
CATALOG_URL = "https://downloads.voodoo.center/{client_id}.lmdb.zst"
def item_key(item_id: int) -> bytes:
    # Records are keyed by the 8-byte big-endian item id (so they sort by id).
    return struct.pack(">Q", item_id)
# 1. Stream the download straight through the zstd decompressor to disk. The
#    archive is never held in memory (compressed or decompressed) as a whole.
dctx = zstandard.ZstdDecompressor()
with requests.get(CATALOG_URL, stream=True) as response:
    response.raise_for_status()
    response.raw.decode_content = True   # undo any HTTP Content-Encoding
    with open("catalog.lmdb", "wb") as db:
        dctx.copy_stream(response.raw, db)
# 2. Open the env read-only and memory-mapped. It is a single FILE, not a
#    directory (subdir=False), and read-only readers must pass lock=False. LMDB
#    pages in only the records you touch, so opening costs almost nothing.
env = lmdb.open("catalog.lmdb", subdir=False, readonly=True, lock=False)
# Look up one item by id — a single page read, not a scan.
with env.begin() as txn:
    raw = txn.get(item_key(689556))
    item = json.loads(raw) if raw else None
    print(item)
# Iterate the whole catalog one record at a time — constant memory, ordered by id.
with env.begin() as txn:
    for key, value in txn.cursor():
        item = json.loads(value)
        if item["in_stock"]:
            price = item["price"] / 100  # cents -> major units
            print(item["id"], item["product_type"], f"{price:.2f}", item["name"])

{
  "id": 689556,
  "name": "Steam Top-up (USD)",
  "product_type": "topup",
  "min_quantity": 1,
  "max_quantity": 1000,
  "price": 500,
  "base_price": 500,
  "subscriber_price": 450,
  "amount_per_price": 1,
  "in_stock": true,
  "updated_at": "2026-07-05T12:00:00+00:00",
  "fields": [
    { "id": 88, "name": "steam_login", "type": "string", "required": true, "options": [] }
  ]
}

# `item` is a record read from the catalog above; `access_token` is your
# Bearer token (see Authentication).
import requests
body = {
    "item_id": item["id"],
    "quantity": 1,                  # within min_quantity..max_quantity
    "merchant_order_id": "order-2024",
    "fields": {
        "steam_login": "test",      # a text field -> its string value
    },
}
response = requests.post(
    "https://api.voodoo.center/v1/orders",
    headers={"Authorization": f"Bearer {access_token}"},
    json=body,
)
response.raise_for_status()
order = response.json()
print(order["id"], order["status"])

import os
import lmdb
import requests
import zstandard
# Your catalog URL from the dashboard (voodoo.center/dashboard/api)
CATALOG_URL = "https://downloads.voodoo.center/{client_id}.lmdb.zst"
DATABASE = "catalog.lmdb"
ETAG_FILE = "catalog.etag"
def refresh_catalog() -> bool:
    """Download the snapshot only if it changed; return True if a new copy was fetched."""
    headers = {}
    # Send the ETag from the last download so the server can reply 304 Not Modified.
    if os.path.exists(ETAG_FILE):
        with open(ETAG_FILE) as f:
            headers["If-None-Match"] = f.read().strip()
    dctx = zstandard.ZstdDecompressor()
    with requests.get(CATALOG_URL, headers=headers, stream=True) as response:
        if response.status_code == 304:          # nothing changed
            return False
        response.raise_for_status()              # surface any real error
        # A 200 reaches here — stream-decompress straight to the LMDB file, so the
        # snapshot is never buffered in memory or written to disk compressed.
        response.raw.decode_content = True
        with open(DATABASE, "wb") as db:
            dctx.copy_stream(response.raw, db)
        etag = response.headers.get("ETag")
    if etag:
        with open(ETAG_FILE, "w") as f:
            f.write(etag)
    return True
def open_catalog() -> lmdb.Environment:
    return lmdb.open(DATABASE, subdir=False, readonly=True, lock=False)
# Run refresh_catalog() on a schedule (e.g. every few minutes). It returns True
# only when a new snapshot was fetched — reopen the LMDB env when it does, and
# keep using the previous env until then.
if refresh_catalog():
    env = open_catalog()