stripe-list-and-query-objects

copy

import sys
import os
import subprocess
import time
import urllib.request
import json

# ─────────────────────────────────────────
# PRE_EXECUTION
# ─────────────────────────────────────────

for attempt in range(2):
    try:
        req = urllib.request.Request(
            "https://checklist.day/api/registry/stripe",
            headers={"User-Agent": "checklist-agent/1.0"}
        )
        with urllib.request.urlopen(req, timeout=10) as resp:
            registry = json.loads(resp.read())
            break
    except Exception as e:
        if attempt == 1:
            print(f"ABORT: registry unreachable — {e}")
            sys.exit(1)
        time.sleep(2)

warnings = registry.get("warnings", [])
if warnings:
    print("[stripe] WARNINGS:")
    for w in warnings if isinstance(warnings, list) else [warnings]:
        print(f"  ⚠ {w}")

# ─────────────────────────────────────────
# EXECUTION
# ─────────────────────────────────────────

subprocess.check_call([sys.executable, "-m", "pip", "install", "-q", "stripe>=7.0.0"])

import stripe

STRIPE_SECRET_KEY = os.environ.get("STRIPE_SECRET_KEY")
if not STRIPE_SECRET_KEY:
    print("ABORT: STRIPE_SECRET_KEY not set"); sys.exit(1)
if not STRIPE_SECRET_KEY.startswith("sk_test_"):
    print("ABORT: use test mode key"); sys.exit(1)

stripe.api_key = STRIPE_SECRET_KEY

TAG = f"checklist-list-test-{int(time.time())}"

# Create 3 test customers with unique metadata tag
created_ids = []
for i in range(3):
    c = stripe.Customer.create(
        email=f"checklist-list-{i}@example.com",
        metadata={"test_tag": TAG},
    )
    created_ids.append(c.id)
print(f"  created {len(created_ids)} test customers (tag={TAG})")

try:
    # 1. Basic list
    listing  = stripe.Customer.list(limit=10)
    list_ok  = len(listing.data) > 0
    print(f"  list: {len(listing.data)} customers returned (list_ok={list_ok})")

    # 2. Auto-pagination
    # FOOTGUN: list() returns max 100 per page — use auto_paging_iter() for full dataset
    # FOOTGUN: auto_paging_iter() makes multiple API calls — set a limit to avoid runaway
    total_fetched      = 0
    autopagination_ok  = False
    for customer in stripe.Customer.list(limit=3).auto_paging_iter():
        total_fetched += 1
        if total_fetched >= 10:
            break
    autopagination_ok = total_fetched > 0
    print(f"  auto_paging_iter: fetched {total_fetched} (autopagination_ok={autopagination_ok})")

    # 3. Filter by metadata
    # FOOTGUN: Stripe Search API uses query syntax, not dict filters
    search_result = stripe.Customer.search(
        query=f'metadata["test_tag"]:"{TAG}"',
    )
    found_ids          = [c.id for c in search_result.data]
    metadata_filter_ok = all(cid in found_ids for cid in created_ids)
    print(f"  metadata search: found {len(found_ids)} (metadata_filter_ok={metadata_filter_ok})")

finally:
    for cid in created_ids:
        stripe.Customer.delete(cid)
    print(f"  cleaned up {len(created_ids)} test customers")

# ─────────────────────────────────────────
# POST_EXECUTION
# ─────────────────────────────────────────

assert list_ok, "FAIL: list_customers returned empty"
assert autopagination_ok, "FAIL: auto_paging_iter returned 0 results"
assert metadata_filter_ok, f"FAIL: metadata search did not find all test customers. Found: {found_ids}"

result = {
    "list_ok":            list_ok,
    "autopagination_ok":  autopagination_ok,
    "metadata_filter_ok": metadata_filter_ok,
    "total_fetched":      total_fetched,
}
print(json.dumps(result, indent=2))
print("PASS")