From 95ffac2389e9d0b573d95c9000a8d6707e592fdc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Rickard=20Br=C3=BCel=20Gabrielsson?= Date: Tue, 27 Oct 2026 17:45:48 -0400 Subject: [PATCH] client.events: track your users' events from your server, and list them by stream What is left of #52 that 0.6.5 did not already do. `events.track` posts to /v1/events, with `user=` as in `complete`, and the catalog an `Item` names. `events.list` reads /v1/events/own, now with `stream=`: a publishable key, "api", or an import, which is how an app opened from the portal reads the streams it was opened on. `user_id=` on `complete` and `sections` stays as 0.6.5 shipped it. Co-Authored-By: Claude Opus 5.5 --- .claude/skills/portal-app/SKILL.md | 2 + README.md | 12 ++ src/behaviorgpt/_client.py | 3 + src/behaviorgpt/resources/events.py | 189 ++++++++++++++++++++++++++++ src/behaviorgpt/types/__init__.py | 2 + src/behaviorgpt/types/http.py | 11 ++ tests/test_events.py | 189 ++++++++++++++++++++++++++++ 7 files changed, 408 insertions(+) create mode 100644 src/behaviorgpt/resources/events.py create mode 100644 tests/test_events.py diff --git a/.claude/skills/portal-app/SKILL.md b/.claude/skills/portal-app/SKILL.md index f39bd8d..afebbea 100644 --- a/.claude/skills/portal-app/SKILL.md +++ b/.claude/skills/portal-app/SKILL.md @@ -45,6 +45,8 @@ for item in res.items: print(item.catalog_id, item.data["name"]) ``` +For an app that reads events, read each stream with `client.events.list(stream=...)`, or every stream when `streams` is missing; `user_id=` narrows it to one shopper. `client.complete(user_id=...)` ranks for a shopper from their stored events. + With one id, `catalog_ids=[cat]` and `catalog_id=cat` are the same. If the app opens on no catalog and none arrives, fall back to `UNBOXAI_CATALOG_ID` or ask; don't guess an id. The API refuses a catalog the key can't read and catalogs that can't be ranked together; show its message rather than an empty result. ## Workflow diff --git a/README.md b/README.md index 1e860a6..62b9e0b 100644 --- a/README.md +++ b/README.md @@ -102,6 +102,18 @@ res = client.complete([View("paB08NYK61PJ")], user=shopper) If your site already sends events (the portal's browser script, or `POST /events`), pass the person's id instead of a history: `client.complete(user_id=uid, user=shopper)`. The API adds the events it has stored for that id to whatever `history` you pass. With the browser script, the id is in the `unbox_uid` cookie. +From your server, record events with `client.events.track`, and read back what your account has stored with `client.events.list`, newest first: + +```python +client.events.track(uid, View(shoe), user=shopper) +client.events.track(uid, Order([shoe]), timestamp=paid_at) # sent later than it happened + +client.events.list(user_id=uid, limit=20) +client.events.list(stream="api", types=["order"]) # what your server sent; a stream is also a publishable key or an import +``` + +An event's catalog is `catalog_id=`, else the one an `Item` of a response names, else the client's default. Pass `timestamp=` whenever you send an event after it happened, so the history stays in order. + ## Your own catalog The model can only return items it knows about. Out of the box the client uses `retail_catalog`, a sample retail catalog. To use the model on your own data, upload your catalog as a parquet file. A catalog doesn't have to be products for sale: artworks, articles, listings or anything else people browse and pick from work the same way. diff --git a/src/behaviorgpt/_client.py b/src/behaviorgpt/_client.py index 8ae50bc..70365eb 100644 --- a/src/behaviorgpt/_client.py +++ b/src/behaviorgpt/_client.py @@ -14,6 +14,7 @@ from behaviorgpt._exceptions import AuthenticationError, UnboxAIError from behaviorgpt.resources.catalogs import AsyncCatalogs, Catalogs +from behaviorgpt.resources.events import AsyncEvents, Events from behaviorgpt.resources.recommendations import AsyncRecommender, Recommender from behaviorgpt.resources.requests import UnboxAIRequester from behaviorgpt.resources.search import AsyncSearcher, Searcher @@ -376,6 +377,7 @@ def __init__( self.recommendations = Recommender(self._http_client) self.search = Searcher(self._http_client) self.catalogs = Catalogs(self._http_client) + self.events = Events(self._http_client, self) self._sections = Sectioner(self._http_client) def complete( @@ -752,6 +754,7 @@ def __init__( self.recommendations = AsyncRecommender(self._http_client) self.search = AsyncSearcher(self._http_client) self.catalogs = AsyncCatalogs(self._http_client) + self.events = AsyncEvents(self._http_client, self) self._sections = AsyncSectioner(self._http_client) async def __aenter__(self) -> "AsyncUnboxAIClient": diff --git a/src/behaviorgpt/resources/events.py b/src/behaviorgpt/resources/events.py new file mode 100644 index 0000000..c38c436 --- /dev/null +++ b/src/behaviorgpt/resources/events.py @@ -0,0 +1,189 @@ +from datetime import UTC, datetime +from typing import Optional, Protocol, Sequence +from uuid import uuid4 + +import httpx + +from behaviorgpt.resources._shared import handle_response +from behaviorgpt.types import Domains, Order, UserEvent +from behaviorgpt.types.http import TrackedEvent +from behaviorgpt.types.user import User + +PATH = "/events" +OWN_PATH = "/events/own" +# the server's limit +MAX_LISTED = 200 + + +class EventContext(Protocol): + domains: Domains + timezone: str + default_catalog_id: str + + +def _event_catalog(event: UserEvent) -> Optional[str]: + """The catalog the event's items say they are in, if they say.""" + if isinstance(event, Order): + named = {entry.catalog_id for entry in event.items or ()} - {None} + if len(named) > 1: + raise ValueError( + "an Order's items are in one catalog; track one Order per catalog" + ) + return next(iter(named), None) + return getattr(event, "catalog_id", None) + + +def event_body( + context: EventContext, + user_id: str, + event: UserEvent, + *, + catalog_id: Optional[str], + timestamp: Optional[datetime], + user: Optional[User], +) -> dict: + """The body `track` posts for `event`.""" + if timestamp is None: + timestamp = datetime.now(tz=UTC) + elif timestamp.tzinfo is None: + raise ValueError("timestamp must be timezone-aware") + + payload = event.model_dump(mode="json", exclude_none=True) + if isinstance(event, Order): + if not event.items: + raise ValueError("track an Order with the items bought: Order([item, ...])") + # The API stores orders with a cart id and a unit price per item; the + # model reads only which items were bought. + payload["cart_id"] = event.cart_id or str(uuid4()) + payload["items"] = [ + { + "product": entry.item, + "quantity": entry.quantity, + "unit_selling_price": 0.0, + } + for entry in event.items + ] + + domains = user.domains(context.domains.market) if user else context.domains + return { + "user_id": user_id, + "store_id": catalog_id or _event_catalog(event) or context.default_catalog_id, + "event": payload, + "domains": domains.model_dump(mode="json", exclude_none=True), + "timezone": (user.timezone if user else None) or context.timezone, + "timestamp": timestamp.isoformat(), + } + + +def list_params( + user_id: Optional[str], + limit: int, + types: Optional[Sequence[str]], + stream: Optional[str], +) -> list[tuple[str, str]]: + if not 1 <= limit <= MAX_LISTED: + raise ValueError(f"limit must be between 1 and {MAX_LISTED}, got {limit}") + params: list[tuple[str, str]] = [("limit", str(limit))] + if user_id is not None: + params.append(("user_id", user_id)) + params.extend(("type", t) for t in types or ()) + if stream is not None: + params.append(("stream", stream)) + return params + + +def tracked_events(data: dict) -> list[TrackedEvent]: + return [TrackedEvent(**entry) for entry in data["events"]] + + +class Events: + """Your users' events. Record them with `track`; `complete(user_id=...)` + and `sections(user_id=...)` then read them as that user's history.""" + + def __init__(self, http_client: httpx.Client, context: EventContext): + self._client = http_client + self._context = context + + def track( + self, + user_id: str, + event: UserEvent, + *, + catalog_id: Optional[str] = None, + timestamp: Optional[datetime] = None, + user: Optional[User] = None, + ) -> str: + """Record that `user_id` did `event`, and return the event's id. + + `timestamp` is when it happened, now by default. Pass it when you send + events later than they happen, so that they stay in order. The catalog + is `catalog_id`, else the one an `Item` of a response names, else the + client's default. `user` says who they are, as in `complete`. + """ + body = event_body( + self._context, + user_id, + event, + catalog_id=catalog_id, + timestamp=timestamp, + user=user, + ) + return handle_response(self._client.post(PATH, json=body))["id"] + + def list( + self, + *, + user_id: Optional[str] = None, + limit: int = 50, + types: Optional[Sequence[str]] = None, + stream: Optional[str] = None, + ) -> list[TrackedEvent]: + """Your account's latest events, newest first: all your users', or + `user_id`'s. `types` keeps only those event types, such as + `["viewItem", "search"]`, and `stream` only the events of one stream: + a publishable key, "api" for those sent with your key, or an import + such as "import_shop". `limit` can be at most 200.""" + params = list_params(user_id, limit, types, stream) + response = self._client.get(OWN_PATH, params=params) + return tracked_events(handle_response(response)) + + +class AsyncEvents: + """`Events` for `AsyncUnboxAIClient`.""" + + def __init__(self, http_client: httpx.AsyncClient, context: EventContext): + self._client = http_client + self._context = context + + async def track( + self, + user_id: str, + event: UserEvent, + *, + catalog_id: Optional[str] = None, + timestamp: Optional[datetime] = None, + user: Optional[User] = None, + ) -> str: + """See `Events.track`.""" + body = event_body( + self._context, + user_id, + event, + catalog_id=catalog_id, + timestamp=timestamp, + user=user, + ) + return handle_response(await self._client.post(PATH, json=body))["id"] + + async def list( + self, + *, + user_id: Optional[str] = None, + limit: int = 50, + types: Optional[Sequence[str]] = None, + stream: Optional[str] = None, + ) -> list[TrackedEvent]: + """See `Events.list`.""" + params = list_params(user_id, limit, types, stream) + response = await self._client.get(OWN_PATH, params=params) + return tracked_events(handle_response(response)) diff --git a/src/behaviorgpt/types/__init__.py b/src/behaviorgpt/types/__init__.py index 98fa53b..a242eec 100644 --- a/src/behaviorgpt/types/__init__.py +++ b/src/behaviorgpt/types/__init__.py @@ -28,6 +28,7 @@ Section, SectionsResponse, SimilarItemsRequest, + TrackedEvent, UnboxAIRequest, UnboxAIResponse, ) @@ -62,6 +63,7 @@ "CartItem", "Section", "SectionsResponse", + "TrackedEvent", ] diff --git a/src/behaviorgpt/types/http.py b/src/behaviorgpt/types/http.py index eb3da93..79fbde5 100644 --- a/src/behaviorgpt/types/http.py +++ b/src/behaviorgpt/types/http.py @@ -90,6 +90,17 @@ def validate_pagination(self) -> "UnboxAIRequest": return self +class TrackedEvent(BaseModel): + """An event as the API stored it: `event` is as the API stores it, and + `stream` is where it came from (a publishable key, "api", an import).""" + + user_id: str + timestamp: datetime + event: dict + store_id: Optional[str] = None + stream: Optional[str] = None + + class UnboxAIResponse(BaseModel): items: List[Item] offset: int diff --git a/tests/test_events.py b/tests/test_events.py new file mode 100644 index 0000000..3c47c3d --- /dev/null +++ b/tests/test_events.py @@ -0,0 +1,189 @@ +"""Tracked events: `events.track` records them, `events.list` reads them back.""" + +import asyncio +import json +from datetime import UTC, datetime + +import httpx +import pytest + +from behaviorgpt import AsyncUnboxAIClient, Order, Search, UnboxAIClient, User, View +from behaviorgpt.types import Item + +BASE = "http://sdk.test/v1" +MOMENT = datetime(2026, 10, 23, 12, 0, tzinfo=UTC) +LISTED = { + "events": [ + { + "user_id": "u1", + "timestamp": "2026-10-23T12:03:00Z", + "store_id": "cat", + "stream": "api", + "event": {"type": "viewItem", "product": "p3"}, + "domains": {"market": "US"}, + "timezone": "UTC", + } + ] +} + + +def handler_for(requests: list[httpx.Request]): + def handler(request: httpx.Request) -> httpx.Response: + requests.append(request) + if request.url.path == "/v1/events": + return httpx.Response(201, json={"id": "ev-1"}) + return httpx.Response(200, json=LISTED) + + return handler + + +def mock_client() -> tuple[UnboxAIClient, list[httpx.Request]]: + requests: list[httpx.Request] = [] + client = UnboxAIClient("US", api_key="test", base_url=BASE) + client._http_client._transport = httpx.MockTransport(handler_for(requests)) + return client, requests + + +def body(request: httpx.Request) -> dict: + return json.loads(request.content) + + +def test_track_posts_the_event_and_returns_its_id(): + client, requests = mock_client() + event_id = client.events.track("u1", View("sku-1"), timestamp=MOMENT) + + assert event_id == "ev-1" + assert requests[0].method == "POST" and requests[0].url.path == "/v1/events" + assert body(requests[0]) == { + "user_id": "u1", + "store_id": "retail_catalog", + "event": {"type": "viewItem", "product": "sku-1"}, + "domains": { + "market": "US", + "postal_code": "unk", + "gender": "unk", + "age_group": "unk", + }, + "timezone": "UTC", + "timestamp": "2026-10-23T12:00:00+00:00", + } + + +def test_track_sends_searches_and_the_catalog(): + client, requests = mock_client() + client.events.track("u1", Search("sofa"), catalog_id="cat") + sent = body(requests[0]) + assert sent["event"] == {"type": "search", "query": "sofa"} + assert sent["store_id"] == "cat" + assert datetime.fromisoformat(sent["timestamp"]).tzinfo is not None + + +def test_track_takes_the_catalog_an_item_names(): + client, requests = mock_client() + item = Item(id="r1", data={}, score=1.0, catalog_id="other") + client.events.track("u1", View(item)) + client.events.track("u1", Order([item])) + client.events.track("u1", View(item), catalog_id="cat") + assert [body(r)["store_id"] for r in requests] == ["other", "other", "cat"] + + +def test_track_refuses_an_order_over_several_catalogs(): + client, requests = mock_client() + a = Item(id="a", data={}, score=1.0, catalog_id="one") + b = Item(id="b", data={}, score=1.0, catalog_id="two") + with pytest.raises(ValueError, match="one Order per catalog"): + client.events.track("u1", Order([a, b])) + assert requests == [] + + +def test_track_sends_who_the_user_is(): + client, requests = mock_client() + shopper = User(gender="female", age=34, market="SE", timezone="Europe/Stockholm") + client.events.track("u1", View("a"), user=shopper) + sent = body(requests[0]) + assert sent["domains"] == { + "market": "SE", + "postal_code": "unk", + "gender": "female", + "age_group": "26-35", + } + assert sent["timezone"] == "Europe/Stockholm" + + +def test_track_sends_orders_as_the_api_stores_them(): + client, requests = mock_client() + client.events.track("u1", Order(["a", ("b", 2)], cart_id="cart-7")) + assert body(requests[0])["event"] == { + "type": "order", + "cart_id": "cart-7", + "items": [ + {"product": "a", "quantity": 1, "unit_selling_price": 0.0}, + {"product": "b", "quantity": 2, "unit_selling_price": 0.0}, + ], + } + + +def test_track_gives_an_order_a_cart_id(): + client, requests = mock_client() + client.events.track("u1", Order(["a"])) + assert body(requests[0])["event"]["cart_id"] + + +def test_track_refuses_an_order_without_items(): + client, requests = mock_client() + with pytest.raises(ValueError, match="items bought"): + client.events.track("u1", Order([])) + assert requests == [] + + +def test_track_refuses_a_naive_timestamp(): + client, _ = mock_client() + with pytest.raises(ValueError, match="timezone-aware"): + client.events.track("u1", View("a"), timestamp=datetime(2026, 10, 23)) + + +def test_list_reads_the_accounts_events(): + client, requests = mock_client() + events = client.events.list( + user_id="u1", limit=10, types=["viewItem", "search"], stream="api" + ) + + assert requests[0].method == "GET" and requests[0].url.path == "/v1/events/own" + assert requests[0].url.params.multi_items() == [ + ("limit", "10"), + ("user_id", "u1"), + ("type", "viewItem"), + ("type", "search"), + ("stream", "api"), + ] + assert [(ev.user_id, ev.stream, ev.event) for ev in events] == [ + ("u1", "api", {"type": "viewItem", "product": "p3"}) + ] + assert events[0].timestamp == datetime(2026, 10, 23, 12, 3, tzinfo=UTC) + + +def test_list_sends_only_the_limit_by_default(): + client, requests = mock_client() + client.events.list() + assert requests[0].url.params.multi_items() == [("limit", "50")] + + +@pytest.mark.parametrize("limit", [0, 201]) +def test_list_refuses_a_limit_out_of_range(limit): + client, _ = mock_client() + with pytest.raises(ValueError, match="limit"): + client.events.list(limit=limit) + + +def test_async_client_tracks_and_lists(): + requests: list[httpx.Request] = [] + + async def run() -> None: + async with AsyncUnboxAIClient("US", api_key="test", base_url=BASE) as client: + client._http_client._transport = httpx.MockTransport(handler_for(requests)) + assert await client.events.track("u1", View("a")) == "ev-1" + assert len(await client.events.list(stream="api")) == 1 + + asyncio.run(run()) + assert [r.url.path for r in requests] == ["/v1/events", "/v1/events/own"] + assert requests[1].url.params["stream"] == "api"