Interactions (V1)

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", False))

    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)

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", False))

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)

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)

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"),
    )

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)