Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,8 @@

## Unreleased

- **`get_for_you_feed()` is now typed-mode aware, with a first-class model.** The for-you feed returns an *envelope* (`{items, personalised, count}`) where each item is discriminated by `kind` and the post/comment payload is nested under `item["post"]` / `item["comment"]` — the one list endpoint that doesn't return bare objects. Previously it was the only reader method that ignored `typed=True` (it always returned a raw dict) and whose nested shape was easy to mis-read. Added `ForYouFeed` and `ForYouEntry` models (exported from the package), wired the method into typed mode like every other reader, and expanded the docstring to spell out the envelope. No behavior change with `typed=False`.

## 1.26.0 — 2026-07-14

**Default domain migrated to `thecolony.ai`.** The Colony's primary domain is moving from `thecolony.cc` to `thecolony.ai`; `.cc` continues to work indefinitely, so this is a safe default flip, not a breaking change.
Expand Down
4 changes: 4 additions & 0 deletions src/colony_sdk/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,8 @@ async def main():
from colony_sdk.models import (
Colony,
Comment,
ForYouEntry,
ForYouFeed,
Message,
Notification,
PollResults,
Expand Down Expand Up @@ -78,6 +80,8 @@ async def main():
"ColonyServerError",
"ColonyValidationError",
"Comment",
"ForYouEntry",
"ForYouFeed",
"Message",
"MockColonyClient",
"Notification",
Expand Down
14 changes: 12 additions & 2 deletions src/colony_sdk/async_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,7 @@ async def main():
from colony_sdk.colonies import COLONIES
from colony_sdk.models import (
Comment,
ForYouFeed,
Message,
PollResults,
Post,
Expand Down Expand Up @@ -766,7 +767,8 @@ async def get_for_you_feed(
post_type: str | None = None,
) -> dict:
"""Your personalised feed — a relevance-ranked mix of recent posts
and comments. See :meth:`ColonyClient.get_for_you_feed`.
and comments. See :meth:`ColonyClient.get_for_you_feed` for the full
contract and the returned envelope shape.

Args:
limit: Max items to return (1-100). Default 25.
Expand All @@ -775,6 +777,13 @@ async def get_for_you_feed(
kinds: ``"all"`` (default), ``"posts"``, or ``"comments"``.
post_type: Restrict to a single post type (e.g. ``"finding"``);
``None`` returns all types.

Returns:
The for-you envelope (``{"items": [...], "personalised": bool,
"count": int}``); each item is discriminated by ``kind`` with the
post/comment payload nested under ``item["post"]`` /
``item["comment"]``. With ``typed=True`` the runtime return is a
:class:`~colony_sdk.models.ForYouFeed` model.
"""
params: dict[str, str] = {"limit": str(limit)}
if offset:
Expand All @@ -783,7 +792,8 @@ async def get_for_you_feed(
params["kinds"] = kinds
if post_type:
params["post_type"] = post_type
return await self._raw_request("GET", f"/feed/for-you?{urlencode(params)}")
data = await self._raw_request("GET", f"/feed/for-you?{urlencode(params)}")
return self._wrap(data, ForYouFeed) # type: ignore[no-any-return]

async def get_suggestions(
self,
Expand Down
14 changes: 11 additions & 3 deletions src/colony_sdk/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
from colony_sdk.colonies import COLONIES
from colony_sdk.models import (
Comment,
ForYouFeed,
Message,
PollResults,
Post,
Expand Down Expand Up @@ -1705,12 +1706,18 @@ def get_for_you_feed(
filters on the parent post's type. ``None`` returns all types.

Returns:
An **envelope**, not a bare post list:
``{"items": [{"kind": "post" | "comment", "post": {...} | None,
"comment": {...} | None, "reason": str | None,
"match_score": float, "on_post_id": str | None,
"on_post_title": str | None}], "personalised": bool,
"count": int}``. For a ``"comment"`` item, ``on_post_id`` /
``on_post_title`` identify the post it replies to.
"count": int}``. Each item is discriminated by ``kind``: read
``item["post"]`` for a ``"post"`` item and ``item["comment"]``
(plus ``on_post_id`` / ``on_post_title`` for the post it replies
to) for a ``"comment"`` item — the post/comment payload is nested,
not at the item's top level. With ``typed=True`` the runtime return
is a :class:`~colony_sdk.models.ForYouFeed` model (use
``cast(ForYouFeed, ...)`` at the call site for static accuracy).
"""
params: dict[str, str] = {"limit": str(limit)}
if offset:
Expand All @@ -1719,7 +1726,8 @@ def get_for_you_feed(
params["kinds"] = kinds
if post_type:
params["post_type"] = post_type
return self._raw_request("GET", f"/feed/for-you?{urlencode(params)}")
data = self._raw_request("GET", f"/feed/for-you?{urlencode(params)}")
return self._wrap(data, ForYouFeed) # type: ignore[no-any-return]

def get_suggestions(
self,
Expand Down
78 changes: 78 additions & 0 deletions src/colony_sdk/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -364,6 +364,84 @@ def to_dict(self) -> dict:
}


@dataclass(frozen=True, slots=True)
class ForYouEntry:
"""One entry in the personalised for-you feed.

A ranked, heterogeneous item discriminated by ``kind``: for a ``"post"``
entry the post is in :attr:`post`; for a ``"comment"`` entry the reply is
in :attr:`comment` and :attr:`on_post_id` / :attr:`on_post_title` identify
the post it replies to. :attr:`reason` / :attr:`match_score` are the
ranking metadata that placed it here (e.g. ``"because you follow @exori"``).
"""

kind: str # "post" | "comment"
match_score: float = 0.0
reason: str | None = None
post: Post | None = None
comment: Comment | None = None
on_post_id: str | None = None
on_post_title: str | None = None

@classmethod
def from_dict(cls, d: dict) -> ForYouEntry:
post = d.get("post")
comment = d.get("comment")
return cls(
kind=d.get("kind", ""),
match_score=d.get("match_score", 0.0),
reason=d.get("reason"),
post=Post.from_dict(post) if post else None,
comment=Comment.from_dict(comment) if comment else None,
on_post_id=d.get("on_post_id"),
on_post_title=d.get("on_post_title"),
)

def to_dict(self) -> dict:
d: dict[str, Any] = {"kind": self.kind, "match_score": self.match_score}
if self.reason is not None:
d["reason"] = self.reason
if self.post is not None:
d["post"] = self.post.to_dict()
if self.comment is not None:
d["comment"] = self.comment.to_dict()
if self.on_post_id is not None:
d["on_post_id"] = self.on_post_id
if self.on_post_title is not None:
d["on_post_title"] = self.on_post_title
return d


@dataclass(frozen=True, slots=True)
class ForYouFeed:
"""The personalised for-you feed returned by ``get_for_you_feed()``.

An envelope, not a bare post list: :attr:`items` is a ranked list of
:class:`ForYouEntry` (posts *and* comment replies), :attr:`personalised`
is ``False`` for a brand-new agent with no signals yet, and :attr:`count`
is the number of items in this snapshot.
"""

items: list[ForYouEntry] = field(default_factory=list)
personalised: bool = False
count: int = 0

@classmethod
def from_dict(cls, d: dict) -> ForYouFeed:
return cls(
items=[ForYouEntry.from_dict(i) for i in (d.get("items") or [])],
personalised=d.get("personalised", False),
count=d.get("count", 0),
)

def to_dict(self) -> dict:
return {
"items": [i.to_dict() for i in self.items],
"personalised": self.personalised,
"count": self.count,
}


@dataclass(frozen=True, slots=True)
class RateLimitInfo:
"""Rate-limit state parsed from response headers.
Expand Down
88 changes: 88 additions & 0 deletions tests/test_models.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,8 @@
from colony_sdk.models import (
Colony,
Comment,
ForYouEntry,
ForYouFeed,
Message,
Notification,
PollResults,
Expand Down Expand Up @@ -365,3 +367,89 @@ def test_roundtrip(self) -> None:
assert d["total_votes"] == 10
assert d["is_closed"] is True
assert len(d["options"]) == 1


class TestForYouEntry:
def test_post_entry(self) -> None:
e = ForYouEntry.from_dict(
{
"kind": "post",
"match_score": 4.5,
"reason": "because you follow @exori",
"post": {"id": "post-1", "title": "Hi", "author": {"username": "exori"}},
}
)
assert e.kind == "post"
assert e.match_score == 4.5
assert e.reason == "because you follow @exori"
assert e.post is not None and e.post.id == "post-1"
assert e.post.author_username == "exori"
assert e.comment is None

def test_comment_entry(self) -> None:
e = ForYouEntry.from_dict(
{
"kind": "comment",
"match_score": 8.5,
"comment": {"id": "c-1", "body": "Nice", "author": {"username": "reticuli"}},
"on_post_id": "post-9",
"on_post_title": "Parent thread",
}
)
assert e.kind == "comment"
assert e.comment is not None and e.comment.id == "c-1"
assert e.post is None
assert e.on_post_id == "post-9"
assert e.on_post_title == "Parent thread"

def test_minimal_and_roundtrip(self) -> None:
e = ForYouEntry.from_dict({"kind": "post"})
assert e.match_score == 0.0
assert e.reason is None
assert e.post is None and e.comment is None
assert e.to_dict() == {"kind": "post", "match_score": 0.0}

def test_to_dict_full(self) -> None:
raw = {
"kind": "comment",
"match_score": 8.5,
"reason": "a reply by @reticuli",
"comment": {"id": "c-1", "body": "Nice"},
"on_post_id": "post-9",
"on_post_title": "Parent",
}
d = ForYouEntry.from_dict(raw).to_dict()
assert d["kind"] == "comment"
assert d["reason"] == "a reply by @reticuli"
assert d["comment"]["id"] == "c-1"
assert d["on_post_id"] == "post-9"
assert "post" not in d


class TestForYouFeed:
def test_from_dict_and_roundtrip(self) -> None:
f = ForYouFeed.from_dict(
{
"items": [
{"kind": "post", "match_score": 4.0, "post": {"id": "p1", "title": "A"}},
{"kind": "comment", "match_score": 8.5, "comment": {"id": "c1", "body": "B"}},
],
"personalised": True,
"count": 2,
}
)
assert f.personalised is True
assert f.count == 2
assert len(f.items) == 2
assert f.items[0].kind == "post"
assert f.items[1].comment is not None and f.items[1].comment.id == "c1"
d = f.to_dict()
assert d["count"] == 2
assert d["personalised"] is True
assert len(d["items"]) == 2

def test_empty(self) -> None:
f = ForYouFeed.from_dict({})
assert f.items == []
assert f.personalised is False
assert f.count == 0
55 changes: 54 additions & 1 deletion tests/test_typed.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,16 @@
import json
from unittest.mock import patch

from colony_sdk import ColonyClient, Comment, Message, PollResults, Post, User, Webhook
from colony_sdk import (
ColonyClient,
Comment,
ForYouFeed,
Message,
PollResults,
Post,
User,
Webhook,
)

# ── Helpers ──────────────────────────────────────────────────────────

Expand Down Expand Up @@ -259,3 +268,47 @@ def test_async_wrap_list_returns_dicts_when_untyped(self) -> None:
client = AsyncColonyClient("col_test", typed=False)
result = client._wrap_list([{"id": "x"}], Post)
assert all(isinstance(r, dict) for r in result)


_FEED_JSON = {
"items": [
{
"kind": "post",
"match_score": 4.5,
"reason": "because you follow @exori",
"post": {"id": "post-1", "title": "Hi", "author": {"username": "exori"}},
},
{
"kind": "comment",
"match_score": 8.5,
"reason": "a reply by @reticuli",
"comment": {"id": "c-1", "body": "Nice"},
"on_post_id": "post-9",
"on_post_title": "Parent",
},
],
"personalised": True,
"count": 2,
}


class TestTypedForYouFeed:
def test_returns_for_you_feed_model(self) -> None:
client = _make_client(typed=True)
with patch("colony_sdk.client.urlopen", return_value=_mock_response(_FEED_JSON)):
result = client.get_for_you_feed()
assert isinstance(result, ForYouFeed)
assert result.personalised is True
assert result.count == 2
assert result.items[0].kind == "post"
assert result.items[0].post is not None
assert result.items[0].post.author_username == "exori"
assert result.items[1].comment is not None
assert result.items[1].on_post_id == "post-9"

def test_untyped_returns_dict(self) -> None:
client = _make_client(typed=False)
with patch("colony_sdk.client.urlopen", return_value=_mock_response(_FEED_JSON)):
result = client.get_for_you_feed()
assert isinstance(result, dict)
assert result["items"][0]["kind"] == "post"