Example: storefront script

1
#!/usr/bin/env python3
2
"""
3
ListBee storefront script.
4
5
Usage:
6
  # Create listings from catalog file
7
  python storefront.py create --catalog catalog.json
8
9
  # Monitor orders (poll every 30 seconds)
10
  python storefront.py orders --interval 30
11
12
  # Both: create listings then watch for orders
13
  python storefront.py create --catalog catalog.json --then-watch
14
"""
15
16
import argparse
17
import json
18
import os
19
import sys
20
import time
21
import uuid
22
from datetime import datetime, timezone
23
24
import httpx
25
26
BASE = "https://api.listbee.so"
27
28
29
def get_client() -> tuple[httpx.Client, str]:
30
    """Return (client, api_key). Reads LISTBEE_API_KEY from environment."""
31
    api_key = os.environ.get("LISTBEE_API_KEY")
32
    if not api_key:
33
        print("Error: LISTBEE_API_KEY environment variable not set.", file=sys.stderr)
34
        sys.exit(1)
35
36
    client = httpx.Client(
37
        base_url=BASE,
38
        headers={"Authorization": f"Bearer {api_key}"},
39
        timeout=30.0,
40
    )
41
    return client, api_key
42
43
44
def create_listings(client: httpx.Client, catalog_path: str) -> list[dict]:
45
    """Read catalog JSON and create all listings. Return created listing objects."""
46
    with open(catalog_path) as f:
47
        catalog = json.load(f)
48
49
    print(f"Creating {len(catalog)} listings...")
50
    created = []
51
52
    for item in catalog:
53
        idempotency_key = str(uuid.uuid4())
54
55
        try:
56
            resp = client.post(
57
                "/v1/listings",
58
                json=item,
59
                headers={"Idempotency-Key": idempotency_key},
60
            )
61
            resp.raise_for_status()
62
            listing = resp.json()
63
            created.append(listing)
64
            print(
65
                f"  ✓ {listing['name']:<40} {listing['url']}"
66
                f"  sellable={listing['readiness']['sellable']}"
67
            )
68
        except httpx.HTTPStatusError as exc:
69
            error = exc.response.json()
70
            print(
71
                f"  ✗ {item.get('name', '?'):<40} "
72
                f"error={error.get('code')} {error.get('detail')}",
73
                file=sys.stderr,
74
            )
75
76
    print(f"\nCreated {len(created)} of {len(catalog)} listings.")
77
    return created
78
79
80
def poll_orders(client: httpx.Client, interval: int = 30) -> None:
81
    """Poll GET /v1/orders and print new orders as they arrive."""
82
    print(f"Watching for orders (polling every {interval}s)... Ctrl+C to stop.\n")
83
84
    seen_ids: set[str] = set()
85
    cursor: str | None = None
86
87
    while True:
88
        try:
89
            params: dict = {"limit": 50}
90
            if cursor:
91
                params["cursor"] = cursor
92
93
            resp = client.get("/v1/orders", params=params)
94
            resp.raise_for_status()
95
            result = resp.json()
96
97
            for order in result["data"]:
98
                if order["id"] not in seen_ids:
99
                    seen_ids.add(order["id"])
100
                    ts = datetime.fromisoformat(order["created_at"].replace("Z", "+00:00"))
101
                    local_ts = ts.astimezone().strftime("%Y-%m-%d %H:%M:%S")
102
                    amount = order["amount"] / 100
103
                    print(
104
                        f"[{local_ts}] New order {order['id']}"
105
                        f"  buyer={order['buyer_email']}"
106
                        f"  amount=${amount:.2f} {order['currency'].upper()}"
107
                        f"  listing={order['listing_id']}"
108
                    )
109
110
            # Update cursor for next page (if paginating)
111
            if result.get("cursor"):
112
                cursor = result["cursor"]
113
114
        except httpx.HTTPStatusError as exc:
115
            print(f"Error polling orders: {exc}", file=sys.stderr)
116
        except KeyboardInterrupt:
117
            print("\nStopped.")
118
            break
119
120
        time.sleep(interval)
121
122
123
def cmd_create(args: argparse.Namespace) -> None:
124
    client, _ = get_client()
125
    listings = create_listings(client, args.catalog)
126
127
    not_sellable = [l for l in listings if not l["readiness"]["sellable"]]
128
    if not_sellable:
129
        print(f"\n{len(not_sellable)} listing(s) not yet sellable:")
130
        for l in not_sellable:
131
            next_code = l["readiness"]["next"]
132
            print(f"  {l['name']}: next action = {next_code}")
133
134
    if args.then_watch:
135
        print()
136
        poll_orders(client, interval=30)
137
138
139
def cmd_orders(args: argparse.Namespace) -> None:
140
    client, _ = get_client()
141
    poll_orders(client, interval=args.interval)
142
143
144
def main() -> None:
145
    parser = argparse.ArgumentParser(description="ListBee storefront script")
146
    subparsers = parser.add_subparsers(dest="command", required=True)
147
148
    create_parser = subparsers.add_parser("create", help="Create listings from catalog JSON")
149
    create_parser.add_argument("--catalog", required=True, help="Path to catalog.json")
150
    create_parser.add_argument(
151
        "--then-watch", action="store_true", help="Watch for orders after creating"
152
    )
153
    create_parser.set_defaults(func=cmd_create)
154
155
    orders_parser = subparsers.add_parser("orders", help="Poll for new orders")
156
    orders_parser.add_argument(
157
        "--interval", type=int, default=30, help="Poll interval in seconds (default: 30)"
158
    )
159
    orders_parser.set_defaults(func=cmd_orders)
160
161
    args = parser.parse_args()
162
    args.func(args)
163
164
165
if __name__ == "__main__":
166
    main()