Interactions

list, get and the writes (create, update, delete) use V1. The per-type reads (list_emails, list_meetings, get_meeting, list_calls, list_chat_messages and their iter_* forms) use V2.

Service for managing interactions (meetings, calls, emails, chats).

V2 provides read-only metadata; V1 supports full CRUD.

Source code in affinity/services/v1_only.py

class InteractionService:
    """
    Service for managing interactions (meetings, calls, emails, chats).

    V2 provides read-only metadata; V1 supports full CRUD.
    """

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

    def list(
        self,
        *,
        type: InteractionType | None = None,
        start_time: datetime | None = None,
        end_time: datetime | None = None,
        person_id: PersonId | None = None,
        company_id: CompanyId | None = None,
        opportunity_id: OpportunityId | None = None,
        page_size: int | None = None,
        page_token: str | None = None,
    ) -> PaginatedResponse[Interaction]:
        """
        Get interactions with filtering.

        All parameters are validated before calling the API:
        - type, start_time, end_time are required
        - At least one entity ID (person_id, company_id, or opportunity_id)
        - Date range must be <= 365 days
        - start_time must be before end_time

        For ranges exceeding 365 days, use iter() which automatically chunks.

        Returns V1 paginated response with `data` and `next_page_token`.

        Raises:
            ValueError: If required parameters are missing or invalid.
        """
        if type is None:
            raise ValueError(
                "type is required for interactions API. "
                "Use InteractionType.EMAIL, MEETING, CALL, or CHAT_MESSAGE."
            )
        if start_time is None:
            raise ValueError(
                "start_time is required for interactions API. "
                "Use iter() for automatic date range handling."
            )
        if end_time is None:
            raise ValueError(
                "end_time is required for interactions API. "
                "Use iter() for automatic date range handling."
            )
        if (start_time.tzinfo is None) != (end_time.tzinfo is None):
            raise ValueError(
                "start_time and end_time must both be timezone-aware or both naive. "
                "Recommended: use timezone-aware datetimes (e.g., datetime.now(timezone.utc))."
            )
        if end_time <= start_time:
            raise ValueError("start_time must be before end_time.")
        if (end_time - start_time) > timedelta(days=_MAX_INTERACTION_RANGE_DAYS):
            raise ValueError(
                f"Date range exceeds {_MAX_INTERACTION_RANGE_DAYS} days. "
                f"Use iter() which automatically chunks large ranges."
            )
        if not any(x is not None for x in (person_id, company_id, opportunity_id)):
            raise ValueError(
                "At least one entity filter is required: person_id, company_id, or opportunity_id."
            )
        params: dict[str, Any] = {"type": int(type)}
        params["start_time"] = start_time.isoformat()
        params["end_time"] = end_time.isoformat()
        if person_id is not None:
            params["person_id"] = int(person_id)
        if company_id is not None:
            params["organization_id"] = int(company_id)
        if opportunity_id is not None:
            params["opportunity_id"] = int(opportunity_id)
        if page_size is not None:
            params["page_size"] = page_size
        if page_token is not None:
            params["page_token"] = page_token

        data = self._client.get("/interactions", params=params or None, v1=True)
        items: Any = None
        if int(type) in (int(InteractionType.MEETING), int(InteractionType.CALL)):
            items = data.get("events")
        elif int(type) == int(InteractionType.CHAT_MESSAGE):
            items = data.get("chat_messages")
        elif int(type) == int(InteractionType.EMAIL):
            items = data.get("emails")

        if items is None:
            items = (
                data.get("interactions")
                or data.get("events")
                or data.get("emails")
                or data.get("chat_messages")
                or data.get("data", [])
            )
        if not isinstance(items, list):
            items = []
        return PaginatedResponse[Interaction](
            data=[Interaction.model_validate(i) for i in items],
            next_page_token=data.get("next_page_token") or data.get("nextPageToken"),
        )

    def get(self, interaction_id: InteractionId, type: InteractionType) -> Interaction:
        """Get a single interaction by ID and type."""
        data = self._client.get(
            f"/interactions/{int(interaction_id)}",
            params={"type": int(type)},
            v1=True,
        )
        return Interaction.model_validate(data)

    def create(self, data: InteractionCreate) -> Interaction:
        """Create a new interaction (manually logged)."""
        payload = data.model_dump(by_alias=True, mode="python", exclude_none=True)
        _coerce_isoformat(payload, ("date",))

        result = self._client.post("/interactions", json=payload, v1=True)
        return Interaction.model_validate(result)

    def update(
        self,
        interaction_id: InteractionId,
        type: InteractionType,
        data: InteractionUpdate,
    ) -> Interaction:
        """Update an interaction."""
        payload = data.model_dump(
            by_alias=True,
            mode="python",
            exclude_unset=True,
            exclude_none=True,
        )
        payload["type"] = int(type)
        _coerce_isoformat(payload, ("date",))

        result = self._client.put(
            f"/interactions/{int(interaction_id)}",
            json=payload,
            v1=True,
        )
        return Interaction.model_validate(result)

    def delete(self, interaction_id: InteractionId, type: InteractionType) -> bool:
        """Delete an interaction."""
        result = self._client.delete(
            f"/interactions/{int(interaction_id)}",
            params={"type": int(type)},
            v1=True,
        )
        return bool(result.get("success", True))

    def iter(
        self,
        *,
        type: InteractionType | None = None,
        start_time: datetime | None = None,
        end_time: datetime | None = None,
        person_id: PersonId | None = None,
        company_id: CompanyId | None = None,
        opportunity_id: OpportunityId | None = None,
        page_size: int | None = None,
    ) -> PageIterator[Interaction]:
        """
        Iterate through all interactions with automatic pagination and date chunking.

        Automatically splits date ranges exceeding 365 days into chunks
        and bridges them with synthetic cursors for seamless iteration.

        Args:
            type: Interaction type (required).
            start_time: Start of date range (required).
            end_time: End of date range (defaults to now if not provided).
            person_id: Filter by person.
            company_id: Filter by company.
            opportunity_id: Filter by opportunity.
            page_size: Page size for API calls.

        Returns:
            PageIterator that yields Interaction objects
        """
        if type is None:
            raise ValueError(
                "type is required for interactions API. "
                "Use InteractionType.EMAIL, MEETING, CALL, or CHAT_MESSAGE."
            )
        if start_time is None:
            raise ValueError("start_time is required for interactions API.")
        if not any(x is not None for x in (person_id, company_id, opportunity_id)):
            raise ValueError(
                "At least one entity filter is required: person_id, company_id, or opportunity_id."
            )
        resolved_end = end_time if end_time is not None else datetime.now(timezone.utc)
        if (start_time.tzinfo is None) != (resolved_end.tzinfo is None):
            raise ValueError(
                "start_time and end_time must both be timezone-aware or both naive. "
                "Recommended: use timezone-aware datetimes (e.g., datetime.now(timezone.utc))."
            )
        if resolved_end <= start_time:
            raise ValueError("start_time must be before end_time.")
        chunks = _chunk_date_range(start_time, resolved_end)
        chunk_index = 0
        chunk_sentinel = f"__chunk_{uuid.uuid4().hex}__"

        def fetch_page(cursor: str | None) -> PaginatedResponse[Interaction]:
            nonlocal chunk_index
            if cursor == chunk_sentinel:
                chunk_index += 1
                cursor = None
            if chunk_index >= len(chunks):
                return PaginatedResponse[Interaction](data=[])
            c_start, c_end = chunks[chunk_index]
            response = self.list(
                type=type,
                start_time=c_start,
                end_time=c_end,
                person_id=person_id,
                company_id=company_id,
                opportunity_id=opportunity_id,
                page_size=page_size,
                page_token=cursor,
            )
            if response.next_cursor is None and chunk_index < len(chunks) - 1:
                response.next_page_token = chunk_sentinel
            return response

        return PageIterator(fetch_page)

    def list_emails(
        self,
        *,
        after: datetime | None = None,
        before: datetime | None = None,
        created_after: datetime | None = None,
        updated_after: datetime | None = None,
        limit: int | None = None,
        cursor: str | None = None,
    ) -> PaginatedResponse[EmailV2]:
        """One page of emails across the organization (V2), the ones the key's user may see.

        ``after`` (inclusive) / ``before`` (exclusive) filter the sent time (``sentAt``);
        ``created_after`` / ``updated_after`` filter ``createdAt`` / ``updatedAt`` (inclusive),
        all rounded outward to whole seconds. There is no sort option, and ``updatedAt`` is
        empty for items never changed: for delta sync, page to the end and track the latest
        times yourself. ``limit`` is 1-100; ``cursor`` (a previous ``next_cursor``) can't be
        combined with other arguments.
        """
        check_cursor_alone(cursor, after, before, created_after, updated_after, limit)
        if cursor is not None:
            return page_of(EmailV2, self._client.get_url(cursor))
        params = interaction_params(
            "sentAt",
            after=after,
            before=before,
            created_after=created_after,
            updated_after=updated_after,
            limit=limit,
        )
        return page_of(EmailV2, self._client.get("/emails", params=params or None))

    def iter_emails(
        self,
        *,
        after: datetime | None = None,
        before: datetime | None = None,
        created_after: datetime | None = None,
        updated_after: datetime | None = None,
        limit: int | None = None,
    ) -> Iterator[EmailV2]:
        """All emails matching the filters (every page); see :meth:`list_emails`."""

        def fetch_page(next_url: str | None) -> PaginatedResponse[EmailV2]:
            if next_url:
                return self.list_emails(cursor=next_url)
            return self.list_emails(
                after=after,
                before=before,
                created_after=created_after,
                updated_after=updated_after,
                limit=limit,
            )

        return PageIterator(fetch_page)

    def list_meetings(
        self,
        *,
        after: datetime | None = None,
        before: datetime | None = None,
        created_after: datetime | None = None,
        updated_after: datetime | None = None,
        limit: int | None = None,
        cursor: str | None = None,
    ) -> PaginatedResponse[MeetingV2]:
        """One page of meetings across the organization (V2), the ones the key's user may see.

        ``after`` (inclusive) / ``before`` (exclusive) filter the start time (``startTime``);
        ``created_after`` / ``updated_after`` filter ``createdAt`` / ``updatedAt`` (inclusive),
        all rounded outward to whole seconds. There is no sort option, and ``updatedAt`` is
        empty for items never changed: for delta sync, page to the end and track the latest
        times yourself. ``limit`` is 1-100; ``cursor`` (a previous ``next_cursor``) can't be
        combined with other arguments.
        """
        check_cursor_alone(cursor, after, before, created_after, updated_after, limit)
        if cursor is not None:
            return page_of(MeetingV2, self._client.get_url(cursor))
        params = interaction_params(
            "startTime",
            after=after,
            before=before,
            created_after=created_after,
            updated_after=updated_after,
            limit=limit,
        )
        return page_of(MeetingV2, self._client.get("/meetings", params=params or None))

    def get_meeting(self, meeting_id: int) -> MeetingV2:
        """One meeting (V2, Affinity API 2026-07-15+), if the key's user may see it."""
        return MeetingV2.model_validate(
            self._client.get(
                f"/meetings/{int(meeting_id)}", min_api_version=MEETING_MIN_API_VERSION
            )
        )

    def iter_meetings(
        self,
        *,
        after: datetime | None = None,
        before: datetime | None = None,
        created_after: datetime | None = None,
        updated_after: datetime | None = None,
        limit: int | None = None,
    ) -> Iterator[MeetingV2]:
        """All meetings matching the filters (every page); see :meth:`list_meetings`."""

        def fetch_page(next_url: str | None) -> PaginatedResponse[MeetingV2]:
            if next_url:
                return self.list_meetings(cursor=next_url)
            return self.list_meetings(
                after=after,
                before=before,
                created_after=created_after,
                updated_after=updated_after,
                limit=limit,
            )

        return PageIterator(fetch_page)

    def list_calls(
        self,
        *,
        after: datetime | None = None,
        before: datetime | None = None,
        created_after: datetime | None = None,
        updated_after: datetime | None = None,
        limit: int | None = None,
        cursor: str | None = None,
    ) -> PaginatedResponse[CallV2]:
        """One page of calls across the organization (V2), the ones the key's user may see.

        ``after`` (inclusive) / ``before`` (exclusive) filter the start time (``startTime``);
        ``created_after`` / ``updated_after`` filter ``createdAt`` / ``updatedAt`` (inclusive),
        all rounded outward to whole seconds. There is no sort option, and ``updatedAt`` is
        empty for items never changed: for delta sync, page to the end and track the latest
        times yourself. ``limit`` is 1-100; ``cursor`` (a previous ``next_cursor``) can't be
        combined with other arguments.
        """
        check_cursor_alone(cursor, after, before, created_after, updated_after, limit)
        if cursor is not None:
            return page_of(CallV2, self._client.get_url(cursor))
        params = interaction_params(
            "startTime",
            after=after,
            before=before,
            created_after=created_after,
            updated_after=updated_after,
            limit=limit,
        )
        return page_of(CallV2, self._client.get("/calls", params=params or None))

    def iter_calls(
        self,
        *,
        after: datetime | None = None,
        before: datetime | None = None,
        created_after: datetime | None = None,
        updated_after: datetime | None = None,
        limit: int | None = None,
    ) -> Iterator[CallV2]:
        """All calls matching the filters (every page); see :meth:`list_calls`."""

        def fetch_page(next_url: str | None) -> PaginatedResponse[CallV2]:
            if next_url:
                return self.list_calls(cursor=next_url)
            return self.list_calls(
                after=after,
                before=before,
                created_after=created_after,
                updated_after=updated_after,
                limit=limit,
            )

        return PageIterator(fetch_page)

    def list_chat_messages(
        self,
        *,
        after: datetime | None = None,
        before: datetime | None = None,
        created_after: datetime | None = None,
        updated_after: datetime | None = None,
        limit: int | None = None,
        cursor: str | None = None,
    ) -> PaginatedResponse[ChatMessageV2]:
        """One page of chat messages across the organization (V2), the ones the key's user may see.

        ``after`` (inclusive) / ``before`` (exclusive) filter the sent time (``sentAt``);
        ``created_after`` / ``updated_after`` filter ``createdAt`` / ``updatedAt`` (inclusive),
        all rounded outward to whole seconds. There is no sort option, and ``updatedAt`` is
        empty for items never changed: for delta sync, page to the end and track the latest
        times yourself. ``limit`` is 1-100; ``cursor`` (a previous ``next_cursor``) can't be
        combined with other arguments.
        """
        check_cursor_alone(cursor, after, before, created_after, updated_after, limit)
        if cursor is not None:
            return page_of(ChatMessageV2, self._client.get_url(cursor))
        params = interaction_params(
            "sentAt",
            after=after,
            before=before,
            created_after=created_after,
            updated_after=updated_after,
            limit=limit,
        )
        return page_of(ChatMessageV2, self._client.get("/chat-messages", params=params or None))

    def iter_chat_messages(
        self,
        *,
        after: datetime | None = None,
        before: datetime | None = None,
        created_after: datetime | None = None,
        updated_after: datetime | None = None,
        limit: int | None = None,
    ) -> Iterator[ChatMessageV2]:
        """All chat messages matching the filters (every page); see :meth:`list_chat_messages`."""

        def fetch_page(next_url: str | None) -> PaginatedResponse[ChatMessageV2]:
            if next_url:
                return self.list_chat_messages(cursor=next_url)
            return self.list_chat_messages(
                after=after,
                before=before,
                created_after=created_after,
                updated_after=updated_after,
                limit=limit,
            )

        return PageIterator(fetch_page)

create(data: InteractionCreate) -> Interaction

Create a new interaction (manually logged).

Source code in affinity/services/v1_only.py

def create(self, data: InteractionCreate) -> Interaction:
    """Create a new interaction (manually logged)."""
    payload = data.model_dump(by_alias=True, mode="python", exclude_none=True)
    _coerce_isoformat(payload, ("date",))

    result = self._client.post("/interactions", json=payload, v1=True)
    return Interaction.model_validate(result)

delete(interaction_id: InteractionId, type: InteractionType) -> bool

Delete an interaction.

Source code in affinity/services/v1_only.py

def delete(self, interaction_id: InteractionId, type: InteractionType) -> bool:
    """Delete an interaction."""
    result = self._client.delete(
        f"/interactions/{int(interaction_id)}",
        params={"type": int(type)},
        v1=True,
    )
    return bool(result.get("success", True))

get(interaction_id: InteractionId, type: InteractionType) -> Interaction

Get a single interaction by ID and type.

Source code in affinity/services/v1_only.py

def get(self, interaction_id: InteractionId, type: InteractionType) -> Interaction:
    """Get a single interaction by ID and type."""
    data = self._client.get(
        f"/interactions/{int(interaction_id)}",
        params={"type": int(type)},
        v1=True,
    )
    return Interaction.model_validate(data)

get_meeting(meeting_id: int) -> MeetingV2

One meeting (V2, Affinity API 2026-07-15+), if the key's user may see it.

Source code in affinity/services/v1_only.py

def get_meeting(self, meeting_id: int) -> MeetingV2:
    """One meeting (V2, Affinity API 2026-07-15+), if the key's user may see it."""
    return MeetingV2.model_validate(
        self._client.get(
            f"/meetings/{int(meeting_id)}", min_api_version=MEETING_MIN_API_VERSION
        )
    )

iter(*, type: InteractionType | None = None, start_time: datetime | None = None, end_time: datetime | None = None, person_id: PersonId | None = None, company_id: CompanyId | None = None, opportunity_id: OpportunityId | None = None, page_size: int | None = None) -> PageIterator[Interaction]

Iterate through all interactions with automatic pagination and date chunking.

Automatically splits date ranges exceeding 365 days into chunks and bridges them with synthetic cursors for seamless iteration.

Parameters:

Name	Type	Description	Default
type	InteractionType | None	Interaction type (required).	None
start_time	datetime | None	Start of date range (required).	None
end_time	datetime | None	End of date range (defaults to now if not provided).	None
person_id	PersonId | None	Filter by person.	None
company_id	CompanyId | None	Filter by company.	None
opportunity_id	OpportunityId | None	Filter by opportunity.	None
page_size	int | None	Page size for API calls.	None

Returns:

Type	Description
PageIterator[Interaction]	PageIterator that yields Interaction objects

Source code in affinity/services/v1_only.py

def iter(
    self,
    *,
    type: InteractionType | None = None,
    start_time: datetime | None = None,
    end_time: datetime | None = None,
    person_id: PersonId | None = None,
    company_id: CompanyId | None = None,
    opportunity_id: OpportunityId | None = None,
    page_size: int | None = None,
) -> PageIterator[Interaction]:
    """
    Iterate through all interactions with automatic pagination and date chunking.

    Automatically splits date ranges exceeding 365 days into chunks
    and bridges them with synthetic cursors for seamless iteration.

    Args:
        type: Interaction type (required).
        start_time: Start of date range (required).
        end_time: End of date range (defaults to now if not provided).
        person_id: Filter by person.
        company_id: Filter by company.
        opportunity_id: Filter by opportunity.
        page_size: Page size for API calls.

    Returns:
        PageIterator that yields Interaction objects
    """
    if type is None:
        raise ValueError(
            "type is required for interactions API. "
            "Use InteractionType.EMAIL, MEETING, CALL, or CHAT_MESSAGE."
        )
    if start_time is None:
        raise ValueError("start_time is required for interactions API.")
    if not any(x is not None for x in (person_id, company_id, opportunity_id)):
        raise ValueError(
            "At least one entity filter is required: person_id, company_id, or opportunity_id."
        )
    resolved_end = end_time if end_time is not None else datetime.now(timezone.utc)
    if (start_time.tzinfo is None) != (resolved_end.tzinfo is None):
        raise ValueError(
            "start_time and end_time must both be timezone-aware or both naive. "
            "Recommended: use timezone-aware datetimes (e.g., datetime.now(timezone.utc))."
        )
    if resolved_end <= start_time:
        raise ValueError("start_time must be before end_time.")
    chunks = _chunk_date_range(start_time, resolved_end)
    chunk_index = 0
    chunk_sentinel = f"__chunk_{uuid.uuid4().hex}__"

    def fetch_page(cursor: str | None) -> PaginatedResponse[Interaction]:
        nonlocal chunk_index
        if cursor == chunk_sentinel:
            chunk_index += 1
            cursor = None
        if chunk_index >= len(chunks):
            return PaginatedResponse[Interaction](data=[])
        c_start, c_end = chunks[chunk_index]
        response = self.list(
            type=type,
            start_time=c_start,
            end_time=c_end,
            person_id=person_id,
            company_id=company_id,
            opportunity_id=opportunity_id,
            page_size=page_size,
            page_token=cursor,
        )
        if response.next_cursor is None and chunk_index < len(chunks) - 1:
            response.next_page_token = chunk_sentinel
        return response

    return PageIterator(fetch_page)

iter_calls(*, after: datetime | None = None, before: datetime | None = None, created_after: datetime | None = None, updated_after: datetime | None = None, limit: int | None = None) -> Iterator[CallV2]

All calls matching the filters (every page); see :meth:list_calls.

Source code in affinity/services/v1_only.py

def iter_calls(
    self,
    *,
    after: datetime | None = None,
    before: datetime | None = None,
    created_after: datetime | None = None,
    updated_after: datetime | None = None,
    limit: int | None = None,
) -> Iterator[CallV2]:
    """All calls matching the filters (every page); see :meth:`list_calls`."""

    def fetch_page(next_url: str | None) -> PaginatedResponse[CallV2]:
        if next_url:
            return self.list_calls(cursor=next_url)
        return self.list_calls(
            after=after,
            before=before,
            created_after=created_after,
            updated_after=updated_after,
            limit=limit,
        )

    return PageIterator(fetch_page)

iter_chat_messages(*, after: datetime | None = None, before: datetime | None = None, created_after: datetime | None = None, updated_after: datetime | None = None, limit: int | None = None) -> Iterator[ChatMessageV2]

All chat messages matching the filters (every page); see :meth:list_chat_messages.

Source code in affinity/services/v1_only.py

def iter_chat_messages(
    self,
    *,
    after: datetime | None = None,
    before: datetime | None = None,
    created_after: datetime | None = None,
    updated_after: datetime | None = None,
    limit: int | None = None,
) -> Iterator[ChatMessageV2]:
    """All chat messages matching the filters (every page); see :meth:`list_chat_messages`."""

    def fetch_page(next_url: str | None) -> PaginatedResponse[ChatMessageV2]:
        if next_url:
            return self.list_chat_messages(cursor=next_url)
        return self.list_chat_messages(
            after=after,
            before=before,
            created_after=created_after,
            updated_after=updated_after,
            limit=limit,
        )

    return PageIterator(fetch_page)

iter_emails(*, after: datetime | None = None, before: datetime | None = None, created_after: datetime | None = None, updated_after: datetime | None = None, limit: int | None = None) -> Iterator[EmailV2]

All emails matching the filters (every page); see :meth:list_emails.

Source code in affinity/services/v1_only.py

def iter_emails(
    self,
    *,
    after: datetime | None = None,
    before: datetime | None = None,
    created_after: datetime | None = None,
    updated_after: datetime | None = None,
    limit: int | None = None,
) -> Iterator[EmailV2]:
    """All emails matching the filters (every page); see :meth:`list_emails`."""

    def fetch_page(next_url: str | None) -> PaginatedResponse[EmailV2]:
        if next_url:
            return self.list_emails(cursor=next_url)
        return self.list_emails(
            after=after,
            before=before,
            created_after=created_after,
            updated_after=updated_after,
            limit=limit,
        )

    return PageIterator(fetch_page)

iter_meetings(*, after: datetime | None = None, before: datetime | None = None, created_after: datetime | None = None, updated_after: datetime | None = None, limit: int | None = None) -> Iterator[MeetingV2]

All meetings matching the filters (every page); see :meth:list_meetings.

Source code in affinity/services/v1_only.py

def iter_meetings(
    self,
    *,
    after: datetime | None = None,
    before: datetime | None = None,
    created_after: datetime | None = None,
    updated_after: datetime | None = None,
    limit: int | None = None,
) -> Iterator[MeetingV2]:
    """All meetings matching the filters (every page); see :meth:`list_meetings`."""

    def fetch_page(next_url: str | None) -> PaginatedResponse[MeetingV2]:
        if next_url:
            return self.list_meetings(cursor=next_url)
        return self.list_meetings(
            after=after,
            before=before,
            created_after=created_after,
            updated_after=updated_after,
            limit=limit,
        )

    return PageIterator(fetch_page)

list(*, type: InteractionType | None = None, start_time: datetime | None = None, end_time: datetime | None = None, person_id: PersonId | None = None, company_id: CompanyId | None = None, opportunity_id: OpportunityId | None = None, page_size: int | None = None, page_token: str | None = None) -> PaginatedResponse[Interaction]

Get interactions with filtering.

All parameters are validated before calling the API: - type, start_time, end_time are required - At least one entity ID (person_id, company_id, or opportunity_id) - Date range must be <= 365 days - start_time must be before end_time

For ranges exceeding 365 days, use iter() which automatically chunks.

Returns V1 paginated response with data and next_page_token.

Raises:

Type	Description
ValueError	If required parameters are missing or invalid.

Source code in affinity/services/v1_only.py

def list(
    self,
    *,
    type: InteractionType | None = None,
    start_time: datetime | None = None,
    end_time: datetime | None = None,
    person_id: PersonId | None = None,
    company_id: CompanyId | None = None,
    opportunity_id: OpportunityId | None = None,
    page_size: int | None = None,
    page_token: str | None = None,
) -> PaginatedResponse[Interaction]:
    """
    Get interactions with filtering.

    All parameters are validated before calling the API:
    - type, start_time, end_time are required
    - At least one entity ID (person_id, company_id, or opportunity_id)
    - Date range must be <= 365 days
    - start_time must be before end_time

    For ranges exceeding 365 days, use iter() which automatically chunks.

    Returns V1 paginated response with `data` and `next_page_token`.

    Raises:
        ValueError: If required parameters are missing or invalid.
    """
    if type is None:
        raise ValueError(
            "type is required for interactions API. "
            "Use InteractionType.EMAIL, MEETING, CALL, or CHAT_MESSAGE."
        )
    if start_time is None:
        raise ValueError(
            "start_time is required for interactions API. "
            "Use iter() for automatic date range handling."
        )
    if end_time is None:
        raise ValueError(
            "end_time is required for interactions API. "
            "Use iter() for automatic date range handling."
        )
    if (start_time.tzinfo is None) != (end_time.tzinfo is None):
        raise ValueError(
            "start_time and end_time must both be timezone-aware or both naive. "
            "Recommended: use timezone-aware datetimes (e.g., datetime.now(timezone.utc))."
        )
    if end_time <= start_time:
        raise ValueError("start_time must be before end_time.")
    if (end_time - start_time) > timedelta(days=_MAX_INTERACTION_RANGE_DAYS):
        raise ValueError(
            f"Date range exceeds {_MAX_INTERACTION_RANGE_DAYS} days. "
            f"Use iter() which automatically chunks large ranges."
        )
    if not any(x is not None for x in (person_id, company_id, opportunity_id)):
        raise ValueError(
            "At least one entity filter is required: person_id, company_id, or opportunity_id."
        )
    params: dict[str, Any] = {"type": int(type)}
    params["start_time"] = start_time.isoformat()
    params["end_time"] = end_time.isoformat()
    if person_id is not None:
        params["person_id"] = int(person_id)
    if company_id is not None:
        params["organization_id"] = int(company_id)
    if opportunity_id is not None:
        params["opportunity_id"] = int(opportunity_id)
    if page_size is not None:
        params["page_size"] = page_size
    if page_token is not None:
        params["page_token"] = page_token

    data = self._client.get("/interactions", params=params or None, v1=True)
    items: Any = None
    if int(type) in (int(InteractionType.MEETING), int(InteractionType.CALL)):
        items = data.get("events")
    elif int(type) == int(InteractionType.CHAT_MESSAGE):
        items = data.get("chat_messages")
    elif int(type) == int(InteractionType.EMAIL):
        items = data.get("emails")

    if items is None:
        items = (
            data.get("interactions")
            or data.get("events")
            or data.get("emails")
            or data.get("chat_messages")
            or data.get("data", [])
        )
    if not isinstance(items, list):
        items = []
    return PaginatedResponse[Interaction](
        data=[Interaction.model_validate(i) for i in items],
        next_page_token=data.get("next_page_token") or data.get("nextPageToken"),
    )

list_calls(*, after: datetime | None = None, before: datetime | None = None, created_after: datetime | None = None, updated_after: datetime | None = None, limit: int | None = None, cursor: str | None = None) -> PaginatedResponse[CallV2]

One page of calls across the organization (V2), the ones the key's user may see.

after (inclusive) / before (exclusive) filter the start time (startTime); created_after / updated_after filter createdAt / updatedAt (inclusive), all rounded outward to whole seconds. There is no sort option, and updatedAt is empty for items never changed: for delta sync, page to the end and track the latest times yourself. limit is 1-100; cursor (a previous next_cursor) can't be combined with other arguments.

Source code in affinity/services/v1_only.py

def list_calls(
    self,
    *,
    after: datetime | None = None,
    before: datetime | None = None,
    created_after: datetime | None = None,
    updated_after: datetime | None = None,
    limit: int | None = None,
    cursor: str | None = None,
) -> PaginatedResponse[CallV2]:
    """One page of calls across the organization (V2), the ones the key's user may see.

    ``after`` (inclusive) / ``before`` (exclusive) filter the start time (``startTime``);
    ``created_after`` / ``updated_after`` filter ``createdAt`` / ``updatedAt`` (inclusive),
    all rounded outward to whole seconds. There is no sort option, and ``updatedAt`` is
    empty for items never changed: for delta sync, page to the end and track the latest
    times yourself. ``limit`` is 1-100; ``cursor`` (a previous ``next_cursor``) can't be
    combined with other arguments.
    """
    check_cursor_alone(cursor, after, before, created_after, updated_after, limit)
    if cursor is not None:
        return page_of(CallV2, self._client.get_url(cursor))
    params = interaction_params(
        "startTime",
        after=after,
        before=before,
        created_after=created_after,
        updated_after=updated_after,
        limit=limit,
    )
    return page_of(CallV2, self._client.get("/calls", params=params or None))

list_chat_messages(*, after: datetime | None = None, before: datetime | None = None, created_after: datetime | None = None, updated_after: datetime | None = None, limit: int | None = None, cursor: str | None = None) -> PaginatedResponse[ChatMessageV2]

One page of chat messages across the organization (V2), the ones the key's user may see.

after (inclusive) / before (exclusive) filter the sent time (sentAt); created_after / updated_after filter createdAt / updatedAt (inclusive), all rounded outward to whole seconds. There is no sort option, and updatedAt is empty for items never changed: for delta sync, page to the end and track the latest times yourself. limit is 1-100; cursor (a previous next_cursor) can't be combined with other arguments.

Source code in affinity/services/v1_only.py

def list_chat_messages(
    self,
    *,
    after: datetime | None = None,
    before: datetime | None = None,
    created_after: datetime | None = None,
    updated_after: datetime | None = None,
    limit: int | None = None,
    cursor: str | None = None,
) -> PaginatedResponse[ChatMessageV2]:
    """One page of chat messages across the organization (V2), the ones the key's user may see.

    ``after`` (inclusive) / ``before`` (exclusive) filter the sent time (``sentAt``);
    ``created_after`` / ``updated_after`` filter ``createdAt`` / ``updatedAt`` (inclusive),
    all rounded outward to whole seconds. There is no sort option, and ``updatedAt`` is
    empty for items never changed: for delta sync, page to the end and track the latest
    times yourself. ``limit`` is 1-100; ``cursor`` (a previous ``next_cursor``) can't be
    combined with other arguments.
    """
    check_cursor_alone(cursor, after, before, created_after, updated_after, limit)
    if cursor is not None:
        return page_of(ChatMessageV2, self._client.get_url(cursor))
    params = interaction_params(
        "sentAt",
        after=after,
        before=before,
        created_after=created_after,
        updated_after=updated_after,
        limit=limit,
    )
    return page_of(ChatMessageV2, self._client.get("/chat-messages", params=params or None))

list_emails(*, after: datetime | None = None, before: datetime | None = None, created_after: datetime | None = None, updated_after: datetime | None = None, limit: int | None = None, cursor: str | None = None) -> PaginatedResponse[EmailV2]

One page of emails across the organization (V2), the ones the key's user may see.

after (inclusive) / before (exclusive) filter the sent time (sentAt); created_after / updated_after filter createdAt / updatedAt (inclusive), all rounded outward to whole seconds. There is no sort option, and updatedAt is empty for items never changed: for delta sync, page to the end and track the latest times yourself. limit is 1-100; cursor (a previous next_cursor) can't be combined with other arguments.

Source code in affinity/services/v1_only.py

def list_emails(
    self,
    *,
    after: datetime | None = None,
    before: datetime | None = None,
    created_after: datetime | None = None,
    updated_after: datetime | None = None,
    limit: int | None = None,
    cursor: str | None = None,
) -> PaginatedResponse[EmailV2]:
    """One page of emails across the organization (V2), the ones the key's user may see.

    ``after`` (inclusive) / ``before`` (exclusive) filter the sent time (``sentAt``);
    ``created_after`` / ``updated_after`` filter ``createdAt`` / ``updatedAt`` (inclusive),
    all rounded outward to whole seconds. There is no sort option, and ``updatedAt`` is
    empty for items never changed: for delta sync, page to the end and track the latest
    times yourself. ``limit`` is 1-100; ``cursor`` (a previous ``next_cursor``) can't be
    combined with other arguments.
    """
    check_cursor_alone(cursor, after, before, created_after, updated_after, limit)
    if cursor is not None:
        return page_of(EmailV2, self._client.get_url(cursor))
    params = interaction_params(
        "sentAt",
        after=after,
        before=before,
        created_after=created_after,
        updated_after=updated_after,
        limit=limit,
    )
    return page_of(EmailV2, self._client.get("/emails", params=params or None))

list_meetings(*, after: datetime | None = None, before: datetime | None = None, created_after: datetime | None = None, updated_after: datetime | None = None, limit: int | None = None, cursor: str | None = None) -> PaginatedResponse[MeetingV2]

One page of meetings across the organization (V2), the ones the key's user may see.

after (inclusive) / before (exclusive) filter the start time (startTime); created_after / updated_after filter createdAt / updatedAt (inclusive), all rounded outward to whole seconds. There is no sort option, and updatedAt is empty for items never changed: for delta sync, page to the end and track the latest times yourself. limit is 1-100; cursor (a previous next_cursor) can't be combined with other arguments.

Source code in affinity/services/v1_only.py

def list_meetings(
    self,
    *,
    after: datetime | None = None,
    before: datetime | None = None,
    created_after: datetime | None = None,
    updated_after: datetime | None = None,
    limit: int | None = None,
    cursor: str | None = None,
) -> PaginatedResponse[MeetingV2]:
    """One page of meetings across the organization (V2), the ones the key's user may see.

    ``after`` (inclusive) / ``before`` (exclusive) filter the start time (``startTime``);
    ``created_after`` / ``updated_after`` filter ``createdAt`` / ``updatedAt`` (inclusive),
    all rounded outward to whole seconds. There is no sort option, and ``updatedAt`` is
    empty for items never changed: for delta sync, page to the end and track the latest
    times yourself. ``limit`` is 1-100; ``cursor`` (a previous ``next_cursor``) can't be
    combined with other arguments.
    """
    check_cursor_alone(cursor, after, before, created_after, updated_after, limit)
    if cursor is not None:
        return page_of(MeetingV2, self._client.get_url(cursor))
    params = interaction_params(
        "startTime",
        after=after,
        before=before,
        created_after=created_after,
        updated_after=updated_after,
        limit=limit,
    )
    return page_of(MeetingV2, self._client.get("/meetings", params=params or None))

update(interaction_id: InteractionId, type: InteractionType, data: InteractionUpdate) -> Interaction

Update an interaction.

Source code in affinity/services/v1_only.py

def update(
    self,
    interaction_id: InteractionId,
    type: InteractionType,
    data: InteractionUpdate,
) -> Interaction:
    """Update an interaction."""
    payload = data.model_dump(
        by_alias=True,
        mode="python",
        exclude_unset=True,
        exclude_none=True,
    )
    payload["type"] = int(type)
    _coerce_isoformat(payload, ("date",))

    result = self._client.put(
        f"/interactions/{int(interaction_id)}",
        json=payload,
        v1=True,
    )
    return Interaction.model_validate(result)