Skip to content

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 
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
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,
    ) -> V1PaginatedResponse[Interaction]:
        """
        Get interactions with optional filtering.

        The Affinity API requires:
        - type: Interaction type (meeting, call, email, chat)
        - start_time and end_time: Date range (max 1 year)
        - One entity ID: person_id, company_id, or opportunity_id

        Returns V1 paginated response with `data` and `next_page_token`.
        """
        params: dict[str, Any] = {}
        if type is not None:
            params["type"] = int(type)
        if start_time:
            params["start_time"] = start_time.isoformat()
        if end_time:
            params["end_time"] = end_time.isoformat()
        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 page_size:
            params["page_size"] = page_size
        if page_token:
            params["page_token"] = page_token

        data = self._client.get("/interactions", params=params or None, v1=True)
        items: Any = None
        if type is not 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 V1PaginatedResponse[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))

create(data: InteractionCreate) -> Interaction

Create a new interaction (manually logged).

Source code in affinity/services/v1_only.py 
459
460
461
462
463
464
465
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 
490
491
492
493
494
495
496
497
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 
450
451
452
453
454
455
456
457
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)

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) -> V1PaginatedResponse[Interaction]

Get interactions with optional filtering.

The Affinity API requires: - type: Interaction type (meeting, call, email, chat) - start_time and end_time: Date range (max 1 year) - One entity ID: person_id, company_id, or opportunity_id

Returns V1 paginated response with data and next_page_token.

Source code in affinity/services/v1_only.py 
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
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,
) -> V1PaginatedResponse[Interaction]:
    """
    Get interactions with optional filtering.

    The Affinity API requires:
    - type: Interaction type (meeting, call, email, chat)
    - start_time and end_time: Date range (max 1 year)
    - One entity ID: person_id, company_id, or opportunity_id

    Returns V1 paginated response with `data` and `next_page_token`.
    """
    params: dict[str, Any] = {}
    if type is not None:
        params["type"] = int(type)
    if start_time:
        params["start_time"] = start_time.isoformat()
    if end_time:
        params["end_time"] = end_time.isoformat()
    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 page_size:
        params["page_size"] = page_size
    if page_token:
        params["page_token"] = page_token

    data = self._client.get("/interactions", params=params or None, v1=True)
    items: Any = None
    if type is not 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 V1PaginatedResponse[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 
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
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)