Models

Affinity data models.

All Pydantic models are available from this module.

Tip

ID types and enums live in affinity.types.

AffinityList

Bases: AffinityModel

A list (spreadsheet) in Affinity.

Named AffinityList to avoid collision with Python's list type.

Note

client.lists.get() reads the list from V2, so fields is None (use client.lists.get_fields(list_id)) and additional_permissions is empty. Use client.lists.get_size(list_id) for the number of entries (read from V1).

Source code in affinity/models/entities.py

class AffinityList(AffinityModel):
    """
    A list (spreadsheet) in Affinity.

    Named AffinityList to avoid collision with Python's list type.

    Note:
        ``client.lists.get()`` reads the list from V2, so ``fields`` is ``None`` (use
        ``client.lists.get_fields(list_id)``) and ``additional_permissions`` is empty. Use
        ``client.lists.get_size(list_id)`` for the number of entries (read from V1).
    """

    id: ListId
    name: str
    type: ListType
    # V1 sends `public`, V2 `isPublic`; serialized as `public`.
    is_public: bool = Field(alias="public", validation_alias=AliasChoices("public", "isPublic"))
    owner_id: UserId = Field(alias="ownerId")
    creator_id: UserId | None = Field(None, alias="creatorId")
    created_at: ISODatetime | None = Field(None, alias="createdAt")

    # Fields on this list (returned for single list fetch)
    fields: list[FieldMetadata] | None = None

    # Permissions
    additional_permissions: list[ListPermission] = Field(
        default_factory=list, alias="additionalPermissions"
    )

    # Internal - not guaranteed accurate from V2. Excluded from serialization.
    # Populated from listSize (V2) or list_size (V1) via model_validator + model_post_init.
    _list_size_hint: int = PrivateAttr(default=0)

    # Temporary field to pass list_size from validator to model_post_init (excluded from output)
    list_size_temp: int | None = Field(None, exclude=True, repr=False)

    @model_validator(mode="before")
    @classmethod
    def _extract_list_size(cls, data: dict[str, Any]) -> dict[str, Any]:
        if isinstance(data, Mapping):
            data = dict(data)
            # Extract listSize and pass via temp field (V2 uses listSize, V1 uses list_size)
            if "listSize" in data:
                data["list_size_temp"] = data.pop("listSize")
            elif "list_size" in data:
                data["list_size_temp"] = data.pop("list_size")
        return data

    def model_post_init(self, __context: Any) -> None:
        """Transfer list_size_temp to _list_size_hint private attr."""
        if self.list_size_temp is not None:
            object.__setattr__(self, "_list_size_hint", self.list_size_temp)
            # Clear the temp field
            object.__setattr__(self, "list_size_temp", None)

model_post_init(__context: Any) -> None

Transfer list_size_temp to _list_size_hint private attr.

Source code in affinity/models/entities.py

def model_post_init(self, __context: Any) -> None:
    """Transfer list_size_temp to _list_size_hint private attr."""
    if self.list_size_temp is not None:
        object.__setattr__(self, "_list_size_hint", self.list_size_temp)
        # Clear the temp field
        object.__setattr__(self, "list_size_temp", None)

AffinityModel

Bases: BaseModel

Base model with common configuration.

Source code in affinity/models/entities.py

class AffinityModel(BaseModel):
    """Base model with common configuration."""

    model_config = ConfigDict(
        extra="ignore",  # Ignore unknown fields from API
        populate_by_name=True,  # Allow both alias and field name
        use_enum_values=True,  # Serialize enums as values
        validate_assignment=True,  # Validate on attribute assignment
    )

AsyncPageIterator

Bases: Generic[T]

Asynchronous iterator that automatically fetches all pages.

Usage

async for item in client.companies.all(): print(item.name)

Source code in affinity/models/pagination.py

class AsyncPageIterator(Generic[T]):
    """
    Asynchronous iterator that automatically fetches all pages.

    Usage:
        async for item in client.companies.all():
            print(item.name)
    """

    def __init__(
        self,
        fetch_page: Callable[[str | None], Awaitable[PaginatedResponse[T]]],
        initial_cursor: str | None = None,
    ):
        self._fetch_page = fetch_page
        self._next_cursor = initial_cursor
        self._current_page: list[T] = []
        self._index = 0
        self._exhausted = False
        self._guard = _CursorGuard(initial_cursor)

    def __aiter__(self) -> AsyncIterator[T]:
        return self

    async def __anext__(self) -> T:
        while True:
            # If we have items in current page, return next
            if self._index < len(self._current_page):
                item = self._current_page[self._index]
                self._index += 1
                return item

            # Need to fetch next page
            if self._exhausted:
                raise StopAsyncIteration

            requested_url = self._next_cursor
            response = await self._fetch_page(requested_url)
            self._current_page = list(response.data)
            self._next_cursor = response.next_cursor
            self._index = 0

            # Stop at the last page, or when paging can't progress (see _CursorGuard).
            if self._guard.last_page(requested_url, response):
                self._exhausted = True

            # Empty pages can still legitimately include nextUrl; keep paging
            # until we get data or the cursor is exhausted.
            if not self._current_page:
                if not self._exhausted:
                    continue
                raise StopAsyncIteration

    async def pages(
        self,
        *,
        on_progress: Callable[[PaginationProgress], None] | None = None,
    ) -> AsyncIterator[PaginatedResponse[T]]:
        """
        Iterate through pages (not individual items).

        Args:
            on_progress: Optional callback fired after fetching each page.
                Receives PaginationProgress with page_number, items_in_page,
                items_so_far, and has_next. Callbacks should be lightweight;
                heavy processing should happen outside the callback to avoid
                blocking iteration.

        Yields:
            PaginatedResponse objects for each page.

        Example:
            def report(p: PaginationProgress):
                print(f"Page {p.page_number}: {p.items_so_far} items so far")

            async for page in client.persons.all().pages(on_progress=report):
                process(page.data)
        """
        page_number = 0
        items_so_far = 0
        guard = _CursorGuard(self._next_cursor)

        while True:
            requested_url = self._next_cursor
            response = await self._fetch_page(requested_url)
            self._next_cursor = response.next_cursor
            page_number += 1
            items_in_page = len(response.data)
            items_so_far += items_in_page

            # Stop at the last page, or when paging can't progress (see _CursorGuard).
            last = guard.last_page(requested_url, response)

            if response.data:
                if on_progress:
                    on_progress(
                        PaginationProgress(
                            page_number=page_number,
                            items_in_page=items_in_page,
                            items_so_far=items_so_far,
                            has_next=not last,
                        )
                    )
                yield response

            if last:
                break

    async def all(self, *, limit: int | None = _DEFAULT_LIMIT) -> list[T]:
        """
        Fetch all items across all pages into a list.

        Args:
            limit: Maximum items to fetch. Default 100,000. Set to None for unlimited.

        Returns:
            List of all items.

        Raises:
            TooManyResultsError: If results exceed limit.

        Note:
            The check occurs after extending results, so the final list may exceed
            limit by up to one page before the error is raised.

        Example:
            # Default - safe for most use cases
            persons = [p async for p in client.persons.all()]  # Using async iterator

            # Or use .all() method with limit check
            it = AsyncPageIterator(fetch_page)
            persons = await it.all()  # Returns list, raises if > 100k

            # Explicit unlimited for large exports
            all_persons = await it.all(limit=None)

            # Custom limit
            persons = await it.all(limit=500_000)
        """
        results: list[T] = []

        async for page in self.pages():
            results.extend(page.data)

            if limit is not None and len(results) > limit:
                raise TooManyResultsError(
                    f"Exceeded limit={limit:,} items. "
                    f"Use pages() for streaming, add a filter, or pass limit=None."
                )

        return results

all(*, limit: int | None = _DEFAULT_LIMIT) -> list[T] async

Fetch all items across all pages into a list.

Parameters:

Name	Type	Description	Default
limit	int | None	Maximum items to fetch. Default 100,000. Set to None for unlimited.	_DEFAULT_LIMIT

Returns:

Type	Description
list[T]	List of all items.

Raises:

Type	Description
TooManyResultsError	If results exceed limit.

Note

The check occurs after extending results, so the final list may exceed limit by up to one page before the error is raised.

Example

Default - safe for most use cases

persons = [p async for p in client.persons.all()] # Using async iterator

Or use .all() method with limit check

it = AsyncPageIterator(fetch_page) persons = await it.all() # Returns list, raises if > 100k

Explicit unlimited for large exports

all_persons = await it.all(limit=None)

Custom limit

persons = await it.all(limit=500_000)

Source code in affinity/models/pagination.py

async def all(self, *, limit: int | None = _DEFAULT_LIMIT) -> list[T]:
    """
    Fetch all items across all pages into a list.

    Args:
        limit: Maximum items to fetch. Default 100,000. Set to None for unlimited.

    Returns:
        List of all items.

    Raises:
        TooManyResultsError: If results exceed limit.

    Note:
        The check occurs after extending results, so the final list may exceed
        limit by up to one page before the error is raised.

    Example:
        # Default - safe for most use cases
        persons = [p async for p in client.persons.all()]  # Using async iterator

        # Or use .all() method with limit check
        it = AsyncPageIterator(fetch_page)
        persons = await it.all()  # Returns list, raises if > 100k

        # Explicit unlimited for large exports
        all_persons = await it.all(limit=None)

        # Custom limit
        persons = await it.all(limit=500_000)
    """
    results: list[T] = []

    async for page in self.pages():
        results.extend(page.data)

        if limit is not None and len(results) > limit:
            raise TooManyResultsError(
                f"Exceeded limit={limit:,} items. "
                f"Use pages() for streaming, add a filter, or pass limit=None."
            )

    return results

pages(*, on_progress: Callable[[PaginationProgress], None] | None = None) -> AsyncIterator[PaginatedResponse[T]] async

Iterate through pages (not individual items).

Parameters:

Name	Type	Description	Default
on_progress	Callable[[PaginationProgress], None] | None	Optional callback fired after fetching each page. Receives PaginationProgress with page_number, items_in_page, items_so_far, and has_next. Callbacks should be lightweight; heavy processing should happen outside the callback to avoid blocking iteration.	None

Yields:

Type	Description
AsyncIterator[PaginatedResponse[T]]	PaginatedResponse objects for each page.

Example

def report(p: PaginationProgress): print(f"Page {p.page_number}: {p.items_so_far} items so far")

async for page in client.persons.all().pages(on_progress=report): process(page.data)

Source code in affinity/models/pagination.py

async def pages(
    self,
    *,
    on_progress: Callable[[PaginationProgress], None] | None = None,
) -> AsyncIterator[PaginatedResponse[T]]:
    """
    Iterate through pages (not individual items).

    Args:
        on_progress: Optional callback fired after fetching each page.
            Receives PaginationProgress with page_number, items_in_page,
            items_so_far, and has_next. Callbacks should be lightweight;
            heavy processing should happen outside the callback to avoid
            blocking iteration.

    Yields:
        PaginatedResponse objects for each page.

    Example:
        def report(p: PaginationProgress):
            print(f"Page {p.page_number}: {p.items_so_far} items so far")

        async for page in client.persons.all().pages(on_progress=report):
            process(page.data)
    """
    page_number = 0
    items_so_far = 0
    guard = _CursorGuard(self._next_cursor)

    while True:
        requested_url = self._next_cursor
        response = await self._fetch_page(requested_url)
        self._next_cursor = response.next_cursor
        page_number += 1
        items_in_page = len(response.data)
        items_so_far += items_in_page

        # Stop at the last page, or when paging can't progress (see _CursorGuard).
        last = guard.last_page(requested_url, response)

        if response.data:
            if on_progress:
                on_progress(
                    PaginationProgress(
                        page_number=page_number,
                        items_in_page=items_in_page,
                        items_so_far=items_so_far,
                        has_next=not last,
                    )
                )
            yield response

        if last:
            break

BatchOperationResponse

Bases: AffinityModel

Response from batch field operations.

Affinity's update-fields PATCH returns {"operation": "update-fields"} on success and an error for the whole batch otherwise (all-or-nothing), so results is usually empty.

Source code in affinity/models/pagination.py

class BatchOperationResponse(AffinityModel):
    """Response from batch field operations.

    Affinity's ``update-fields`` PATCH returns ``{"operation": "update-fields"}`` on success and
    an error for the whole batch otherwise (all-or-nothing), so ``results`` is usually empty.
    """

    operation: str | None = None
    results: list[BatchOperationResult] = Field(default_factory=list)

    @property
    def all_successful(self) -> bool:
        return all(r.success for r in self.results)

    @property
    def failures(self) -> list[BatchOperationResult]:
        return [r for r in self.results if not r.success]

BatchOperationResult

Bases: AffinityModel

Result of a single operation in a batch.

Source code in affinity/models/pagination.py

class BatchOperationResult(AffinityModel):
    """Result of a single operation in a batch."""

    field_id: str = Field(alias="fieldId")
    success: bool
    error: str | None = None

CallV2

Bases: _Event

A call (V2 GET /v2/calls).

Source code in affinity/models/interactions_v2.py

class CallV2(_Event):
    """A call (V2 ``GET /v2/calls``)."""

ChatMessageV2

Bases: AffinityModel

A chat message (V2 GET /v2/chat-messages).

Source code in affinity/models/interactions_v2.py

class ChatMessageV2(AffinityModel):
    """A chat message (V2 ``GET /v2/chat-messages``)."""

    id: int
    sent_at: ISODatetime = Field(alias="sentAt")
    logging_type: str | None = Field(None, alias="loggingType")
    direction: str | None = None
    creator: PersonSummary | None = None
    created_at: ISODatetime = Field(alias="createdAt")
    updated_at: ISODatetime | None = Field(None, alias="updatedAt")
    participants: list[PersonSummary] = Field(default_factory=list)
    participants_total: int | None = None

    @model_validator(mode="before")
    @classmethod
    def _flatten_participants(cls, data: Any) -> Any:
        return _flatten(data, "participantsPreview", "participants", "participants_total")

Company

Bases: AffinityModel

Full company representation.

Note: Called Organization in V1 API.

Source code in affinity/models/entities.py

class Company(AffinityModel):
    """
    Full company representation.

    Note: Called Organization in V1 API.
    """

    id: CompanyId
    name: str
    domain: str | None = None
    domains: list[str] = Field(default_factory=list)
    # V1 sends "global", V2 sends "isGlobal"; output keeps "global" (unchanged JSON contract).
    is_global: bool = Field(
        False, alias="global", validation_alias=AliasChoices("global", "isGlobal")
    )

    # Associations
    person_ids: list[PersonId] = Field(default_factory=list, alias="personIds")
    opportunity_ids: list[OpportunityId] = Field(default_factory=list, alias="opportunityIds")

    # Field values (requested-vs-not-requested preserved)
    fields: FieldValues = Field(default_factory=FieldValues, alias="fields")
    fields_raw: list[dict[str, Any]] | None = Field(default=None, exclude=True)

    @model_validator(mode="before")
    @classmethod
    def _normalize_null_lists_before(cls, value: Any) -> Any:
        value = _normalize_null_lists(
            value,
            (
                "domains",
                "personIds",
                "person_ids",
                "opportunityIds",
                "opportunity_ids",
            ),
        )
        return _preserve_fields_raw(value)

    @model_validator(mode="after")
    def _mark_fields_not_requested_when_omitted(self) -> Company:
        if "fields" not in self.__pydantic_fields_set__:
            self.fields.requested = False
        return self

    # List entries (returned for single company fetch)
    list_entries: list[ListEntry] | None = Field(None, alias="listEntries")

    # Interaction dates
    interaction_dates: InteractionDates | None = Field(None, alias="interactionDates")

    # Detailed interaction data with person IDs (returned when with_interaction_persons=True)
    interactions: Interactions | None = None

CompanyCreate

Bases: AffinityModel

Data for creating a new company (V1 API).

Source code in affinity/models/entities.py

class CompanyCreate(AffinityModel):
    """Data for creating a new company (V1 API)."""

    name: str
    domain: str | None = None
    person_ids: list[PersonId] = Field(default_factory=list)

CompanyDuplicateSuggestion

Bases: AffinityModel

Two company profiles Affinity thinks are the same company.

Merge duplicates[0] into primary with companies.merge(primary.id, duplicate.id). match_criteria: domain-match (high confidence), domain-redirect (medium), name (low).

Source code in affinity/models/discovery.py

class CompanyDuplicateSuggestion(AffinityModel):
    """Two company profiles Affinity thinks are the same company.

    Merge ``duplicates[0]`` into ``primary`` with ``companies.merge(primary.id, duplicate.id)``.
    ``match_criteria``: ``domain-match`` (high confidence), ``domain-redirect`` (medium),
    ``name`` (low).
    """

    id: str
    primary: DuplicateCompanyProfile = Field(alias="primaryProfile")
    # Always one today
    duplicates: list[DuplicateCompanyProfile] = Field(alias="duplicateProfiles")
    match_criteria: str = Field(alias="matchCriteria")

CompanyMergeState

Bases: AffinityModel

One company merge (V2 GET /v2/company-merges).

Source code in affinity/models/secondary.py

class CompanyMergeState(AffinityModel):
    """One company merge (V2 ``GET /v2/company-merges``)."""

    id: int
    status: MergeStatus
    task_id: str = Field(alias="taskId")
    started_at: ISODatetime = Field(alias="startedAt")
    primary_company_id: CompanyId = Field(alias="primaryCompanyId")
    duplicate_company_id: CompanyId = Field(alias="duplicateCompanyId")
    completed_at: ISODatetime | None = Field(None, alias="completedAt")
    error_message: str | None = Field(None, alias="errorMessage")

CompanyRef

Bases: AffinityModel

A bare company reference ({"id": …}).

Source code in affinity/models/secondary.py

class CompanyRef(AffinityModel):
    """A bare company reference (``{"id": …}``)."""

    id: CompanyId

CompanyUpdate

Bases: AffinityModel

Data for updating a company (V1 API).

Source code in affinity/models/entities.py

class CompanyUpdate(AffinityModel):
    """Data for updating a company (V1 API)."""

    name: str | None = None
    domain: str | None = None
    person_ids: list[PersonId] | None = None

ConnectionCompany

Bases: AffinityModel

A company in an inferred connection.

Source code in affinity/models/discovery.py

class ConnectionCompany(AffinityModel):
    """A company in an inferred connection."""

    id: CompanyId
    name: str | None = None
    domain: str | None = None

ConnectionTarget

Bases: AffinityModel

Someone outside your Affinity data you'd like to know.

full_name is always set when known; API 2024-01-01 sends first_name/last_name instead (later versions send only the full name).

Source code in affinity/models/discovery.py

class ConnectionTarget(AffinityModel):
    """Someone outside your Affinity data you'd like to know.

    ``full_name`` is always set when known; API 2024-01-01 sends ``first_name``/``last_name``
    instead (later versions send only the full name).
    """

    full_name: str | None = Field(None, alias="fullName")
    first_name: str | None = Field(None, alias="firstName")
    last_name: str | None = Field(None, alias="lastName")
    title: str | None = None
    linkedin_url: str | None = Field(None, alias="linkedinUrl")
    # Where they work today (not necessarily where the connection comes from)
    current_company: ConnectionCompany = Field(alias="currentCompany")

    @model_validator(mode="after")
    def _full_name(self) -> ConnectionTarget:
        if self.full_name is None and (self.first_name or self.last_name):
            self.full_name = f"{self.first_name or ''} {self.last_name or ''}".strip()
        return self

CoworkerConnection

Bases: AffinityModel

Someone you know (source) who worked with the target at shared_employer.

Source code in affinity/models/discovery.py

class CoworkerConnection(AffinityModel):
    """Someone you know (``source``) who worked with the target at ``shared_employer``."""

    source: RelationshipPerson
    shared_employer: ConnectionCompany
    overlap_start: date
    overlap_end: date

    @classmethod
    def from_api(cls, data: dict[str, object]) -> CoworkerConnection:
        inference = data.get("inference") or {}
        assert isinstance(inference, dict)
        return cls.model_validate(
            {
                "source": data.get("source"),
                "shared_employer": inference.get("sharedEmployer"),
                "overlap_start": inference.get("overlapStartDate"),
                "overlap_end": inference.get("overlapEndDate"),
            }
        )

CoworkerConnectionGroup

Bases: AffinityModel

A target and the people you know who worked with them, strongest first.

Source code in affinity/models/discovery.py

class CoworkerConnectionGroup(AffinityModel):
    """A target and the people you know who worked with them, strongest first."""

    target: ConnectionTarget
    connections: list[CoworkerConnection]

    @classmethod
    def from_api(cls, data: dict[str, object]) -> CoworkerConnectionGroup:
        raw = data.get("connections") or []
        assert isinstance(raw, list)
        return cls(
            target=ConnectionTarget.model_validate(data.get("target")),
            connections=[CoworkerConnection.from_api(c) for c in raw],
        )

DealRole

Bases: AffinityModel

A person's role in a deal.

Source code in affinity/models/discovery.py

class DealRole(AffinityModel):
    """A person's role in a deal."""

    title: str | None = None
    company: ConnectionCompany
    start_date: date | None = Field(None, alias="startDate")
    # None while they're still in the role
    end_date: date | None = Field(None, alias="endDate")

DropdownOption

Bases: AffinityModel

A selectable option in a dropdown field.

type ("dropdown", "ranked-dropdown" or "status-dropdown"), status_category and win_rate come from the V2 dropdown-options endpoints; V1 payloads leave them None and use integer colours.

Source code in affinity/models/entities.py

class DropdownOption(AffinityModel):
    """A selectable option in a dropdown field.

    ``type`` (``"dropdown"``, ``"ranked-dropdown"`` or ``"status-dropdown"``), ``status_category``
    and ``win_rate`` come from the V2 dropdown-options endpoints; V1 payloads leave them ``None``
    and use integer colours.
    """

    id: DropdownOptionId
    text: str
    rank: int | None = None
    color: int | str | None = None
    type: str | None = None
    status_category: str | None = Field(None, alias="statusCategory")
    win_rate: int | None = Field(None, alias="winRate")

DuplicateCompanyProfile

Bases: AffinityModel

A company in a duplicate suggestion.

Source code in affinity/models/discovery.py

class DuplicateCompanyProfile(AffinityModel):
    """A company in a duplicate suggestion."""

    id: CompanyId
    name: str
    domain: str | None = None
    domains: list[str] = Field(default_factory=list)
    is_enriched: bool = Field(False, alias="isEnriched")
    is_global: bool = Field(False, alias="isGlobal")

EmailV2

Bases: AffinityModel

An email (V2 GET /v2/emails). A subject the user may not see is "********".

Source code in affinity/models/interactions_v2.py

class EmailV2(AffinityModel):
    """An email (V2 ``GET /v2/emails``). A subject the user may not see is ``"********"``."""

    id: int
    sent_at: ISODatetime = Field(alias="sentAt")
    logging_type: str | None = Field(None, alias="loggingType")
    direction: str | None = None
    subject: str | None = None
    created_at: ISODatetime = Field(alias="createdAt")
    updated_at: ISODatetime | None = Field(None, alias="updatedAt")
    from_: InteractionAttendee | None = Field(None, alias="from")
    to: list[InteractionAttendee] = Field(default_factory=list)
    to_total: int | None = None
    cc: list[InteractionAttendee] = Field(default_factory=list)
    cc_total: int | None = None

    @model_validator(mode="before")
    @classmethod
    def _flatten_participants(cls, data: Any) -> Any:
        data = _flatten(data, "toParticipantsPreview", "to", "to_total")
        return _flatten(data, "ccParticipantsPreview", "cc", "cc_total")

EntityFile

Bases: AffinityModel

A file attached to an entity.

Source code in affinity/models/secondary.py

class EntityFile(AffinityModel):
    """A file attached to an entity."""

    id: FileId
    name: str
    size: int
    # Observed missing in some V1 responses; treat as optional for robustness.
    content_type: str | None = Field(None, alias="contentType")

    # Associated entity
    person_id: PersonId | None = Field(None, alias="personId")
    # V1 sends `organization_id`; accept the camelCase form too. Serialization keeps
    # `organizationId` so `model_dump(by_alias=True)` output is unchanged.
    company_id: CompanyId | None = Field(
        None,
        validation_alias=AliasChoices("organization_id", "organizationId"),
        serialization_alias="organizationId",
    )
    opportunity_id: OpportunityId | None = Field(None, alias="opportunityId")

    # Uploader
    uploader_id: UserId = Field(alias="uploaderId")

    # Timestamps
    created_at: ISODatetime = Field(alias="createdAt")

FieldCreate

Bases: AffinityModel

Data for creating a new field (V1 API).

Source code in affinity/models/entities.py

class FieldCreate(AffinityModel):
    """Data for creating a new field (V1 API)."""

    model_config = ConfigDict(use_enum_values=False)

    name: str
    entity_type: EntityType
    value_type: FieldValueType
    list_id: ListId | None = None
    allows_multiple: bool = False
    is_list_specific: bool = False
    is_required: bool = False

FieldMetadata

Bases: AffinityModel

Metadata about a field (column) in Affinity.

Includes both V1 numeric IDs and V2 string IDs for enriched fields.

Source code in affinity/models/entities.py

class FieldMetadata(AffinityModel):
    """
    Metadata about a field (column) in Affinity.

    Includes both V1 numeric IDs and V2 string IDs for enriched fields.
    """

    model_config = ConfigDict(use_enum_values=False)

    id: AnyFieldId  # Can be int (field-123) or string (affinity-data-description)
    name: str
    value_type: FieldValueType = Field(alias="valueType")
    allows_multiple: bool = Field(False, alias="allowsMultiple")
    value_type_raw: str | int | None = Field(None, exclude=True)

    # V2 field type classification
    type: str | None = None  # "enriched", "list", "global", "relationship-intelligence"

    # V1 specific fields
    list_id: ListId | None = Field(None, alias="listId")
    track_changes: bool = Field(False, alias="trackChanges")
    enrichment_source: str | None = Field(None, alias="enrichmentSource")
    is_required: bool = Field(False, alias="isRequired")

    # V2: null for built-in fields (no creation time) and fields without a description
    created_at: ISODatetime | None = Field(None, alias="createdAt")
    description: str | None = None

    # V2, only when requested via the fields endpoints' `includes` parameter. Tri-state:
    # key absent = not requested, null = not filterable/sortable, object = how. Kept raw until
    # typed models land; use is_filterable / is_sortable to read the state.
    filterability: dict[str, Any] | None = None
    sortability: dict[str, Any] | None = None

    # Dropdown options for dropdown fields
    dropdown_options: list[DropdownOption] = Field(default_factory=list, alias="dropdownOptions")

    @property
    def is_filterable(self) -> bool | None:
        """True/False when filterability was requested; None when it was not."""
        if "filterability" not in self.model_fields_set:
            return None
        return self.filterability is not None

    @property
    def is_sortable(self) -> bool | None:
        """True/False when sortability was requested; None when it was not."""
        if "sortability" not in self.model_fields_set:
            return None
        return self.sortability is not None

    @model_validator(mode="before")
    @classmethod
    def _preserve_value_type_raw(cls, value: Any) -> Any:
        if not isinstance(value, Mapping):
            return value

        data: dict[str, Any] = dict(value)
        raw = data.get("valueType")
        if raw is None and "value_type" in data:
            raw = data.get("value_type")
        data["value_type_raw"] = raw
        return data

    @model_validator(mode="after")
    def _coerce_allows_multiple_from_value_type(self) -> FieldMetadata:
        # If the server returns a `*-multi` value type, treat it as authoritative for multiplicity.
        try:
            text = str(self.value_type)
        except Exception:
            text = ""
        if text.endswith("-multi") and not self.allows_multiple:
            _logger.debug(
                "FieldMetadata allowsMultiple mismatch: valueType=%s allowsMultiple=%s "
                "(auto-correcting)",
                text,
                self.allows_multiple,
            )
            self.allows_multiple = True
        return self

is_filterable: bool | None property

True/False when filterability was requested; None when it was not.

is_sortable: bool | None property

True/False when sortability was requested; None when it was not.

FieldValue

Bases: AffinityModel

A single field value (cell data).

The value can be various types depending on the field's value_type.

Source code in affinity/models/entities.py

class FieldValue(AffinityModel):
    """
    A single field value (cell data).

    The value can be various types depending on the field's value_type.
    """

    id: FieldValueId
    field_id: AnyFieldId = Field(alias="fieldId")
    entity_id: int = Field(alias="entityId")
    list_entry_id: ListEntryId | None = Field(None, alias="listEntryId")

    # The actual value - type depends on field type
    value: Any

    # Timestamps
    created_at: ISODatetime | None = Field(None, alias="createdAt")
    updated_at: ISODatetime | None = Field(None, alias="updatedAt")

FieldValueChange

Bases: AffinityModel

Historical change to a field value.

Source code in affinity/models/entities.py

class FieldValueChange(AffinityModel):
    """Historical change to a field value."""

    id: FieldValueChangeId
    field_id: FieldId = Field(alias="fieldId")
    entity_id: int = Field(alias="entityId")
    list_entry_id: ListEntryId | None = Field(None, alias="listEntryId")
    action_type: FieldValueChangeAction = Field(alias="actionType")
    value: Any
    changed_at: ISODatetime = Field(alias="changedAt")
    changer: PersonSummary | None = None

FieldValueChangeEntity

Bases: AffinityModel

The entity (person, company or opportunity) whose field value changed.

Source code in affinity/models/secondary.py

class FieldValueChangeEntity(AffinityModel):
    """The entity (person, company or opportunity) whose field value changed."""

    id: int

FieldValueChangeField

Bases: AffinityModel

The field a V2 field value change belongs to.

Source code in affinity/models/secondary.py

class FieldValueChangeField(AffinityModel):
    """The field a V2 field value change belongs to."""

    id: str
    entity_type: str = Field(alias="entityType")
    name: str
    type: str  # "global" or "list"

FieldValueChangeListEntry

Bases: AffinityModel

The list entry a V2 field value change occurred in.

Source code in affinity/models/secondary.py

class FieldValueChangeListEntry(AffinityModel):
    """The list entry a V2 field value change occurred in."""

    id: ListEntryId
    list_id: ListId = Field(alias="listId")

FieldValueChangeV2

Bases: AffinityModel

A field value change from GET /v2/field-value-changes (all entities and fields).

action_type is "add", "update" or "delete" (V2 names; V1 uses create/update/ delete). value_type is the field's value type (e.g. "ranked-dropdown") and value is kept as the API sends it (its shape depends on value_type; a deleted dropdown option or entity comes back as a reference with a displayValue snapshot). changer is None for system changes such as enrichment.

Source code in affinity/models/secondary.py

class FieldValueChangeV2(AffinityModel):
    """A field value change from ``GET /v2/field-value-changes`` (all entities and fields).

    ``action_type`` is ``"add"``, ``"update"`` or ``"delete"`` (V2 names; V1 uses create/update/
    delete). ``value_type`` is the field's value type (e.g. ``"ranked-dropdown"``) and ``value``
    is kept as the API sends it (its shape depends on ``value_type``; a deleted dropdown option
    or entity comes back as a reference with a ``displayValue`` snapshot). ``changer`` is
    ``None`` for system changes such as enrichment.
    """

    id: int
    field: FieldValueChangeField
    entity: FieldValueChangeEntity
    list_entry: FieldValueChangeListEntry | None = Field(None, alias="listEntry")
    changer: User | None = None
    changed_at: ISODatetime = Field(alias="changedAt")
    action_type: str = Field(alias="actionType")
    value_type: str | None = Field(None, alias="type")
    value: Any = None

FieldValueCreate

Bases: AffinityModel

Data for creating a field value (V1 API).

Source code in affinity/models/entities.py

class FieldValueCreate(AffinityModel):
    """Data for creating a field value (V1 API)."""

    field_id: FieldId
    entity_id: int
    value: Any
    list_entry_id: ListEntryId | None = None

FileRef

Bases: AffinityModel

The file matched by a file search (file in a result).

Source code in affinity/models/search.py

class FileRef(AffinityModel):
    """The file matched by a file search (``file`` in a result)."""

    id: FileId
    name: str = ""

FileSearchResult

Bases: AffinityModel

One file search hit: the file, the page of the passage, and the passage.

Source code in affinity/models/search.py

class FileSearchResult(AffinityModel):
    """One file search hit: the file, the page of the passage, and the passage."""

    file: FileRef
    # Null for files without page structure.
    page_number: int | None = Field(None, alias="pageNumber")
    preview: str = ""

FileV2

Bases: AffinityModel

A file's metadata from V2 (files.feed).

V2 files carry no person/company/opportunity; download_url (single-file reads only) is a signed URL that expires 60 seconds after the response.

Source code in affinity/models/secondary.py

class FileV2(AffinityModel):
    """A file's metadata from V2 (``files.feed``).

    V2 files carry no person/company/opportunity; ``download_url`` (single-file reads only) is
    a signed URL that expires 60 seconds after the response.
    """

    id: FileId
    name: str
    size: int
    # MIME type; None when Affinity didn't record it
    content_type: str | None = Field(None, alias="type")
    # None when the uploader isn't recorded
    creator: PersonRef | None = None
    created_at: ISODatetime = Field(alias="createdAt")
    updated_at: ISODatetime | None = Field(None, alias="updatedAt")
    download_url: str | None = Field(None, alias="downloadUrl")

FundingRound

Bases: AffinityModel

The funding round behind an investor-executive connection.

Source code in affinity/models/discovery.py

class FundingRound(AffinityModel):
    """The funding round behind an investor-executive connection."""

    date: date
    series: str | None = None
    amount: float | None = None
    currency: str | None = Field(None, alias="currencyCode")

Grant

Bases: AffinityModel

API key grant information.

Source code in affinity/models/secondary.py

class Grant(AffinityModel):
    """API key grant information."""

    type: str
    scopes: list[str] = Field(default_factory=list)
    created_at: ISODatetime = Field(alias="createdAt")

    @model_validator(mode="before")
    @classmethod
    def _coerce_scope_to_scopes(cls, value: Any) -> Any:
        if not isinstance(value, dict):
            return value
        if "scopes" in value:
            return value
        scope = value.get("scope")
        if isinstance(scope, str):
            updated = dict(value)
            updated["scopes"] = [scope]
            return updated
        return value

    @property
    def scope(self) -> str | None:
        """
        Backwards-compatible convenience for older response shapes.

        Returns the first scope string when present, otherwise None.
        """
        return self.scopes[0] if self.scopes else None

scope: str | None property

Backwards-compatible convenience for older response shapes.

Returns the first scope string when present, otherwise None.

Interaction

Bases: AffinityModel

An interaction (email, meeting, call, or chat message).

Different interaction types have different fields available.

Source code in affinity/models/secondary.py

class Interaction(AffinityModel):
    """
    An interaction (email, meeting, call, or chat message).

    Different interaction types have different fields available.
    """

    id: InteractionId
    type: InteractionType
    date: ISODatetime

    # Associated persons
    persons: list[PersonSummary] = Field(default_factory=list)
    attendees: list[str] = Field(default_factory=list)

    # Meeting/Call specific
    title: str | None = None
    start_time: ISODatetime | None = Field(None, alias="startTime")
    end_time: ISODatetime | None = Field(None, alias="endTime")

    # Email specific
    subject: str | None = None

    # Chat/Email direction
    direction: InteractionDirection | None = None

    # Notes attached to this interaction
    notes: list[NoteId] = Field(default_factory=list)

    # Logging info
    logging_type: InteractionLoggingType | None = Field(None, alias="loggingType")
    manual_creator_id: UserId | None = Field(None, alias="manualCreatorId")
    updated_at: ISODatetime | None = Field(None, alias="updatedAt")

InteractionAttendee

Bases: AffinityModel

A participant: an email address, and the Affinity person when one matches.

Source code in affinity/models/interactions_v2.py

class InteractionAttendee(AffinityModel):
    """A participant: an email address, and the Affinity person when one matches."""

    email_address: str | None = Field(None, alias="emailAddress")
    person: PersonSummary | None = None

InteractionCreate

Bases: AffinityModel

Data for creating an interaction (V1 API).

Note: The API only supports person IDs as participants. The Affinity UI's "Also add to... search for an entity" feature (associating an interaction with a company or opportunity) has no API equivalent.

Source code in affinity/models/secondary.py

class InteractionCreate(AffinityModel):
    """Data for creating an interaction (V1 API).

    Note: The API only supports person IDs as participants. The Affinity UI's
    "Also add to... search for an entity" feature (associating an interaction
    with a company or opportunity) has no API equivalent.
    """

    type: InteractionType
    person_ids: list[PersonId]
    content: str
    date: ISODatetime
    direction: InteractionDirection | None = None

InteractionUpdate

Bases: AffinityModel

Data for updating an interaction (V1 API).

Source code in affinity/models/secondary.py

class InteractionUpdate(AffinityModel):
    """Data for updating an interaction (V1 API)."""

    person_ids: list[PersonId] | None = None
    content: str | None = None
    date: ISODatetime | None = None
    direction: InteractionDirection | None = None

InvestorExecutiveConnection

Bases: AffinityModel

Someone you know (source, the investor) who invested in a company where the target was an executive.

Source code in affinity/models/discovery.py

class InvestorExecutiveConnection(AffinityModel):
    """Someone you know (``source``, the investor) who invested in a company where the target
    was an executive."""

    source: RelationshipPerson
    funding_round: FundingRound = Field(alias="fundingEvent")
    investor: DealRole
    executive: DealRole

    @classmethod
    def from_api(cls, data: dict[str, object]) -> InvestorExecutiveConnection:
        deal = data.get("dealContext") or {}
        assert isinstance(deal, dict)
        return cls.model_validate({"source": data.get("source"), **deal})

InvestorExecutiveConnectionGroup

Bases: AffinityModel

A target (an executive) and the investors you know who backed their company.

Source code in affinity/models/discovery.py

class InvestorExecutiveConnectionGroup(AffinityModel):
    """A target (an executive) and the investors you know who backed their company."""

    target: ConnectionTarget
    connections: list[InvestorExecutiveConnection]

    @classmethod
    def from_api(cls, data: dict[str, object]) -> InvestorExecutiveConnectionGroup:
        raw = data.get("connections") or []
        assert isinstance(raw, list)
        return cls(
            target=ConnectionTarget.model_validate(data.get("target")),
            connections=[InvestorExecutiveConnection.from_api(c) for c in raw],
        )

ListCreate

Bases: AffinityModel

A list to create (V2). The caller becomes its creator and owner.

owner_id and additional_permissions can only be set through V1: with either, the list is created through the V1 API.

Source code in affinity/models/entities.py

class ListCreate(AffinityModel):
    """A list to create (V2). The caller becomes its creator and owner.

    ``owner_id`` and ``additional_permissions`` can only be set through V1: with either, the
    list is created through the V1 API.
    """

    name: str
    type: ListType
    is_public: bool
    owner_id: UserId | None = None
    additional_permissions: list[ListPermission] = Field(default_factory=list)

ListEntry

Bases: AffinityModel

A row in a list, linking an entity to a list.

Holds the list-specific field values and which entity the row is for (entity_id, entity_type). entity (a summary of that entity) is filled only where the endpoint embeds it, e.g. a list's entries; it is None from companies.get_list_entries() and persons.get_list_entries().

Source code in affinity/models/entities.py

class ListEntry(AffinityModel):
    """
    A row in a list, linking an entity to a list.

    Holds the list-specific field values and which entity the row is for (``entity_id``,
    ``entity_type``). ``entity`` (a summary of that entity) is filled only where the endpoint
    embeds it, e.g. a list's entries; it is ``None`` from ``companies.get_list_entries()`` and
    ``persons.get_list_entries()``.
    """

    id: ListEntryId
    list_id: ListId = Field(alias="listId")
    list_name: str | None = Field(None, alias="listName")
    creator_id: UserId | None = Field(None, alias="creatorId")
    entity_id: int | None = Field(None, alias="entityId")
    entity_type: EntityType | None = Field(None, alias="entityType")
    created_at: ISODatetime = Field(alias="createdAt")

    # The entity this entry represents (can be Person, Company, or Opportunity)
    entity: PersonSummary | CompanySummary | OpportunitySummary | dict[str, Any] | None = None

    # Field values on this list entry (requested-vs-not-requested preserved)
    fields: FieldValues = Field(default_factory=FieldValues, alias="fields")
    fields_raw: list[dict[str, Any]] | None = Field(default=None, exclude=True)

    @model_validator(mode="before")
    @classmethod
    def _coerce_entity_by_entity_type(cls, value: Any) -> Any:
        """
        The v1 list-entry payload includes `entity_type` alongside a minimal `entity` dict.

        Some entity summaries overlap in shape (e.g. opportunity and company both have
        `{id, name}`), so we must use `entity_type` as the discriminator to avoid mis-parsing.
        """
        if not isinstance(value, Mapping):
            return value

        data: dict[str, Any] = dict(value)
        fields = data.get("fields")
        if isinstance(fields, list):
            data["fields_raw"] = fields

        entity = data.get("entity")
        if not isinstance(entity, Mapping):
            return data

        raw_entity_type = data.get("entityType")
        if raw_entity_type is None:
            raw_entity_type = data.get("entity_type")
        if raw_entity_type is None:
            return data

        try:
            entity_type = EntityType(raw_entity_type)
        except ValueError as e:
            _logger.debug("Unknown entity type %r, returning raw data: %s", raw_entity_type, e)
            return data

        if entity_type == EntityType.PERSON:
            try:
                data["entity"] = PersonSummary.model_validate(entity)
            except Exception as e:
                _logger.debug("Failed to validate PersonSummary, returning raw data: %s", e)
                return data
        elif entity_type == EntityType.ORGANIZATION:
            try:
                data["entity"] = CompanySummary.model_validate(entity)
            except Exception as e:
                _logger.debug("Failed to validate CompanySummary, returning raw data: %s", e)
                return data
        elif entity_type == EntityType.OPPORTUNITY:
            try:
                data["entity"] = OpportunitySummary.model_validate(entity)
            except Exception as e:
                _logger.debug("Failed to validate OpportunitySummary, returning raw data: %s", e)
                return data

        return data

    @model_validator(mode="after")
    def _mark_fields_not_requested_when_omitted(self) -> ListEntry:
        if "fields" not in self.__pydantic_fields_set__:
            self.fields.requested = False
        return self

ListEntryCreate

Bases: AffinityModel

Data for adding an entity to a list (V1 API).

Source code in affinity/models/entities.py

class ListEntryCreate(AffinityModel):
    """Data for adding an entity to a list (V1 API)."""

    entity_id: int
    creator_id: UserId | None = None

ListEntryWithEntity

Bases: AffinityModel

List entry with full entity data included (V2 response format).

Source code in affinity/models/entities.py

class ListEntryWithEntity(AffinityModel):
    """List entry with full entity data included (V2 response format)."""

    id: ListEntryId
    list_id: ListId = Field(alias="listId")
    creator: PersonSummary | None = None  # older payload shape
    creator_id: UserId | None = Field(None, alias="creatorId")  # V2 (null for some entries)
    created_at: ISODatetime = Field(alias="createdAt")

    # Entity type and data
    type: str  # "person", "company", or "opportunity"
    entity: Person | Company | Opportunity | None = None

    # Field values — always requested=False at entry level because the V2 API
    # nests field data inside entity.fields, not here. Use entry.entity.fields
    # to access field data, or pass entries to FieldResolver.get() which
    # auto-delegates to the inner entity.
    fields: FieldValues = Field(default_factory=FieldValues, alias="fields")
    fields_raw: list[dict[str, Any]] | None = Field(default=None, exclude=True)

    @model_validator(mode="before")
    @classmethod
    def _preserve_fields_raw_before(cls, value: Any) -> Any:
        value = _preserve_fields_raw(value)
        return cls._coerce_entity_by_type(value)

    @staticmethod
    def _coerce_entity_by_type(value: Any) -> Any:
        """Parse ``entity`` as the model ``type`` names. A minimal opportunity (``{id, name}``)
        also fits ``Company``, so the union alone can pick the wrong model (as for V1
        ``ListEntry``, which discriminates on ``entity_type``)."""
        if not isinstance(value, Mapping):
            return value
        entity = value.get("entity")
        raw_type = value.get("type")
        if not isinstance(entity, Mapping) or not isinstance(raw_type, str):
            return value
        models: dict[str, type[AffinityModel]] = {
            "person": Person,
            "company": Company,
            "organization": Company,
            "opportunity": Opportunity,
        }
        model = models.get(raw_type.lower())
        if model is None:
            return value
        data: dict[str, Any] = dict(value)
        try:
            data["entity"] = model.model_validate(entity)
        except ValidationError as e:
            _logger.debug("Failed to validate %s entity, using the raw data: %s", raw_type, e)
            return value
        return data

    @model_validator(mode="after")
    def _mark_fields_not_requested_when_omitted(self) -> ListEntryWithEntity:
        if "fields" not in self.__pydantic_fields_set__:
            self.fields.requested = False
        return self

ListPermission

Bases: AffinityModel

Additional permission on a list.

Source code in affinity/models/entities.py

class ListPermission(AffinityModel):
    """Additional permission on a list."""

    internal_person_id: UserId = Field(alias="internalPersonId")
    role_id: ListRole = Field(alias="roleId")

ListSummary

Bases: AffinityModel

Minimal list reference used by relationship endpoints.

Source code in affinity/models/entities.py

class ListSummary(AffinityModel):
    """Minimal list reference used by relationship endpoints."""

    id: ListId
    name: str | None = None
    type: ListType | None = None
    # V1 sends `public`, V2 `isPublic`; serialized as `public`.
    is_public: bool | None = Field(
        None, alias="public", validation_alias=AliasChoices("public", "isPublic")
    )
    owner_id: UserId | None = Field(None, alias="ownerId")
    creator_id: UserId | None = Field(None, alias="creatorId")
    list_size: int | None = Field(None, alias="listSize")
    created_at: ISODatetime | None = Field(None, alias="createdAt")

MeetingV2

Bases: _Event

A meeting (V2 GET /v2/meetings).

Source code in affinity/models/interactions_v2.py

class MeetingV2(_Event):
    """A meeting (V2 ``GET /v2/meetings``)."""

    organizer: InteractionAttendee | None = None

Note

Bases: AffinityModel

A note attached to one or more entities.

Notes can be plain text, HTML, or AI-generated meeting summaries.

Source code in affinity/models/secondary.py

class Note(AffinityModel):
    """
    A note attached to one or more entities.

    Notes can be plain text, HTML, or AI-generated meeting summaries.
    """

    id: NoteId
    creator_id: UserId = Field(alias="creatorId")
    content: str | None = None
    type: NoteType = NoteType.PLAIN_TEXT

    # Associated entities
    person_ids: list[PersonId] = Field(default_factory=list, alias="personIds")
    associated_person_ids: list[PersonId] = Field(default_factory=list, alias="associatedPersonIds")
    interaction_person_ids: list[PersonId] = Field(
        default_factory=list, alias="interactionPersonIds"
    )
    mentioned_person_ids: list[PersonId] = Field(default_factory=list, alias="mentionedPersonIds")
    company_ids: list[CompanyId] = Field(default_factory=list, alias="organizationIds")
    opportunity_ids: list[OpportunityId] = Field(default_factory=list, alias="opportunityIds")

    # Interaction association
    interaction_id: int | None = Field(None, alias="interactionId")
    interaction_type: InteractionType | None = Field(None, alias="interactionType")
    is_meeting: bool = Field(False, alias="isMeeting")

    # Thread support
    parent_id: NoteId | None = Field(None, alias="parentId")

    # Timestamps
    created_at: ISODatetime = Field(alias="createdAt")
    updated_at: ISODatetime | None = Field(None, alias="updatedAt")

NoteContent

Bases: AffinityModel

Note content in V2 format.

Source code in affinity/models/secondary.py

class NoteContent(AffinityModel):
    """Note content in V2 format."""

    html: str | None = None

NoteCreate

Bases: AffinityModel

A note to create (V2 API).

Attach it to at least one person, company or opportunity, or make it a reply with parent_id (a reply has no attachments of its own). type says how content is written: PLAIN_TEXT (default; escaped and wrapped in paragraphs) or HTML.

Source code in affinity/models/secondary.py

class NoteCreate(AffinityModel):
    """A note to create (V2 API).

    Attach it to at least one person, company or opportunity, or make it a reply with
    ``parent_id`` (a reply has no attachments of its own). ``type`` says how ``content`` is
    written: ``PLAIN_TEXT`` (default; escaped and wrapped in paragraphs) or ``HTML``.
    """

    content: str
    type: NoteType = NoteType.PLAIN_TEXT
    person_ids: list[PersonId] = Field(default_factory=list)
    company_ids: list[CompanyId] = Field(
        default_factory=list, validation_alias=AliasChoices("company_ids", "organization_ids")
    )
    opportunity_ids: list[OpportunityId] = Field(default_factory=list)
    parent_id: NoteId | None = None  # For reply notes
    creator_id: UserId | None = None
    created_at: ISODatetime | None = None

    @model_validator(mode="after")
    def _reply_has_no_entities(self) -> NoteCreate:
        if self.parent_id is not None and (
            self.person_ids or self.company_ids or self.opportunity_ids
        ):
            raise ValueError(
                "A reply (parent_id) can't have its own persons, companies or opportunities"
            )
        return self

NoteRef

Bases: AffinityModel

The note matched by a note search (note in a result).

Source code in affinity/models/search.py

class NoteRef(AffinityModel):
    """The note matched by a note search (``note`` in a result)."""

    id: NoteId
    # Open enum: an unknown kind is kept as-is instead of failing validation.
    kind: NoteKind | None = None

NoteSearchResult

Bases: AffinityModel

One note search hit: the note and its best-matching passage.

Source code in affinity/models/search.py

class NoteSearchResult(AffinityModel):
    """One note search hit: the note and its best-matching passage."""

    note: NoteRef
    # The matching passage (up to 2,000 characters; often the full note text).
    preview: str = ""

NoteUpdate

Bases: AffinityModel

Changes to a note: only what is set is sent.

type says how content is written (PLAIN_TEXT by default, or HTML), whatever the note was before; V2 stores both as HTML. person_ids/company_ids/ opportunity_ids replace the note's attachments of that kind ([] clears them; root notes only).

Source code in affinity/models/secondary.py

class NoteUpdate(AffinityModel):
    """Changes to a note: only what is set is sent.

    ``type`` says how ``content`` is written (``PLAIN_TEXT`` by default, or ``HTML``), whatever
    the note was before; V2 stores both as HTML. ``person_ids``/``company_ids``/
    ``opportunity_ids`` replace the note's attachments of that kind (``[]`` clears them; root
    notes only).
    """

    content: str | None = None
    type: NoteType = NoteType.PLAIN_TEXT
    person_ids: list[PersonId] | None = None
    company_ids: list[CompanyId] | None = None
    opportunity_ids: list[OpportunityId] | None = None

NoteV2

Bases: AffinityModel

V2 API note format.

replies_count and the attached companies / persons / opportunities (first ones, with *_total) are filled only when requested with includes (notes.feed / notes.get); otherwise they are None.

Source code in affinity/models/secondary.py

class NoteV2(AffinityModel):
    """V2 API note format.

    ``replies_count`` and the attached ``companies`` / ``persons`` / ``opportunities`` (first ones,
    with ``*_total``) are filled only when requested with ``includes`` (``notes.feed`` /
    ``notes.get``); otherwise they are ``None``.
    """

    id: NoteId
    # Discriminator: "entities", "interaction", "ai-notetaker", "ai-notetaker-reply", "user-reply"
    type: str
    content: NoteContent
    creator: PersonSummary | None = None
    created_at: ISODatetime = Field(alias="createdAt")
    updated_at: ISODatetime | None = Field(None, alias="updatedAt")

    # Parent for replies
    parent: dict[str, Any] | None = None

    # Mentions
    mentions: list[dict[str, Any]] = Field(default_factory=list)

    # Interaction / AI-notetaker notes: the meeting, and its transcript (None when the org
    # doesn't keep transcripts or it was deleted)
    interaction: dict[str, Any] | None = None
    transcript_id: int | None = Field(None, alias="transcriptId")

    # Only with includes=
    replies_count: int | None = Field(None, alias="repliesCount")
    companies: list[CompanySummary] | None = None
    companies_total: int | None = None
    persons: list[PersonSummary] | None = None
    persons_total: int | None = None
    opportunities: list[OpportunitySummary] | None = None
    opportunities_total: int | None = None

    @model_validator(mode="before")
    @classmethod
    def _flatten_previews(cls, data: Any) -> Any:
        if not isinstance(data, dict):
            return data
        for preview, key in (
            ("companiesPreview", "companies"),
            ("personsPreview", "persons"),
            ("opportunitiesPreview", "opportunities"),
        ):
            if isinstance(data.get(preview), dict):
                block = data[preview]
                data = {
                    **data,
                    key: block.get("data") or [],
                    f"{key}_total": block.get("totalCount"),
                }
                data.pop(preview, None)
        return data

Opportunity

Bases: AffinityModel

Deal/opportunity in a pipeline.

Note

The V2 API returns empty person_ids and company_ids arrays even when associations exist. Use client.opportunities.get_associated_person_ids() or client.opportunities.get_details() to retrieve association data.

See the opportunity-associations guide for details.

Source code in affinity/models/entities.py

class Opportunity(AffinityModel):
    """
    Deal/opportunity in a pipeline.

    Note:
        The V2 API returns empty ``person_ids`` and ``company_ids`` arrays even
        when associations exist. Use ``client.opportunities.get_associated_person_ids()``
        or ``client.opportunities.get_details()`` to retrieve association data.

        See the opportunity-associations guide for details.
    """

    id: OpportunityId
    name: str
    list_id: ListId | None = Field(None, alias="listId")
    list_name: str | None = Field(None, alias="listName")

    # Restricted Opportunities (V2). When is_redacted is True the requesting user cannot see
    # this opportunity and ``name`` is masked as "[Hidden]". V1 payloads never carry these,
    # so they default to False even though the V2 spec marks them required.
    is_restricted: bool = Field(False, alias="isRestricted")
    is_redacted: bool = Field(False, alias="isRedacted")

    # Associations (Note: V2 API returns empty arrays; use get_details() or
    # get_associated_person_ids() for populated data)
    person_ids: list[PersonId] = Field(default_factory=list, alias="personIds")
    # V1 responses (get_details, update) send organization_ids
    company_ids: list[CompanyId] = Field(
        default_factory=list,
        alias="organizationIds",
        validation_alias=AliasChoices(
            "organizationIds", "organization_ids", "companyIds", "company_ids"
        ),
    )

    # Field values (requested-vs-not-requested preserved)
    fields: FieldValues = Field(default_factory=FieldValues, alias="fields")
    fields_raw: list[dict[str, Any]] | None = Field(default=None, exclude=True)

    @model_validator(mode="before")
    @classmethod
    def _preserve_fields_raw_before(cls, value: Any) -> Any:
        return _preserve_fields_raw(value)

    # List entries
    list_entries: list[ListEntry] | None = Field(None, alias="listEntries")

    @model_validator(mode="after")
    def _post_validate(self) -> Opportunity:
        # Mark fields as not requested when omitted
        if "fields" not in self.__pydantic_fields_set__:
            self.fields.requested = False

        # Extract list_id from list_entries when not set directly
        # (V1 API returns list_id in list_entries, not at top level)
        if self.list_id is None and self.list_entries:
            self.list_id = self.list_entries[0].list_id

        return self

OpportunityCreate

Bases: AffinityModel

Data for creating a new opportunity (V1 API).

Source code in affinity/models/entities.py

class OpportunityCreate(AffinityModel):
    """Data for creating a new opportunity (V1 API)."""

    name: str
    list_id: ListId
    person_ids: list[PersonId] = Field(default_factory=list)
    company_ids: list[CompanyId] = Field(default_factory=list, alias="organization_ids")

OpportunityRef

Bases: AffinityModel

A bare opportunity reference ({"id": …}).

Source code in affinity/models/secondary.py

class OpportunityRef(AffinityModel):
    """A bare opportunity reference (``{"id": …}``)."""

    id: OpportunityId

OpportunitySummary

Bases: AffinityModel

Minimal opportunity data returned in nested contexts.

Source code in affinity/models/entities.py

class OpportunitySummary(AffinityModel):
    """Minimal opportunity data returned in nested contexts."""

    id: OpportunityId
    name: str

OpportunityUpdate

Bases: AffinityModel

Data for updating an opportunity (V1 API).

Source code in affinity/models/entities.py

class OpportunityUpdate(AffinityModel):
    """Data for updating an opportunity (V1 API)."""

    name: str | None = None
    person_ids: list[PersonId] | None = None
    company_ids: list[CompanyId] | None = Field(None, alias="organization_ids")

PageIterator

Bases: Generic[T]

Synchronous iterator that automatically fetches all pages.

Usage

for item in client.companies.all(): print(item.name)

Source code in affinity/models/pagination.py

class PageIterator(Generic[T]):
    """
    Synchronous iterator that automatically fetches all pages.

    Usage:
        for item in client.companies.all():
            print(item.name)
    """

    def __init__(
        self,
        fetch_page: Callable[[str | None], PaginatedResponse[T]],
        initial_cursor: str | None = None,
    ):
        self._fetch_page = fetch_page
        self._next_cursor = initial_cursor
        self._current_page: list[T] = []
        self._index = 0
        self._exhausted = False
        self._guard = _CursorGuard(initial_cursor)

    def __iter__(self) -> Iterator[T]:
        return self

    def __next__(self) -> T:
        while True:
            # If we have items in current page, return next
            if self._index < len(self._current_page):
                item = self._current_page[self._index]
                self._index += 1
                return item

            # Need to fetch next page
            if self._exhausted:
                raise StopIteration

            requested_url = self._next_cursor
            response = self._fetch_page(requested_url)
            self._current_page = list(response.data)
            self._next_cursor = response.next_cursor
            self._index = 0

            # Stop at the last page, or when paging can't progress (see _CursorGuard).
            if self._guard.last_page(requested_url, response):
                self._exhausted = True

            # Empty pages can still legitimately include nextUrl; keep paging
            # until we get data or the cursor is exhausted.
            if not self._current_page:
                if not self._exhausted:
                    continue
                raise StopIteration

    def pages(
        self,
        *,
        on_progress: Callable[[PaginationProgress], None] | None = None,
    ) -> Iterator[PaginatedResponse[T]]:
        """
        Iterate through pages (not individual items).

        Args:
            on_progress: Optional callback fired after fetching each page.
                Receives PaginationProgress with page_number, items_in_page,
                items_so_far, and has_next. Callbacks should be lightweight;
                heavy processing should happen outside the callback to avoid
                blocking iteration.

        Yields:
            PaginatedResponse objects for each page.

        Example:
            def report(p: PaginationProgress):
                print(f"Page {p.page_number}: {p.items_so_far} items so far")

            for page in client.persons.all().pages(on_progress=report):
                process(page.data)
        """
        page_number = 0
        items_so_far = 0
        guard = _CursorGuard(self._next_cursor)

        while True:
            requested_url = self._next_cursor
            response = self._fetch_page(requested_url)
            self._next_cursor = response.next_cursor
            page_number += 1
            items_in_page = len(response.data)
            items_so_far += items_in_page

            # Stop at the last page, or when paging can't progress (see _CursorGuard).
            last = guard.last_page(requested_url, response)

            if response.data:
                if on_progress:
                    on_progress(
                        PaginationProgress(
                            page_number=page_number,
                            items_in_page=items_in_page,
                            items_so_far=items_so_far,
                            has_next=not last,
                        )
                    )
                yield response

            if last:
                break

    def all(self, *, limit: int | None = _DEFAULT_LIMIT) -> list[T]:
        """
        Fetch all items across all pages into a list.

        Args:
            limit: Maximum items to fetch. Default 100,000. Set to None for unlimited.

        Returns:
            List of all items.

        Raises:
            TooManyResultsError: If results exceed limit.

        Note:
            The check occurs after extending results, so the final list may exceed
            limit by up to one page before the error is raised.

        Example:
            # Default - safe for most use cases
            persons = list(client.persons.all())  # Using iterator

            # Or use .all() method with limit check
            it = PageIterator(fetch_page)
            persons = it.all()  # Returns list, raises if > 100k

            # Explicit unlimited for large exports
            all_persons = it.all(limit=None)

            # Custom limit
            persons = it.all(limit=500_000)
        """
        results: list[T] = []

        for page in self.pages():
            results.extend(page.data)

            if limit is not None and len(results) > limit:
                raise TooManyResultsError(
                    f"Exceeded limit={limit:,} items. "
                    f"Use pages() for streaming, add a filter, or pass limit=None."
                )

        return results

all(*, limit: int | None = _DEFAULT_LIMIT) -> list[T]

Fetch all items across all pages into a list.

Parameters:

Name	Type	Description	Default
limit	int | None	Maximum items to fetch. Default 100,000. Set to None for unlimited.	_DEFAULT_LIMIT

Returns:

Type	Description
list[T]	List of all items.

Raises:

Type	Description
TooManyResultsError	If results exceed limit.

Note

The check occurs after extending results, so the final list may exceed limit by up to one page before the error is raised.

Example

Default - safe for most use cases

persons = list(client.persons.all()) # Using iterator

Or use .all() method with limit check

it = PageIterator(fetch_page) persons = it.all() # Returns list, raises if > 100k

Explicit unlimited for large exports

all_persons = it.all(limit=None)

Custom limit

persons = it.all(limit=500_000)

Source code in affinity/models/pagination.py

def all(self, *, limit: int | None = _DEFAULT_LIMIT) -> list[T]:
    """
    Fetch all items across all pages into a list.

    Args:
        limit: Maximum items to fetch. Default 100,000. Set to None for unlimited.

    Returns:
        List of all items.

    Raises:
        TooManyResultsError: If results exceed limit.

    Note:
        The check occurs after extending results, so the final list may exceed
        limit by up to one page before the error is raised.

    Example:
        # Default - safe for most use cases
        persons = list(client.persons.all())  # Using iterator

        # Or use .all() method with limit check
        it = PageIterator(fetch_page)
        persons = it.all()  # Returns list, raises if > 100k

        # Explicit unlimited for large exports
        all_persons = it.all(limit=None)

        # Custom limit
        persons = it.all(limit=500_000)
    """
    results: list[T] = []

    for page in self.pages():
        results.extend(page.data)

        if limit is not None and len(results) > limit:
            raise TooManyResultsError(
                f"Exceeded limit={limit:,} items. "
                f"Use pages() for streaming, add a filter, or pass limit=None."
            )

    return results

pages(*, on_progress: Callable[[PaginationProgress], None] | None = None) -> Iterator[PaginatedResponse[T]]

Iterate through pages (not individual items).

Parameters:

Name	Type	Description	Default
on_progress	Callable[[PaginationProgress], None] | None	Optional callback fired after fetching each page. Receives PaginationProgress with page_number, items_in_page, items_so_far, and has_next. Callbacks should be lightweight; heavy processing should happen outside the callback to avoid blocking iteration.	None

Yields:

Type	Description
PaginatedResponse[T]	PaginatedResponse objects for each page.

Example

def report(p: PaginationProgress): print(f"Page {p.page_number}: {p.items_so_far} items so far")

for page in client.persons.all().pages(on_progress=report): process(page.data)

Source code in affinity/models/pagination.py

def pages(
    self,
    *,
    on_progress: Callable[[PaginationProgress], None] | None = None,
) -> Iterator[PaginatedResponse[T]]:
    """
    Iterate through pages (not individual items).

    Args:
        on_progress: Optional callback fired after fetching each page.
            Receives PaginationProgress with page_number, items_in_page,
            items_so_far, and has_next. Callbacks should be lightweight;
            heavy processing should happen outside the callback to avoid
            blocking iteration.

    Yields:
        PaginatedResponse objects for each page.

    Example:
        def report(p: PaginationProgress):
            print(f"Page {p.page_number}: {p.items_so_far} items so far")

        for page in client.persons.all().pages(on_progress=report):
            process(page.data)
    """
    page_number = 0
    items_so_far = 0
    guard = _CursorGuard(self._next_cursor)

    while True:
        requested_url = self._next_cursor
        response = self._fetch_page(requested_url)
        self._next_cursor = response.next_cursor
        page_number += 1
        items_in_page = len(response.data)
        items_so_far += items_in_page

        # Stop at the last page, or when paging can't progress (see _CursorGuard).
        last = guard.last_page(requested_url, response)

        if response.data:
            if on_progress:
                on_progress(
                    PaginationProgress(
                        page_number=page_number,
                        items_in_page=items_in_page,
                        items_so_far=items_so_far,
                        has_next=not last,
                    )
                )
            yield response

        if last:
            break

PaginatedResponse

Bases: AffinityModel, Generic[T]

A paginated response from the API.

Provides access to the current page of results and pagination info. Works with all Affinity API endpoints transparently.

Attributes:

Name	Type	Description
data	list[T]	List of items in the current page.
has_next	bool	Whether more pages are available.
next_cursor	str | None	Cursor for fetching the next page (use this for pagination).

Example

page = client.companies.list(limit=100)
while page.has_next:
    process(page.data)
    page = client.companies.list(limit=100, cursor=page.next_cursor)

Tip

Always use next_cursor for pagination. Avoid accessing pagination.next_cursor or next_page_token directly.

Source code in affinity/models/pagination.py

class PaginatedResponse(AffinityModel, Generic[T]):
    """
    A paginated response from the API.

    Provides access to the current page of results and pagination info.
    Works with all Affinity API endpoints transparently.

    Attributes:
        data: List of items in the current page.
        has_next: Whether more pages are available.
        next_cursor: Cursor for fetching the next page (use this for pagination).

    Example:
        ```python
        page = client.companies.list(limit=100)
        while page.has_next:
            process(page.data)
            page = client.companies.list(limit=100, cursor=page.next_cursor)
        ```

    Tip:
        Always use ``next_cursor`` for pagination. Avoid accessing
        ``pagination.next_cursor`` or ``next_page_token`` directly.
    """

    data: list[T] = Field(default_factory=list)
    pagination: PaginationInfo = Field(default_factory=PaginationInfo)
    # Internal: V1 API compatibility field. Users should use next_cursor property.
    next_page_token: str | None = Field(None, alias="nextPageToken")
    _has_next_override: bool | None = PrivateAttr(default=None)
    _filter_stats: FilterStats | None = PrivateAttr(default=None)

    @property
    def total_count(self) -> int | None:
        """Total number of matching items, for endpoints that report it (else ``None``)."""
        return getattr(self.pagination, "total_count", None)

    def __len__(self) -> int:
        """Number of items in current page."""
        return len(self.data)

    @property
    def has_next(self) -> bool:
        """Whether there are more pages."""
        if self._has_next_override is not None:
            return self._has_next_override
        return self.next_cursor is not None

    @property
    def next_cursor(self) -> str | None:
        """
        Cursor for the next page, if any.

        Returns the V2 pagination URL or V1 page token, whichever is present.
        Always use this property instead of accessing ``pagination.next_cursor``
        or ``next_page_token`` directly.
        """
        # Use explicit None check to preserve empty strings (which are valid cursors)
        if self.pagination.next_cursor is not None:
            return self.pagination.next_cursor
        return self.next_page_token

    @property
    def filter_stats(self) -> FilterStats | None:
        """Stats for client-side filtered queries (scanned/matched counts)."""
        return self._filter_stats

filter_stats: FilterStats | None property

Stats for client-side filtered queries (scanned/matched counts).

has_next: bool property

Whether there are more pages.

next_cursor: str | None property

Cursor for the next page, if any.

Returns the V2 pagination URL or V1 page token, whichever is present. Always use this property instead of accessing pagination.next_cursor or next_page_token directly.

total_count: int | None property

Total number of matching items, for endpoints that report it (else None).

__len__() -> int

Number of items in current page.

Source code in affinity/models/pagination.py

def __len__(self) -> int:
    """Number of items in current page."""
    return len(self.data)

PaginationInfo

Bases: AffinityModel

V2 pagination info returned in responses.

Source code in affinity/models/pagination.py

class PaginationInfo(AffinityModel):
    """V2 pagination info returned in responses."""

    next_cursor: str | None = Field(None, alias="nextUrl")
    prev_cursor: str | None = Field(None, alias="prevUrl")

PaginationProgress dataclass

Progress information for pagination callbacks.

Source code in affinity/models/pagination.py

@dataclass
class PaginationProgress:
    """Progress information for pagination callbacks."""

    page_number: int
    """1-indexed page number."""

    items_in_page: int
    """Items in current page."""

    items_so_far: int
    """Cumulative items *including* just-yielded page."""

    has_next: bool
    """Whether more pages exist (matches Page.has_next)."""

has_next: bool instance-attribute

Whether more pages exist (matches Page.has_next).

items_in_page: int instance-attribute

Items in current page.

items_so_far: int instance-attribute

Cumulative items including just-yielded page.

page_number: int instance-attribute

1-indexed page number.

Person

Bases: AffinityModel

Full person representation.

Note: Companies are called Organizations in V1 API.

Source code in affinity/models/entities.py

class Person(AffinityModel):
    """
    Full person representation.

    Note: Companies are called Organizations in V1 API.
    """

    id: PersonId
    first_name: str | None = Field(None, alias="firstName")
    last_name: str | None = Field(None, alias="lastName")
    primary_email: str | None = Field(None, alias="primaryEmailAddress")
    # V2 uses emailAddresses, V1 uses emails - accept both via alias
    emails: list[str] = Field(default_factory=list, alias="emailAddresses")
    type: PersonType = PersonType.EXTERNAL

    @field_validator("type", mode="before")
    @classmethod
    def _coerce_person_type(cls, value: Any) -> Any:
        return _normalize_person_type(value)

    # Associations (V1 sends organization_ids, some payloads organizationIds)
    company_ids: list[CompanyId] = Field(
        default_factory=list,
        alias="organizationIds",
        validation_alias=AliasChoices(
            "organizationIds", "organization_ids", "companyIds", "company_ids"
        ),
    )
    opportunity_ids: list[OpportunityId] = Field(default_factory=list, alias="opportunityIds")

    # V1: only returned when `with_current_organizations=true`
    current_company_ids: list[CompanyId] = Field(
        default_factory=list,
        alias="currentOrganizationIds",
        validation_alias=AliasChoices(
            "currentOrganizationIds",
            "current_organization_ids",
            "currentCompanyIds",
            "current_company_ids",
        ),
    )

    # Field values (requested-vs-not-requested preserved)
    fields: FieldValues = Field(default_factory=FieldValues, alias="fields")
    fields_raw: list[dict[str, Any]] | None = Field(default=None, exclude=True)

    @model_validator(mode="before")
    @classmethod
    def _normalize_null_lists_before(cls, value: Any) -> Any:
        value = _normalize_null_lists(
            value,
            (
                "emails",
                "emailAddresses",
                "companyIds",
                "company_ids",
                "organizationIds",
                "organization_ids",
                "currentCompanyIds",
                "current_company_ids",
                "currentOrganizationIds",
                "current_organization_ids",
                "opportunityIds",
                "opportunity_ids",
            ),
        )
        return _preserve_fields_raw(value)

    @model_validator(mode="after")
    def _mark_fields_not_requested_when_omitted(self) -> Person:
        if "fields" not in self.__pydantic_fields_set__:
            self.fields.requested = False
        return self

    # Interaction dates (V1 format, returned when with_interaction_dates=True)
    interaction_dates: InteractionDates | None = Field(None, alias="interactionDates")

    # Detailed interaction data with person IDs (returned when with_interaction_persons=True)
    interactions: Interactions | None = None

    # List entries (returned for single person fetch)
    list_entries: list[ListEntry] | None = Field(None, alias="listEntries")

    @property
    def full_name(self) -> str:
        """Get the person's full name."""
        parts = [self.first_name, self.last_name]
        return " ".join(p for p in parts if p) or ""

full_name: str property

Get the person's full name.

PersonCreate

Bases: AffinityModel

Data for creating a new person.

Note: "Current Organization" and "Current Job Title" cannot be set during creation. "Current Organization" is a derived/system-managed field (read-only via API, driven by enrichment and email domain). "Current Job Title" can be updated after creation via the field-values endpoint.

Source code in affinity/models/entities.py

class PersonCreate(AffinityModel):
    """Data for creating a new person.

    Note: "Current Organization" and "Current Job Title" cannot be set during
    creation. "Current Organization" is a derived/system-managed field (read-only
    via API, driven by enrichment and email domain). "Current Job Title" can be
    updated after creation via the field-values endpoint.
    """

    first_name: str
    last_name: str
    emails: list[str] = Field(default_factory=list)
    company_ids: list[CompanyId] = Field(default_factory=list, alias="organization_ids")

PersonDuplicateSuggestion

Bases: AffinityModel

Two person profiles Affinity thinks are the same person.

Merge duplicates[0] into primary with persons.merge(primary.id, duplicate.id). match_criteria: intelligent (medium confidence: email matching across sources), name (low).

Source code in affinity/models/discovery.py

class PersonDuplicateSuggestion(AffinityModel):
    """Two person profiles Affinity thinks are the same person.

    Merge ``duplicates[0]`` into ``primary`` with ``persons.merge(primary.id, duplicate.id)``.
    ``match_criteria``: ``intelligent`` (medium confidence: email matching across sources),
    ``name`` (low).
    """

    id: str
    primary: Person = Field(alias="primaryProfile")
    duplicates: list[Person] = Field(alias="duplicateProfiles")
    match_criteria: str = Field(alias="matchCriteria")

PersonMergeState

Bases: AffinityModel

One person merge (V2 GET /v2/person-merges).

Source code in affinity/models/secondary.py

class PersonMergeState(AffinityModel):
    """One person merge (V2 ``GET /v2/person-merges``)."""

    id: int
    status: MergeStatus
    task_id: str = Field(alias="taskId")
    started_at: ISODatetime = Field(alias="startedAt")
    primary_person_id: PersonId = Field(alias="primaryPersonId")
    duplicate_person_id: PersonId = Field(alias="duplicatePersonId")
    completed_at: ISODatetime | None = Field(None, alias="completedAt")
    error_message: str | None = Field(None, alias="errorMessage")

PersonRef

Bases: AffinityModel

A bare person reference ({"id": …}).

Source code in affinity/models/secondary.py

class PersonRef(AffinityModel):
    """A bare person reference (``{"id": …}``)."""

    id: PersonId

PersonUpdate

Bases: AffinityModel

Data for updating a person (V1 API).

Source code in affinity/models/entities.py

class PersonUpdate(AffinityModel):
    """Data for updating a person (V1 API)."""

    first_name: str | None = None
    last_name: str | None = None
    emails: list[str] | None = None
    company_ids: list[CompanyId] | None = Field(None, alias="organization_ids")

RateLimitBucket

Bases: AffinityModel

A single rate limit bucket (quota window).

Source code in affinity/models/rate_limit_snapshot.py

class RateLimitBucket(AffinityModel):
    """A single rate limit bucket (quota window)."""

    limit: int | None = None
    remaining: int | None = None
    reset_seconds: int | None = Field(None, alias="resetSeconds")
    used: int | None = None

RateLimitInfo

Bases: AffinityModel

Rate limit information for an API key.

Source code in affinity/models/secondary.py

class RateLimitInfo(AffinityModel):
    """Rate limit information for an API key."""

    limit: int
    remaining: int
    reset: int  # Seconds until reset
    used: int

RateLimitSnapshot

Bases: AffinityModel

A best-effort snapshot of rate limit state.

Notes: - source="headers" means the snapshot is derived from tracked HTTP response headers. - source="endpoint" means the snapshot is derived from a dedicated endpoint response payload. - source="unknown" means no reliable rate limit information has been observed yet.

Source code in affinity/models/rate_limit_snapshot.py

class RateLimitSnapshot(AffinityModel):
    """
    A best-effort snapshot of rate limit state.

    Notes:
    - `source="headers"` means the snapshot is derived from tracked HTTP response headers.
    - `source="endpoint"` means the snapshot is derived from a dedicated endpoint response payload.
    - `source="unknown"` means no reliable rate limit information has been observed yet.
    """

    api_key_per_minute: RateLimitBucket = Field(
        default_factory=RateLimitBucket, alias="apiKeyPerMinute"
    )
    org_monthly: RateLimitBucket = Field(default_factory=RateLimitBucket, alias="orgMonthly")
    concurrent: RateLimitBucket | None = None

    observed_at: datetime | None = Field(None, alias="observedAt")
    age_seconds: float | None = Field(None, alias="ageSeconds")
    source: RateLimitSource = "unknown"
    request_id: str | None = Field(None, alias="requestId")

RateLimits

Bases: AffinityModel

Current rate limit status.

Source code in affinity/models/secondary.py

class RateLimits(AffinityModel):
    """Current rate limit status."""

    # Absent when no monthly quota applies (documented for v2 `orgPerMonth`; v1 and v2 share
    # one request pool, so v1's `org_monthly` is treated the same way).
    org_monthly: RateLimitInfo | None = Field(
        None,
        alias="orgMonthly",
        validation_alias=AliasChoices("orgPerMonth", "orgMonthly", "org_monthly"),
    )
    # V2 `callerPerMinute`, V1 `api_key_per_minute`
    api_key_per_minute: RateLimitInfo = Field(
        alias="apiKeyPerMinute",
        validation_alias=AliasChoices("callerPerMinute", "apiKeyPerMinute", "api_key_per_minute"),
    )

Relationship

Bases: AffinityModel

How strongly two people are connected.

interaction_score runs from 0.0 to 1.0 (higher is stronger); a relationship known only from LinkedIn has score 0 and linkedin_connected_on set.

Source code in affinity/models/relationships_v2.py

class Relationship(AffinityModel):
    """How strongly two people are connected.

    ``interaction_score`` runs from 0.0 to 1.0 (higher is stronger); a relationship known only
    from LinkedIn has score 0 and ``linkedin_connected_on`` set.
    """

    person1: RelationshipPerson
    person2: RelationshipPerson
    interaction_score: float = Field(alias="interactionScore")
    linkedin_connected_on: date | None = None

    @model_validator(mode="before")
    @classmethod
    def _flatten_linkedin(cls, data: Any) -> Any:
        if isinstance(data, dict) and "linkedIn" in data:
            linked = data.get("linkedIn")
            data = {**data, "linkedin_connected_on": (linked or {}).get("connectedOn")}
            data.pop("linkedIn", None)
        return data

RelationshipPerson

Bases: AffinityModel

One side of a relationship.

Source code in affinity/models/relationships_v2.py

class RelationshipPerson(AffinityModel):
    """One side of a relationship."""

    id: int
    first_name: str | None = Field(None, alias="firstName")
    last_name: str | None = Field(None, alias="lastName")
    primary_email_address: str | None = Field(None, alias="primaryEmailAddress")

RelationshipStrength

Bases: AffinityModel

Relationship strength between internal and external persons.

Source code in affinity/models/secondary.py

class RelationshipStrength(AffinityModel):
    """Relationship strength between internal and external persons."""

    internal_id: UserId = Field(alias="internalId")
    external_id: PersonId = Field(alias="externalId")
    strength: float  # 0.0 to 1.0

Reminder

Bases: AffinityModel

A reminder (V2 shape). At most one of person, company, opportunity is set.

Source code in affinity/models/secondary.py

class Reminder(AffinityModel):
    """A reminder (V2 shape). At most one of ``person``, ``company``, ``opportunity`` is set."""

    id: ReminderIdType
    type: ReminderType
    # active / overdue / completed (a recurring reminder is never completed: completing it
    # moves due_date forward)
    status: ReminderStatus
    content: str | None = None
    due_date: ISODatetime = Field(alias="dueDate")
    # None for one-time reminders
    recurrence: ReminderRecurrence | None = None

    creator: PersonRef | None = None
    owner: PersonRef | None = None
    completer: PersonRef | None = None

    person: PersonRef | None = None
    company: CompanyRef | None = None
    opportunity: OpportunityRef | None = None

    created_at: ISODatetime = Field(alias="createdAt")
    updated_at: ISODatetime | None = Field(None, alias="updatedAt")
    completed_at: ISODatetime | None = Field(None, alias="completedAt")

    @property
    def reset_type(self) -> ReminderResetType | None:
        """``recurrence.reset_type`` (None for one-time reminders)."""
        return self.recurrence.reset_type if self.recurrence else None

    @property
    def reminder_days(self) -> int | None:
        """``recurrence.period_days`` (None for one-time reminders)."""
        return self.recurrence.period_days if self.recurrence else None

reminder_days: int | None property

recurrence.period_days (None for one-time reminders).

reset_type: ReminderResetType | None property

recurrence.reset_type (None for one-time reminders).

ReminderCreate

Bases: AffinityModel

A reminder to create.

One-time reminders need due_date; recurring ones need reset_type and reminder_days (1-3650; due_date optional). Attach it to exactly one of person_id, company_id, opportunity_id.

Source code in affinity/models/secondary.py

class ReminderCreate(AffinityModel):
    """A reminder to create.

    One-time reminders need ``due_date``; recurring ones need ``reset_type`` and
    ``reminder_days`` (1-3650; ``due_date`` optional). Attach it to exactly one of
    ``person_id``, ``company_id``, ``opportunity_id``.
    """

    owner_id: UserId
    type: ReminderType
    content: str | None = None
    due_date: ISODatetime | None = None
    reset_type: ReminderResetType | None = None
    reminder_days: int | None = None

    person_id: PersonId | None = None
    company_id: CompanyId | None = Field(
        None, validation_alias=AliasChoices("company_id", "organization_id")
    )
    opportunity_id: OpportunityId | None = None

    @model_validator(mode="after")
    def _one_entity(self) -> ReminderCreate:
        given = [x for x in (self.person_id, self.company_id, self.opportunity_id) if x is not None]
        if len(given) > 1:
            raise ValueError("A reminder can be attached to one entity only")
        return self

ReminderRecurrence

Bases: AffinityModel

How a recurring reminder repeats: every period_days days, reset by reset_type.

Source code in affinity/models/secondary.py

class ReminderRecurrence(AffinityModel):
    """How a recurring reminder repeats: every ``period_days`` days, reset by ``reset_type``."""

    period_days: int = Field(alias="periodDays")
    reset_type: ReminderResetType = Field(alias="resetTrigger")

ReminderUpdate

Bases: AffinityModel

Changes to a reminder: fields left as None are not changed.

clear_content=True removes the content. is_completed=True completes the reminder (a recurring one moves to its next due date); False reopens a one-time reminder. Setting type sends the whole update through the V1 API (V2 can't change the type).

Source code in affinity/models/secondary.py

class ReminderUpdate(AffinityModel):
    """Changes to a reminder: fields left as ``None`` are not changed.

    ``clear_content=True`` removes the content. ``is_completed=True`` completes the reminder
    (a recurring one moves to its next due date); ``False`` reopens a one-time reminder.
    Setting ``type`` sends the whole update through the V1 API (V2 can't change the type).
    """

    owner_id: UserId | None = None
    type: ReminderType | None = None
    content: str | None = None
    clear_content: bool = False
    due_date: ISODatetime | None = None
    reset_type: ReminderResetType | None = None
    reminder_days: int | None = None
    is_completed: bool | None = None

    @model_validator(mode="after")
    def _content_or_clear(self) -> ReminderUpdate:
        if self.clear_content and self.content is not None:
            raise ValueError("Pass content or clear_content=True, not both")
        return self

SavedView

Bases: AffinityModel

A saved view configuration for a list.

Source code in affinity/models/entities.py

class SavedView(AffinityModel):
    """A saved view configuration for a list."""

    id: SavedViewId
    name: str
    type: str | None = None  # V2 field: view type
    list_id: ListId | None = Field(None, alias="listId")
    # The Affinity API does not consistently include this field.
    is_default: bool | None = Field(None, alias="isDefault")
    created_at: ISODatetime | None = Field(None, alias="createdAt")

    # Field IDs included in this view
    field_ids: list[str] = Field(default_factory=list, alias="fieldIds")

SemanticCompany

Bases: AffinityModel

A company returned by semantic search.

Source code in affinity/models/search.py

class SemanticCompany(AffinityModel):
    """A company returned by semantic search."""

    id: CompanyId
    name: str = ""
    domain: str | None = None
    domains: list[str] = Field(default_factory=list)
    is_global: bool = Field(False, alias="isGlobal")
    # Raw relevance score as sent by the API (a decimal string such as "0.85").
    score: str | None = None

    @property
    def score_float(self) -> float | None:
        """The score as a float, or None when missing or not a number."""
        if self.score is None:
            return None
        try:
            return float(self.score)
        except (TypeError, ValueError):
            return None

score_float: float | None property

The score as a float, or None when missing or not a number.

SemanticSearchResult

Bases: AffinityModel

Semantic search response: matches plus the API's explanation of the search.

Source code in affinity/models/search.py

class SemanticSearchResult(AffinityModel):
    """Semantic search response: matches plus the API's explanation of the search."""

    explanation: str | None = None
    entity_type: str = Field("companies", alias="entityType")
    data: list[SemanticCompany] = Field(default_factory=list)

Team

Bases: AffinityModel

A team (V2).

members and accessible_lists (the first ones, up to 10, with *_total) are filled only when requested with includes; otherwise they are None. privacy_type needs the "Manage Teams" permission and team privacy settings; otherwise it is None.

Source code in affinity/models/org.py

class Team(AffinityModel):
    """A team (V2).

    ``members`` and ``accessible_lists`` (the first ones, up to 10, with ``*_total``) are filled
    only when requested with ``includes``; otherwise they are ``None``. ``privacy_type`` needs
    the "Manage Teams" permission and team privacy settings; otherwise it is ``None``.
    """

    id: int
    name: str
    created_at: ISODatetime | None = Field(None, alias="createdAt")
    updated_at: ISODatetime | None = Field(None, alias="updatedAt")
    privacy_type: str | None = Field(None, alias="privacyType")

    # Read from membersPreview / accessibleListsPreview ({data, totalCount}), flattened below
    members: list[TeamMember] | None = Field(
        None, validation_alias=AliasChoices("members", "membersPreview")
    )
    members_total: int | None = None
    accessible_lists: list[ListSummary] | None = Field(
        None, validation_alias=AliasChoices("accessible_lists", "accessibleListsPreview")
    )
    accessible_lists_total: int | None = None

    @model_validator(mode="before")
    @classmethod
    def _flatten_previews(cls, data: Any) -> Any:
        if not isinstance(data, dict):
            return data
        data = dict(data)
        for preview, key in (
            ("membersPreview", "members"),
            ("accessibleListsPreview", "accessible_lists"),
        ):
            block = data.pop(preview, None)
            if isinstance(block, dict):
                data[key] = block.get("data", [])
                data[f"{key}_total"] = block.get("totalCount")
        return data

TeamMember

Bases: AffinityModel

A member of a team (from Team.members).

Source code in affinity/models/org.py

class TeamMember(AffinityModel):
    """A member of a team (from ``Team.members``)."""

    # The membership's id (not the user's)
    id: int
    user: User
    added_at: ISODatetime = Field(alias="addedAt")
    # Only with the "Manage Teams" permission and team-based privacy enabled
    privacy_type: str | None = Field(None, alias="privacyType")

Tenant

Bases: AffinityModel

Affinity tenant (organization/team) information.

Source code in affinity/models/secondary.py

class Tenant(AffinityModel):
    """Affinity tenant (organization/team) information."""

    id: TenantId
    name: str
    subdomain: str

Transcript

Bases: AffinityModel

A meeting transcript.

note is the AI Notetaker note it belongs to (with interaction, the meeting). fragments_preview holds the first fragments (only from transcripts.get(); the list endpoint returns metadata only) and fragments_total how many there are in all; use transcripts.fragments() / iter_fragments() for the rest.

Source code in affinity/models/transcripts.py

class Transcript(AffinityModel):
    """A meeting transcript.

    ``note`` is the AI Notetaker note it belongs to (with ``interaction``, the meeting).
    ``fragments_preview`` holds the first fragments (only from ``transcripts.get()``; the list
    endpoint returns metadata only) and ``fragments_total`` how many there are in all; use
    ``transcripts.fragments()`` / ``iter_fragments()`` for the rest.
    """

    id: int
    created_at: ISODatetime = Field(alias="createdAt")
    language_code: str | None = Field(None, alias="languageCode")
    note: NoteV2 | None = None
    fragments_preview: list[TranscriptFragment] = Field(default_factory=list)
    fragments_total: int | None = None

    @model_validator(mode="before")
    @classmethod
    def _flatten_preview(cls, data: Any) -> Any:
        if isinstance(data, dict) and isinstance(data.get("fragmentsPreview"), dict):
            preview = data["fragmentsPreview"]
            data = {
                **data,
                "fragments_preview": preview.get("data") or [],
                "fragments_total": preview.get("totalCount"),
            }
            data.pop("fragmentsPreview", None)
        return data

TranscriptFragment

Bases: AffinityModel

One piece of dialogue. Timestamps are offsets from the start, e.g. "00:00:06".

Source code in affinity/models/transcripts.py

class TranscriptFragment(AffinityModel):
    """One piece of dialogue. Timestamps are offsets from the start, e.g. ``"00:00:06"``."""

    content: str
    speaker: str | None = None
    start_timestamp: str | None = Field(None, alias="startTimestamp")
    end_timestamp: str | None = Field(None, alias="endTimestamp")

UserAccount

Bases: AffinityModel

An internal user of the organization (V2 UserData).

A user's id is also their person id: persons.get(PersonId(user.id)) works. emails and role are filled only for callers with the "Manage Users" permission.

Source code in affinity/models/org.py

class UserAccount(AffinityModel):
    """An internal user of the organization (V2 ``UserData``).

    A user's id is also their person id: ``persons.get(PersonId(user.id))`` works.
    ``emails`` and ``role`` are filled only for callers with the "Manage Users" permission.
    """

    id: UserId
    first_name: str = Field(alias="firstName")
    last_name: str | None = Field(None, alias="lastName")
    primary_email: str | None = Field(None, alias="primaryEmailAddress")
    emails: list[str] | None = Field(None, alias="emailAddresses")
    photo_url: str | None = Field(None, alias="photoUrl")
    status: UserStatus
    role: str | None = None

UserStatus

Bases: OpenStrEnum

Whether a user's account is in use.

Source code in affinity/models/org.py

class UserStatus(OpenStrEnum):
    """Whether a user's account is in use."""

    ACTIVE = "active"
    INVITED = "invited"  # invited, hasn't accepted yet
    DEACTIVATED = "deactivated"

Webhook

Bases: AffinityModel

A webhook: event notifications delivered to an external URL (V2 webhooks.Webhook).

Source code in affinity/models/secondary.py

class Webhook(AffinityModel):
    """A webhook: event notifications delivered to an external URL (V2 ``webhooks.Webhook``)."""

    id: WebhookId
    url: str
    # An empty list means the webhook receives every event type.
    subscriptions: list[WebhookEvent] = Field(default_factory=list)
    status: WebhookStatus
    # None when Affinity has no record of the creator.
    creator: PersonRef | None = None
    created_at: ISODatetime = Field(alias="createdAt")
    # None until the webhook is first updated.
    updated_at: ISODatetime | None = Field(None, alias="updatedAt")

WebhookCreate

Bases: AffinityModel

A webhook to create.

subscriptions is required. An empty list subscribes the webhook to every event type, so every change in the organization is sent to url; name the events you need.

Source code in affinity/models/secondary.py

class WebhookCreate(AffinityModel):
    """A webhook to create.

    ``subscriptions`` is required. **An empty list subscribes the webhook to every event type**,
    so every change in the organization is sent to ``url``; name the events you need.
    """

    url: str
    subscriptions: list[WebhookEvent]

    @model_validator(mode="before")
    @classmethod
    def _no_v1_fields(cls, data: Any) -> Any:
        return _reject_v1_fields(data, {"webhook_url": "url="})

WebhookUpdate

Bases: AffinityModel

Changes to a webhook: only the fields you set are sent.

subscriptions=[] subscribes the webhook to every event type.

Source code in affinity/models/secondary.py

class WebhookUpdate(AffinityModel):
    """Changes to a webhook: only the fields you set are sent.

    ``subscriptions=[]`` subscribes the webhook to every event type.
    """

    url: str | None = None
    subscriptions: list[WebhookEvent] | None = None
    status: WebhookStatus | None = None

    @model_validator(mode="before")
    @classmethod
    def _no_v1_fields(cls, data: Any) -> Any:
        return _reject_v1_fields(
            data, {"webhook_url": "url=", "disabled": "status=WebhookStatus.DISABLED (or ACTIVE)"}
        )

WhoAmI

Bases: AffinityModel

Response from whoami endpoint.

Source code in affinity/models/secondary.py

class WhoAmI(AffinityModel):
    """Response from whoami endpoint."""

    tenant: Tenant
    user: User
    grant: Grant

FieldResolver

Helper for extracting field values by name from entities.

Caches field metadata internally for efficient repeated lookups.

Example

Fetch field metadata once

resolver = FieldResolver(client.companies.get_fields())

Use throughout your code

for company in companies: ... status = resolver.get(company, "Status") ... industry = resolver.get(company, "Industry") ... print(f"{company.name}: {status}, {industry}")

Source code in affinity/field_resolver.py

class FieldResolver:
    """
    Helper for extracting field values by name from entities.

    Caches field metadata internally for efficient repeated lookups.

    Example:
        >>> # Fetch field metadata once
        >>> resolver = FieldResolver(client.companies.get_fields())
        >>>
        >>> # Use throughout your code
        >>> for company in companies:
        ...     status = resolver.get(company, "Status")
        ...     industry = resolver.get(company, "Industry")
        ...     print(f"{company.name}: {status}, {industry}")
    """

    def __init__(self, fields: Sequence[FieldMetadata]) -> None:
        """
        Create a resolver from field metadata.

        Args:
            fields: Field metadata from client.companies.get_fields(),
                    client.persons.get_fields(), etc.
        """
        # Build case-insensitive name -> field(s) mapping.
        # Multiple fields can share the same display name (e.g., enrichment
        # fields from different sources both called "Description").
        self._by_name: dict[str, list[FieldMetadata]] = {}
        self._by_id: dict[str, FieldMetadata] = {}
        self._warned_names: set[str] = set()

        for field in fields:
            name_key = field.name.casefold()
            self._by_name.setdefault(name_key, []).append(field)

            # Register source-qualified key for enrichment fields (e.g., "dealroom:description")
            if field.enrichment_source:
                qualified_key = f"{field.enrichment_source}:{name_key}"
                self._by_name.setdefault(qualified_key, []).append(field)

            self._by_id[str(field.id)] = field

    def get(
        self,
        entity: Company | Person | Opportunity | ListEntry | ListEntryWithEntity,
        field_name: str,
        *,
        resolve: ResolveMode = ResolveMode.RAW,
    ) -> Any:
        """
        Get a field value by name from an entity.

        Args:
            entity: The entity (Company, Person, Opportunity, or ListEntry)
            field_name: The field display name (case-insensitive)
            resolve: ResolveMode.RAW returns extracted values as-is.
                     ResolveMode.TEXT resolves all complex types to strings
                     (dropdowns to text, persons to names, etc.).

        Returns:
            The extracted field value, or None if field not found or not populated.

        Example:
            >>> status = resolver.get(company, "Status")
            >>> print(status)  # "Active"

            >>> # Resolve complex types to text
            >>> stage = resolver.get(opportunity, "Stage", resolve=ResolveMode.TEXT)
            >>> print(stage)  # "Negotiation" instead of 12345
        """
        fields = _field_values(entity)
        if not fields.requested:
            warnings.warn(
                "Fields were not requested for this entity. "
                "Pass field_ids or field_types when fetching to populate field values.",
                UserWarning,
                stacklevel=2,
            )
            return None

        field = self._resolve_name(field_name, strict=False)
        if field is None:
            return None

        value = fields.get_value(str(field.id))

        if resolve == ResolveMode.TEXT and value is not None:
            value = self._resolve_value(value, field.value_type)

        return value

    def get_many(
        self,
        entity: Company | Person | Opportunity | ListEntry | ListEntryWithEntity,
        field_names: Sequence[str],
        *,
        resolve: ResolveMode = ResolveMode.RAW,
    ) -> dict[str, Any]:
        """
        Get multiple field values by name.

        Args:
            entity: The entity (Company, Person, Opportunity, or ListEntry)
            field_names: List of field names to extract
            resolve: ResolveMode.RAW or ResolveMode.TEXT

        Returns:
            Dict mapping field_name -> value (None for missing/unpopulated fields)

        Example:
            >>> values = resolver.get_many(company, ["Status", "Industry", "Size"])
            >>> print(values)
            {'Status': 'Active', 'Industry': 'Tech', 'Size': None}
        """
        return {name: self.get(entity, name, resolve=resolve) for name in field_names}

    def find_field(self, field_name: str) -> FieldMetadata | None:
        """
        Find field metadata by name (case-insensitive).

        Useful for getting the FieldId needed by write operations like
        ``update_field_value()``. Supports source-qualified names like
        ``'dealroom:Description'`` for ambiguous fields.

        Returns:
            The FieldMetadata, or None if no field matches.

        Raises:
            AmbiguousFieldError: If the name matches multiple fields.
                Use a source-qualified name or get_by_id() to disambiguate.
        """
        return self._resolve_name(field_name, strict=True)

    def _resolve_name(self, field_name: str, *, strict: bool) -> FieldMetadata | None:
        """Resolve a field name to metadata.

        Args:
            field_name: Display name (case-insensitive) or source-qualified.
            strict: If True, raise AmbiguousFieldError on ambiguity.
                    If False, warn and return the first match.

        Returns:
            FieldMetadata or None if not found.
        """
        name_key = field_name.casefold()
        candidates = self._by_name.get(name_key, [])
        if not candidates:
            return None

        if len(candidates) == 1:
            return candidates[0]

        # Ambiguous name
        if strict:
            raise AmbiguousFieldError(field_name, candidates)

        # Warn once per ambiguous bare name at access time
        if name_key not in self._warned_names:
            self._warned_names.add(name_key)
            sources = ", ".join(f.enrichment_source or f.type or "unknown" for f in candidates)
            example_source = candidates[0].enrichment_source
            hint = (
                f"Disambiguate with 'source:name' syntax, e.g., '{example_source}:{field_name}'."
                if example_source
                else "Use get_by_id() for unambiguous access."
            )
            warnings.warn(
                f"Ambiguous field name '{field_name}' matches "
                f"{len(candidates)} fields "
                f"({sources}). Using first match. {hint}",
                UserWarning,
                stacklevel=3,
            )
        return candidates[0]

    def _resolve_value(self, value: Any, value_type: FieldValueType | None) -> Any:
        """Resolve a field value to human-readable text based on its type."""
        if value_type in (
            FieldValueType.DROPDOWN,
            FieldValueType.RANKED_DROPDOWN,
            FieldValueType.DROPDOWN_MULTI,
        ):
            return self._resolve_dropdown(value)

        if value_type == FieldValueType.PERSON:
            return resolve_person(value) or value
        if value_type == FieldValueType.PERSON_MULTI:
            if isinstance(value, list):
                return [resolve_person(v) or v for v in value]
            return value

        if value_type == FieldValueType.COMPANY:
            return resolve_company(value) or value
        if value_type == FieldValueType.COMPANY_MULTI:
            if isinstance(value, list):
                return [resolve_company(v) or v for v in value]
            return value

        if value_type == FieldValueType.LOCATION:
            return resolve_location(value) or value
        if value_type == FieldValueType.LOCATION_MULTI:
            if isinstance(value, list):
                return [resolve_location(v) or v for v in value]
            return value

        if value_type == FieldValueType.INTERACTION:
            return resolve_interaction(value) or value

        return value

    def _resolve_dropdown(self, value: Any) -> Any:
        """Resolve dropdown DropdownOption(s) to text."""
        from .models.entities import DropdownOption  # noqa: PLC0415

        if isinstance(value, list):
            return [self._resolve_dropdown(v) for v in value]
        if isinstance(value, DropdownOption):
            return value.text
        return value

    def get_by_id(
        self,
        entity: Company | Person | Opportunity | ListEntry | ListEntryWithEntity,
        field_id: str | AnyFieldId,
        *,
        resolve: ResolveMode = ResolveMode.RAW,
    ) -> Any:
        """
        Get a field value by ID (unambiguous, for duplicate field names).

        Args:
            entity: The entity
            field_id: The field ID (e.g., "field-123" or FieldId(123))
            resolve: ResolveMode.RAW or ResolveMode.TEXT

        Returns:
            The extracted field value, or None if not found.
        """
        field_id_str = str(field_id)
        value = _field_values(entity).get_value(field_id_str)
        if resolve == ResolveMode.TEXT and value is not None:
            field = self._by_id.get(field_id_str)
            value_type = field.value_type if field else None
            value = self._resolve_value(value, value_type)
        return value

    def field_names(self) -> list[str]:
        """Get all available field names (including source-qualified names for ambiguous fields)."""
        seen: set[str] = set()
        names: list[str] = []
        for key, fields in self._by_name.items():
            # Skip source-qualified keys — they'll be added when we detect ambiguity
            if ":" in key:
                continue
            if len(fields) == 1:
                name = fields[0].name
                if name not in seen:
                    seen.add(name)
                    names.append(name)
            else:
                # Ambiguous: emit source-qualified names
                for f in fields:
                    qualified = f"{f.enrichment_source}:{f.name}" if f.enrichment_source else f.name
                    if qualified not in seen:
                        seen.add(qualified)
                        names.append(qualified)
        return names

    def has_field(self, field_name: str) -> bool:
        """Check if a field exists by name.

        Supports source-qualified names like ``'dealroom:Description'``.
        """
        return field_name.casefold() in self._by_name

__init__(fields: Sequence[FieldMetadata]) -> None

Create a resolver from field metadata.

Parameters:

Name	Type	Description	Default
fields	Sequence[FieldMetadata]	Field metadata from client.companies.get_fields(), client.persons.get_fields(), etc.	required

Source code in affinity/field_resolver.py

def __init__(self, fields: Sequence[FieldMetadata]) -> None:
    """
    Create a resolver from field metadata.

    Args:
        fields: Field metadata from client.companies.get_fields(),
                client.persons.get_fields(), etc.
    """
    # Build case-insensitive name -> field(s) mapping.
    # Multiple fields can share the same display name (e.g., enrichment
    # fields from different sources both called "Description").
    self._by_name: dict[str, list[FieldMetadata]] = {}
    self._by_id: dict[str, FieldMetadata] = {}
    self._warned_names: set[str] = set()

    for field in fields:
        name_key = field.name.casefold()
        self._by_name.setdefault(name_key, []).append(field)

        # Register source-qualified key for enrichment fields (e.g., "dealroom:description")
        if field.enrichment_source:
            qualified_key = f"{field.enrichment_source}:{name_key}"
            self._by_name.setdefault(qualified_key, []).append(field)

        self._by_id[str(field.id)] = field

field_names() -> list[str]

Get all available field names (including source-qualified names for ambiguous fields).

Source code in affinity/field_resolver.py

def field_names(self) -> list[str]:
    """Get all available field names (including source-qualified names for ambiguous fields)."""
    seen: set[str] = set()
    names: list[str] = []
    for key, fields in self._by_name.items():
        # Skip source-qualified keys — they'll be added when we detect ambiguity
        if ":" in key:
            continue
        if len(fields) == 1:
            name = fields[0].name
            if name not in seen:
                seen.add(name)
                names.append(name)
        else:
            # Ambiguous: emit source-qualified names
            for f in fields:
                qualified = f"{f.enrichment_source}:{f.name}" if f.enrichment_source else f.name
                if qualified not in seen:
                    seen.add(qualified)
                    names.append(qualified)
    return names

find_field(field_name: str) -> FieldMetadata | None

Find field metadata by name (case-insensitive).

Useful for getting the FieldId needed by write operations like update_field_value(). Supports source-qualified names like 'dealroom:Description' for ambiguous fields.

Returns:

Type	Description
FieldMetadata | None	The FieldMetadata, or None if no field matches.

Raises:

Type	Description
AmbiguousFieldError	If the name matches multiple fields. Use a source-qualified name or get_by_id() to disambiguate.

Source code in affinity/field_resolver.py

def find_field(self, field_name: str) -> FieldMetadata | None:
    """
    Find field metadata by name (case-insensitive).

    Useful for getting the FieldId needed by write operations like
    ``update_field_value()``. Supports source-qualified names like
    ``'dealroom:Description'`` for ambiguous fields.

    Returns:
        The FieldMetadata, or None if no field matches.

    Raises:
        AmbiguousFieldError: If the name matches multiple fields.
            Use a source-qualified name or get_by_id() to disambiguate.
    """
    return self._resolve_name(field_name, strict=True)

get(entity: Company | Person | Opportunity | ListEntry | ListEntryWithEntity, field_name: str, *, resolve: ResolveMode = ResolveMode.RAW) -> Any

Get a field value by name from an entity.

Parameters:

Name	Type	Description	Default
entity	Company | Person | Opportunity | ListEntry | ListEntryWithEntity	The entity (Company, Person, Opportunity, or ListEntry)	required
field_name	str	The field display name (case-insensitive)	required
resolve	ResolveMode	ResolveMode.RAW returns extracted values as-is. ResolveMode.TEXT resolves all complex types to strings (dropdowns to text, persons to names, etc.).	RAW

Returns:

Type	Description
Any	The extracted field value, or None if field not found or not populated.

Example

status = resolver.get(company, "Status") print(status) # "Active"

Resolve complex types to text

stage = resolver.get(opportunity, "Stage", resolve=ResolveMode.TEXT) print(stage) # "Negotiation" instead of 12345

Source code in affinity/field_resolver.py

def get(
    self,
    entity: Company | Person | Opportunity | ListEntry | ListEntryWithEntity,
    field_name: str,
    *,
    resolve: ResolveMode = ResolveMode.RAW,
) -> Any:
    """
    Get a field value by name from an entity.

    Args:
        entity: The entity (Company, Person, Opportunity, or ListEntry)
        field_name: The field display name (case-insensitive)
        resolve: ResolveMode.RAW returns extracted values as-is.
                 ResolveMode.TEXT resolves all complex types to strings
                 (dropdowns to text, persons to names, etc.).

    Returns:
        The extracted field value, or None if field not found or not populated.

    Example:
        >>> status = resolver.get(company, "Status")
        >>> print(status)  # "Active"

        >>> # Resolve complex types to text
        >>> stage = resolver.get(opportunity, "Stage", resolve=ResolveMode.TEXT)
        >>> print(stage)  # "Negotiation" instead of 12345
    """
    fields = _field_values(entity)
    if not fields.requested:
        warnings.warn(
            "Fields were not requested for this entity. "
            "Pass field_ids or field_types when fetching to populate field values.",
            UserWarning,
            stacklevel=2,
        )
        return None

    field = self._resolve_name(field_name, strict=False)
    if field is None:
        return None

    value = fields.get_value(str(field.id))

    if resolve == ResolveMode.TEXT and value is not None:
        value = self._resolve_value(value, field.value_type)

    return value

get_by_id(entity: Company | Person | Opportunity | ListEntry | ListEntryWithEntity, field_id: str | AnyFieldId, *, resolve: ResolveMode = ResolveMode.RAW) -> Any

Get a field value by ID (unambiguous, for duplicate field names).

Parameters:

Name	Type	Description	Default
entity	Company | Person | Opportunity | ListEntry | ListEntryWithEntity	The entity	required
field_id	str | AnyFieldId	The field ID (e.g., "field-123" or FieldId(123))	required
resolve	ResolveMode	ResolveMode.RAW or ResolveMode.TEXT	RAW

Returns:

Type	Description
Any	The extracted field value, or None if not found.

Source code in affinity/field_resolver.py

def get_by_id(
    self,
    entity: Company | Person | Opportunity | ListEntry | ListEntryWithEntity,
    field_id: str | AnyFieldId,
    *,
    resolve: ResolveMode = ResolveMode.RAW,
) -> Any:
    """
    Get a field value by ID (unambiguous, for duplicate field names).

    Args:
        entity: The entity
        field_id: The field ID (e.g., "field-123" or FieldId(123))
        resolve: ResolveMode.RAW or ResolveMode.TEXT

    Returns:
        The extracted field value, or None if not found.
    """
    field_id_str = str(field_id)
    value = _field_values(entity).get_value(field_id_str)
    if resolve == ResolveMode.TEXT and value is not None:
        field = self._by_id.get(field_id_str)
        value_type = field.value_type if field else None
        value = self._resolve_value(value, value_type)
    return value

get_many(entity: Company | Person | Opportunity | ListEntry | ListEntryWithEntity, field_names: Sequence[str], *, resolve: ResolveMode = ResolveMode.RAW) -> dict[str, Any]

Get multiple field values by name.

Parameters:

Name	Type	Description	Default
entity	Company | Person | Opportunity | ListEntry | ListEntryWithEntity	The entity (Company, Person, Opportunity, or ListEntry)	required
field_names	Sequence[str]	List of field names to extract	required
resolve	ResolveMode	ResolveMode.RAW or ResolveMode.TEXT	RAW

Returns:

Type	Description
dict[str, Any]	Dict mapping field_name -> value (None for missing/unpopulated fields)

Example

values = resolver.get_many(company, ["Status", "Industry", "Size"]) print(values) {'Status': 'Active', 'Industry': 'Tech', 'Size': None}

Source code in affinity/field_resolver.py

def get_many(
    self,
    entity: Company | Person | Opportunity | ListEntry | ListEntryWithEntity,
    field_names: Sequence[str],
    *,
    resolve: ResolveMode = ResolveMode.RAW,
) -> dict[str, Any]:
    """
    Get multiple field values by name.

    Args:
        entity: The entity (Company, Person, Opportunity, or ListEntry)
        field_names: List of field names to extract
        resolve: ResolveMode.RAW or ResolveMode.TEXT

    Returns:
        Dict mapping field_name -> value (None for missing/unpopulated fields)

    Example:
        >>> values = resolver.get_many(company, ["Status", "Industry", "Size"])
        >>> print(values)
        {'Status': 'Active', 'Industry': 'Tech', 'Size': None}
    """
    return {name: self.get(entity, name, resolve=resolve) for name in field_names}

has_field(field_name: str) -> bool

Check if a field exists by name.

Supports source-qualified names like 'dealroom:Description'.

Source code in affinity/field_resolver.py

def has_field(self, field_name: str) -> bool:
    """Check if a field exists by name.

    Supports source-qualified names like ``'dealroom:Description'``.
    """
    return field_name.casefold() in self._by_name