Notes

Reads by entity (list, iter) use V1; get, feed, iter_feed, list_replies, search and the writes (create, update, delete) use V2, which stores notes as HTML.

Bases: NoteSearchMixin

Service for managing notes.

V2 provides read-only access; use V1 for create/update/delete.

Source code in affinity/services/v1_only.py

class NoteService(NoteSearchMixin):
    """
    Service for managing notes.

    V2 provides read-only access; use V1 for create/update/delete.
    """

    def __init__(self, client: HTTPClient):
        self._client = client

    def list(
        self,
        *,
        person_id: PersonId | None = None,
        company_id: CompanyId | None = None,
        opportunity_id: OpportunityId | None = None,
        creator_id: UserId | None = None,
        page_size: int | None = None,
        page_token: str | None = None,
    ) -> PaginatedResponse[Note]:
        """
        Get notes filtered by entity or creator.

        Args:
            person_id: Filter notes associated with this person
            company_id: Filter notes associated with this company
            opportunity_id: Filter notes associated with this opportunity
            creator_id: Filter notes created by this user
            page_size: Number of results per page
            page_token: Pagination token from previous response

        Returns:
            PaginatedResponse with notes and next_page_token
        """
        params: dict[str, Any] = {}
        if person_id:
            params["person_id"] = int(person_id)
        if company_id:
            params["organization_id"] = int(company_id)
        if opportunity_id:
            params["opportunity_id"] = int(opportunity_id)
        if creator_id:
            params["creator_id"] = int(creator_id)
        if page_size:
            params["page_size"] = page_size
        if page_token:
            params["page_token"] = page_token

        data = self._client.get("/notes", params=params or None, v1=True)
        items = data.get("notes", data.get("data", []))
        if not isinstance(items, list):
            items = []
        return PaginatedResponse[Note](
            data=[Note.model_validate(n) for n in items],
            next_page_token=data.get("next_page_token") or data.get("nextPageToken"),
        )

    def create(self, data: NoteCreate) -> NoteV2:
        """Create a note (V2) and return it.

        Attach it to at least one person, company or opportunity, or reply to a note with
        ``parent_id``. V2 stores notes as HTML: plain text (the default) is escaped and wrapped
        in paragraphs; with ``type=NoteType.HTML`` the HTML is checked against what the API
        accepts (``<b>``/``<i>`` become ``<strong>``/``<em>``).
        """
        return NoteV2.model_validate(self._client.post("/notes", json=note_create_body(data)))

    def update(self, note_id: NoteId, data: NoteUpdate) -> NoteV2:
        """Change a note's content and/or attachments (V2) and return it as read back.

        Two requests: the update has no response body, so the note is read again with its
        attachments (``includes``). ``data.type`` says whether ``content`` is plain text (the
        default) or HTML. Notes with @mentions can't have their content changed (API 400).
        """
        try:
            self._client.post(f"/notes/{int(note_id)}", json=note_update_body(data))
        except AffinityValidationError as exc:
            if data.content is not None:
                note = None
                with contextlib.suppress(AffinityError):
                    note = self.get(note_id)
                if has_mentions(note):
                    raise mentions_error(exc) from exc
            raise
        return self.get(note_id, includes=True)

    def delete(self, note_id: NoteId) -> bool:
        """Delete a note you created (V2). Deleting a note also deletes its replies."""
        self._client.delete(f"/notes/{int(note_id)}")
        return True

    def iter(
        self,
        *,
        person_id: PersonId | None = None,
        company_id: CompanyId | None = None,
        opportunity_id: OpportunityId | None = None,
        creator_id: UserId | None = None,
        page_size: int | None = None,
    ) -> PageIterator[Note]:
        """
        Iterate through all notes with automatic pagination.

        Args:
            person_id: Filter notes associated with this person
            company_id: Filter notes associated with this company
            opportunity_id: Filter notes associated with this opportunity
            creator_id: Filter notes created by this user
            page_size: Number of results per page

        Returns:
            PageIterator that yields Note objects
        """

        def fetch_page(cursor: str | None) -> PaginatedResponse[Note]:
            return self.list(
                person_id=person_id,
                company_id=company_id,
                opportunity_id=opportunity_id,
                creator_id=creator_id,
                page_size=page_size,
                page_token=cursor,
            )

        return PageIterator(fetch_page)

    # =========================================================================
    # V2 notes (read): NoteV2 with creator, HTML content, interaction, transcript
    # =========================================================================

    def feed(
        self,
        *,
        creator_id: int | None = None,
        created_after: datetime | None = None,
        created_before: datetime | None = None,
        updated_after: datetime | None = None,
        includes: bool | Sequence[str] | None = None,
        limit: int | None = None,
        total_count: bool = False,
        cursor: str | None = None,
    ) -> PaginatedResponse[NoteV2]:
        """One page of notes across the organization as ``NoteV2`` (replies not included).

        Filters: ``creator_id``, ``created_after`` (inclusive) / ``created_before`` (exclusive),
        ``updated_after``, rounded outward to whole seconds. ``includes=True`` (or a list of
        ``"companiesPreview"``, ``"personsPreview"``, ``"opportunitiesPreview"``,
        ``"repliesCount"``) fills ``companies`` / ``persons`` / ``opportunities`` (the first
        ones, with totals) and ``replies_count``. ``limit`` is 1-100. :meth:`list` (V1) and its
        ``Note`` model are unchanged.
        """
        check_cursor_alone(
            cursor,
            creator_id,
            created_after,
            created_before,
            updated_after,
            includes,
            limit,
            total_count,
        )
        if cursor is not None:
            return page_of(NoteV2, self._client.get_url(cursor))
        params = note_params(
            creator_id=creator_id,
            created_after=created_after,
            created_before=created_before,
            updated_after=updated_after,
            limit=limit,
            total_count=total_count,
            includes=includes,
        )
        return page_of(NoteV2, self._client.get("/notes", params=params or None))

    def iter_feed(
        self,
        *,
        creator_id: int | None = None,
        created_after: datetime | None = None,
        created_before: datetime | None = None,
        updated_after: datetime | None = None,
        includes: bool | Sequence[str] | None = None,
        limit: int | None = None,
    ) -> Iterator[NoteV2]:
        """All notes matching the filters (every page); see :meth:`feed`."""

        def fetch_page(next_url: str | None) -> PaginatedResponse[NoteV2]:
            if next_url:
                return self.feed(cursor=next_url)
            return self.feed(
                creator_id=creator_id,
                created_after=created_after,
                created_before=created_before,
                updated_after=updated_after,
                includes=includes,
                limit=limit,
            )

        return PageIterator(fetch_page)

    def get(self, note_id: NoteId, *, includes: bool | Sequence[str] | None = None) -> NoteV2:
        """One note as ``NoteV2`` (``includes`` as in :meth:`feed`)."""
        params = [("includes", value) for value in note_includes(includes)]
        return NoteV2.model_validate(
            self._client.get(f"/notes/{int(note_id)}", params=params or None)
        )

    def list_replies(
        self,
        note_id: NoteId,
        *,
        creator_id: int | None = None,
        created_after: datetime | None = None,
        limit: int | None = None,
        total_count: bool = False,
        cursor: str | None = None,
    ) -> PaginatedResponse[NoteV2]:
        """One page of replies to a note (``NoteV2`` with ``parent``)."""
        check_cursor_alone(cursor, creator_id, created_after, limit, total_count)
        if cursor is not None:
            return page_of(NoteV2, self._client.get_url(cursor))
        params = note_params(
            creator_id=creator_id,
            created_after=created_after,
            created_before=None,
            updated_after=None,
            limit=limit,
            total_count=total_count,
        )
        return page_of(
            NoteV2,
            self._client.get(f"/notes/{int(note_id)}/replies", params=params or None),
        )

    def iter_replies(self, note_id: NoteId, *, limit: int | None = None) -> Iterator[NoteV2]:
        """All replies to a note (every page)."""

        def fetch_page(next_url: str | None) -> PaginatedResponse[NoteV2]:
            if next_url:
                return self.list_replies(note_id, cursor=next_url)
            return self.list_replies(note_id, limit=limit)

        return PageIterator(fetch_page)

    def list_attached_companies(
        self,
        note_id: NoteId,
        *,
        limit: int | None = None,
        total_count: bool = False,
        cursor: str | None = None,
    ) -> PaginatedResponse[Company]:
        """One page of the companies a note is attached to (V2). ``limit`` is 1-100."""
        return self._attached("companies", Company, note_id, limit, total_count, cursor)

    def list_attached_persons(
        self,
        note_id: NoteId,
        *,
        limit: int | None = None,
        total_count: bool = False,
        cursor: str | None = None,
    ) -> PaginatedResponse[Person]:
        """One page of the persons a note is attached to (V2). ``limit`` is 1-100."""
        return self._attached("persons", Person, note_id, limit, total_count, cursor)

    def list_attached_opportunities(
        self,
        note_id: NoteId,
        *,
        limit: int | None = None,
        total_count: bool = False,
        cursor: str | None = None,
    ) -> PaginatedResponse[Opportunity]:
        """One page of the opportunities a note is attached to (V2). ``limit`` is 1-100."""
        return self._attached("opportunities", Opportunity, note_id, limit, total_count, cursor)

    def iter_attached_companies(self, note_id: NoteId) -> Iterator[Company]:
        """Every company a note is attached to (all pages)."""
        return PageIterator(
            lambda c: self.list_attached_companies(note_id, cursor=c, limit=None if c else 100)
        )

    def iter_attached_persons(self, note_id: NoteId) -> Iterator[Person]:
        """Every person a note is attached to (all pages)."""
        return PageIterator(
            lambda c: self.list_attached_persons(note_id, cursor=c, limit=None if c else 100)
        )

    def iter_attached_opportunities(self, note_id: NoteId) -> Iterator[Opportunity]:
        """Every opportunity a note is attached to (all pages)."""
        return PageIterator(
            lambda c: self.list_attached_opportunities(note_id, cursor=c, limit=None if c else 100)
        )

    def _attached(
        self,
        kind: str,
        model: type[Any],
        note_id: NoteId,
        limit: int | None,
        total_count: bool,
        cursor: str | None,
    ) -> PaginatedResponse[Any]:
        check_cursor_alone(cursor, limit, total_count)
        if cursor is not None:
            return page_of(model, self._client.get_url(cursor))
        return page_of(
            model,
            self._client.get(
                f"/notes/{int(note_id)}/attached-{kind}",
                params=_attached_params(limit, total_count) or None,
            ),
        )

create(data: NoteCreate) -> NoteV2

Create a note (V2) and return it.

Attach it to at least one person, company or opportunity, or reply to a note with parent_id. V2 stores notes as HTML: plain text (the default) is escaped and wrapped in paragraphs; with type=NoteType.HTML the HTML is checked against what the API accepts (<b>/<i> become <strong>/<em>).

Source code in affinity/services/v1_only.py

def create(self, data: NoteCreate) -> NoteV2:
    """Create a note (V2) and return it.

    Attach it to at least one person, company or opportunity, or reply to a note with
    ``parent_id``. V2 stores notes as HTML: plain text (the default) is escaped and wrapped
    in paragraphs; with ``type=NoteType.HTML`` the HTML is checked against what the API
    accepts (``<b>``/``<i>`` become ``<strong>``/``<em>``).
    """
    return NoteV2.model_validate(self._client.post("/notes", json=note_create_body(data)))

delete(note_id: NoteId) -> bool

Delete a note you created (V2). Deleting a note also deletes its replies.

Source code in affinity/services/v1_only.py

def delete(self, note_id: NoteId) -> bool:
    """Delete a note you created (V2). Deleting a note also deletes its replies."""
    self._client.delete(f"/notes/{int(note_id)}")
    return True

feed(*, creator_id: int | None = None, created_after: datetime | None = None, created_before: datetime | None = None, updated_after: datetime | None = None, includes: bool | Sequence[str] | None = None, limit: int | None = None, total_count: bool = False, cursor: str | None = None) -> PaginatedResponse[NoteV2]

One page of notes across the organization as NoteV2 (replies not included).

Filters: creator_id, created_after (inclusive) / created_before (exclusive), updated_after, rounded outward to whole seconds. includes=True (or a list of "companiesPreview", "personsPreview", "opportunitiesPreview", "repliesCount") fills companies / persons / opportunities (the first ones, with totals) and replies_count. limit is 1-100. :meth:list (V1) and its Note model are unchanged.

Source code in affinity/services/v1_only.py

def feed(
    self,
    *,
    creator_id: int | None = None,
    created_after: datetime | None = None,
    created_before: datetime | None = None,
    updated_after: datetime | None = None,
    includes: bool | Sequence[str] | None = None,
    limit: int | None = None,
    total_count: bool = False,
    cursor: str | None = None,
) -> PaginatedResponse[NoteV2]:
    """One page of notes across the organization as ``NoteV2`` (replies not included).

    Filters: ``creator_id``, ``created_after`` (inclusive) / ``created_before`` (exclusive),
    ``updated_after``, rounded outward to whole seconds. ``includes=True`` (or a list of
    ``"companiesPreview"``, ``"personsPreview"``, ``"opportunitiesPreview"``,
    ``"repliesCount"``) fills ``companies`` / ``persons`` / ``opportunities`` (the first
    ones, with totals) and ``replies_count``. ``limit`` is 1-100. :meth:`list` (V1) and its
    ``Note`` model are unchanged.
    """
    check_cursor_alone(
        cursor,
        creator_id,
        created_after,
        created_before,
        updated_after,
        includes,
        limit,
        total_count,
    )
    if cursor is not None:
        return page_of(NoteV2, self._client.get_url(cursor))
    params = note_params(
        creator_id=creator_id,
        created_after=created_after,
        created_before=created_before,
        updated_after=updated_after,
        limit=limit,
        total_count=total_count,
        includes=includes,
    )
    return page_of(NoteV2, self._client.get("/notes", params=params or None))

get(note_id: NoteId, *, includes: bool | Sequence[str] | None = None) -> NoteV2

One note as NoteV2 (includes as in :meth:feed).

Source code in affinity/services/v1_only.py

def get(self, note_id: NoteId, *, includes: bool | Sequence[str] | None = None) -> NoteV2:
    """One note as ``NoteV2`` (``includes`` as in :meth:`feed`)."""
    params = [("includes", value) for value in note_includes(includes)]
    return NoteV2.model_validate(
        self._client.get(f"/notes/{int(note_id)}", params=params or None)
    )

iter(*, person_id: PersonId | None = None, company_id: CompanyId | None = None, opportunity_id: OpportunityId | None = None, creator_id: UserId | None = None, page_size: int | None = None) -> PageIterator[Note]

Iterate through all notes with automatic pagination.

Parameters:

Name	Type	Description	Default
person_id	PersonId | None	Filter notes associated with this person	None
company_id	CompanyId | None	Filter notes associated with this company	None
opportunity_id	OpportunityId | None	Filter notes associated with this opportunity	None
creator_id	UserId | None	Filter notes created by this user	None
page_size	int | None	Number of results per page	None

Returns:

Type	Description
PageIterator[Note]	PageIterator that yields Note objects

Source code in affinity/services/v1_only.py

def iter(
    self,
    *,
    person_id: PersonId | None = None,
    company_id: CompanyId | None = None,
    opportunity_id: OpportunityId | None = None,
    creator_id: UserId | None = None,
    page_size: int | None = None,
) -> PageIterator[Note]:
    """
    Iterate through all notes with automatic pagination.

    Args:
        person_id: Filter notes associated with this person
        company_id: Filter notes associated with this company
        opportunity_id: Filter notes associated with this opportunity
        creator_id: Filter notes created by this user
        page_size: Number of results per page

    Returns:
        PageIterator that yields Note objects
    """

    def fetch_page(cursor: str | None) -> PaginatedResponse[Note]:
        return self.list(
            person_id=person_id,
            company_id=company_id,
            opportunity_id=opportunity_id,
            creator_id=creator_id,
            page_size=page_size,
            page_token=cursor,
        )

    return PageIterator(fetch_page)

iter_attached_companies(note_id: NoteId) -> Iterator[Company]

Every company a note is attached to (all pages).

Source code in affinity/services/v1_only.py

def iter_attached_companies(self, note_id: NoteId) -> Iterator[Company]:
    """Every company a note is attached to (all pages)."""
    return PageIterator(
        lambda c: self.list_attached_companies(note_id, cursor=c, limit=None if c else 100)
    )

iter_attached_opportunities(note_id: NoteId) -> Iterator[Opportunity]

Every opportunity a note is attached to (all pages).

Source code in affinity/services/v1_only.py

def iter_attached_opportunities(self, note_id: NoteId) -> Iterator[Opportunity]:
    """Every opportunity a note is attached to (all pages)."""
    return PageIterator(
        lambda c: self.list_attached_opportunities(note_id, cursor=c, limit=None if c else 100)
    )

iter_attached_persons(note_id: NoteId) -> Iterator[Person]

Every person a note is attached to (all pages).

Source code in affinity/services/v1_only.py

def iter_attached_persons(self, note_id: NoteId) -> Iterator[Person]:
    """Every person a note is attached to (all pages)."""
    return PageIterator(
        lambda c: self.list_attached_persons(note_id, cursor=c, limit=None if c else 100)
    )

iter_feed(*, creator_id: int | None = None, created_after: datetime | None = None, created_before: datetime | None = None, updated_after: datetime | None = None, includes: bool | Sequence[str] | None = None, limit: int | None = None) -> Iterator[NoteV2]

All notes matching the filters (every page); see :meth:feed.

Source code in affinity/services/v1_only.py

def iter_feed(
    self,
    *,
    creator_id: int | None = None,
    created_after: datetime | None = None,
    created_before: datetime | None = None,
    updated_after: datetime | None = None,
    includes: bool | Sequence[str] | None = None,
    limit: int | None = None,
) -> Iterator[NoteV2]:
    """All notes matching the filters (every page); see :meth:`feed`."""

    def fetch_page(next_url: str | None) -> PaginatedResponse[NoteV2]:
        if next_url:
            return self.feed(cursor=next_url)
        return self.feed(
            creator_id=creator_id,
            created_after=created_after,
            created_before=created_before,
            updated_after=updated_after,
            includes=includes,
            limit=limit,
        )

    return PageIterator(fetch_page)

iter_replies(note_id: NoteId, *, limit: int | None = None) -> Iterator[NoteV2]

All replies to a note (every page).

Source code in affinity/services/v1_only.py

def iter_replies(self, note_id: NoteId, *, limit: int | None = None) -> Iterator[NoteV2]:
    """All replies to a note (every page)."""

    def fetch_page(next_url: str | None) -> PaginatedResponse[NoteV2]:
        if next_url:
            return self.list_replies(note_id, cursor=next_url)
        return self.list_replies(note_id, limit=limit)

    return PageIterator(fetch_page)

list(*, person_id: PersonId | None = None, company_id: CompanyId | None = None, opportunity_id: OpportunityId | None = None, creator_id: UserId | None = None, page_size: int | None = None, page_token: str | None = None) -> PaginatedResponse[Note]

Get notes filtered by entity or creator.

Parameters:

Name	Type	Description	Default
person_id	PersonId | None	Filter notes associated with this person	None
company_id	CompanyId | None	Filter notes associated with this company	None
opportunity_id	OpportunityId | None	Filter notes associated with this opportunity	None
creator_id	UserId | None	Filter notes created by this user	None
page_size	int | None	Number of results per page	None
page_token	str | None	Pagination token from previous response	None

Returns:

Type	Description
PaginatedResponse[Note]	PaginatedResponse with notes and next_page_token

Source code in affinity/services/v1_only.py

def list(
    self,
    *,
    person_id: PersonId | None = None,
    company_id: CompanyId | None = None,
    opportunity_id: OpportunityId | None = None,
    creator_id: UserId | None = None,
    page_size: int | None = None,
    page_token: str | None = None,
) -> PaginatedResponse[Note]:
    """
    Get notes filtered by entity or creator.

    Args:
        person_id: Filter notes associated with this person
        company_id: Filter notes associated with this company
        opportunity_id: Filter notes associated with this opportunity
        creator_id: Filter notes created by this user
        page_size: Number of results per page
        page_token: Pagination token from previous response

    Returns:
        PaginatedResponse with notes and next_page_token
    """
    params: dict[str, Any] = {}
    if person_id:
        params["person_id"] = int(person_id)
    if company_id:
        params["organization_id"] = int(company_id)
    if opportunity_id:
        params["opportunity_id"] = int(opportunity_id)
    if creator_id:
        params["creator_id"] = int(creator_id)
    if page_size:
        params["page_size"] = page_size
    if page_token:
        params["page_token"] = page_token

    data = self._client.get("/notes", params=params or None, v1=True)
    items = data.get("notes", data.get("data", []))
    if not isinstance(items, list):
        items = []
    return PaginatedResponse[Note](
        data=[Note.model_validate(n) for n in items],
        next_page_token=data.get("next_page_token") or data.get("nextPageToken"),
    )

list_attached_companies(note_id: NoteId, *, limit: int | None = None, total_count: bool = False, cursor: str | None = None) -> PaginatedResponse[Company]

One page of the companies a note is attached to (V2). limit is 1-100.

Source code in affinity/services/v1_only.py

def list_attached_companies(
    self,
    note_id: NoteId,
    *,
    limit: int | None = None,
    total_count: bool = False,
    cursor: str | None = None,
) -> PaginatedResponse[Company]:
    """One page of the companies a note is attached to (V2). ``limit`` is 1-100."""
    return self._attached("companies", Company, note_id, limit, total_count, cursor)

list_attached_opportunities(note_id: NoteId, *, limit: int | None = None, total_count: bool = False, cursor: str | None = None) -> PaginatedResponse[Opportunity]

One page of the opportunities a note is attached to (V2). limit is 1-100.

Source code in affinity/services/v1_only.py

def list_attached_opportunities(
    self,
    note_id: NoteId,
    *,
    limit: int | None = None,
    total_count: bool = False,
    cursor: str | None = None,
) -> PaginatedResponse[Opportunity]:
    """One page of the opportunities a note is attached to (V2). ``limit`` is 1-100."""
    return self._attached("opportunities", Opportunity, note_id, limit, total_count, cursor)

list_attached_persons(note_id: NoteId, *, limit: int | None = None, total_count: bool = False, cursor: str | None = None) -> PaginatedResponse[Person]

One page of the persons a note is attached to (V2). limit is 1-100.

Source code in affinity/services/v1_only.py

def list_attached_persons(
    self,
    note_id: NoteId,
    *,
    limit: int | None = None,
    total_count: bool = False,
    cursor: str | None = None,
) -> PaginatedResponse[Person]:
    """One page of the persons a note is attached to (V2). ``limit`` is 1-100."""
    return self._attached("persons", Person, note_id, limit, total_count, cursor)

list_replies(note_id: NoteId, *, creator_id: int | None = None, created_after: datetime | None = None, limit: int | None = None, total_count: bool = False, cursor: str | None = None) -> PaginatedResponse[NoteV2]

One page of replies to a note (NoteV2 with parent).

Source code in affinity/services/v1_only.py

def list_replies(
    self,
    note_id: NoteId,
    *,
    creator_id: int | None = None,
    created_after: datetime | None = None,
    limit: int | None = None,
    total_count: bool = False,
    cursor: str | None = None,
) -> PaginatedResponse[NoteV2]:
    """One page of replies to a note (``NoteV2`` with ``parent``)."""
    check_cursor_alone(cursor, creator_id, created_after, limit, total_count)
    if cursor is not None:
        return page_of(NoteV2, self._client.get_url(cursor))
    params = note_params(
        creator_id=creator_id,
        created_after=created_after,
        created_before=None,
        updated_after=None,
        limit=limit,
        total_count=total_count,
    )
    return page_of(
        NoteV2,
        self._client.get(f"/notes/{int(note_id)}/replies", params=params or None),
    )

update(note_id: NoteId, data: NoteUpdate) -> NoteV2

Change a note's content and/or attachments (V2) and return it as read back.

Two requests: the update has no response body, so the note is read again with its attachments (includes). data.type says whether content is plain text (the default) or HTML. Notes with @mentions can't have their content changed (API 400).

Source code in affinity/services/v1_only.py

def update(self, note_id: NoteId, data: NoteUpdate) -> NoteV2:
    """Change a note's content and/or attachments (V2) and return it as read back.

    Two requests: the update has no response body, so the note is read again with its
    attachments (``includes``). ``data.type`` says whether ``content`` is plain text (the
    default) or HTML. Notes with @mentions can't have their content changed (API 400).
    """
    try:
        self._client.post(f"/notes/{int(note_id)}", json=note_update_body(data))
    except AffinityValidationError as exc:
        if data.content is not None:
            note = None
            with contextlib.suppress(AffinityError):
                note = self.get(note_id)
            if has_mentions(note):
                raise mentions_error(exc) from exc
        raise
    return self.get(note_id, includes=True)