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

The list_size field was removed in v0.13.0 because the V2 API returns incorrect values (often 0 for non-empty lists). Use client.lists.get_size(list_id) to get accurate list size via V1 API.

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:
        The list_size field was removed in v0.13.0 because the V2 API returns
        incorrect values (often 0 for non-empty lists). Use
        ``client.lists.get_size(list_id)`` to get accurate list size via V1 API.
    """

    id: ListId
    name: str
    type: ListType
    is_public: bool = Field(alias="public")
    owner_id: UserId = Field(alias="ownerId")
    creator_id: UserId | None = Field(None, alias="creatorId")

    # 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]:
        # V2 list endpoints use `isPublic`; v1 uses `public`.
        if isinstance(data, Mapping):
            data = dict(data)
            if "public" not in data and "isPublic" in data:
                data["public"] = data.get("isPublic")
            # 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

    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

            # Guard against pagination loops (no cursor progress).
            if response.has_next and response.next_cursor == requested_url:
                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 response.has_next and not self._exhausted:
                    continue
                self._exhausted = True
                raise StopAsyncIteration

            if not response.has_next:
                self._exhausted = True

    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

        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

            # Guard against pagination loops
            if response.has_next and response.next_cursor == requested_url:
                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=False,  # Loop detected, no more pages
                            )
                        )
                    yield response
                break

            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=response.has_next,
                        )
                    )
                yield response

            if not response.has_next:
                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

    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

        # Guard against pagination loops
        if response.has_next and response.next_cursor == requested_url:
            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=False,  # Loop detected, no more pages
                        )
                    )
                yield response
            break

        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=response.has_next,
                    )
                )
            yield response

        if not response.has_next:
            break

BatchOperationResponse

Bases: AffinityModel

Response from batch field operations.

Source code in affinity/models/pagination.py

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

    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

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)
    is_global: bool = Field(False, alias="global")

    # 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)

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

DropdownOption

Bases: AffinityModel

A selectable option in a dropdown field.

Source code in affinity/models/entities.py

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

    id: DropdownOptionId
    text: str
    rank: int | None = None
    color: int | str | None = None

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")
    company_id: CompanyId | None = Field(None, 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-specific", "global", etc.

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

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

    @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

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

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

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)

    # Manual logging info
    manual_creator_id: UserId | None = Field(None, alias="manualCreatorId")
    updated_at: ISODatetime | None = Field(None, alias="updatedAt")

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

ListCreate

Bases: AffinityModel

Data for creating a new list (V1 API).

Source code in affinity/models/entities.py

class ListCreate(AffinityModel):
    """Data for creating a new list (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.

Contains the entity data and list-specific field values.

Source code in affinity/models/entities.py

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

    Contains the entity data and list-specific field values.
    """

    id: ListEntryId
    list_id: ListId = Field(alias="listId")
    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
    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:
        return _preserve_fields_raw(value)

    @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
    is_public: bool | None = Field(None, alias="public")
    owner_id: UserId | None = Field(None, alias="ownerId")
    list_size: int | None = Field(None, alias="listSize")

    @model_validator(mode="before")
    @classmethod
    def _coerce_v2_is_public(cls, value: Any) -> Any:
        if isinstance(value, Mapping) and "public" not in value and "isPublic" in value:
            data = dict(value)
            data["public"] = data.get("isPublic")
            return data
        return value

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

NoteCreate

Bases: AffinityModel

Data for creating a new note (V1 API).

Source code in affinity/models/secondary.py

class NoteCreate(AffinityModel):
    """Data for creating a new note (V1 API)."""

    content: str
    type: NoteType = NoteType.PLAIN_TEXT
    person_ids: list[PersonId] = Field(default_factory=list)
    company_ids: list[CompanyId] = Field(default_factory=list, alias="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

NoteUpdate

Bases: AffinityModel

Data for updating a note (V1 API).

Source code in affinity/models/secondary.py

class NoteUpdate(AffinityModel):
    """Data for updating a note (V1 API)."""

    content: str

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

    # 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")
    company_ids: list[CompanyId] = Field(default_factory=list, alias="organizationIds")

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

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

    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

            # Guard against pagination loops (no cursor progress).
            if response.has_next and response.next_cursor == requested_url:
                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 response.has_next and not self._exhausted:
                    continue
                self._exhausted = True
                raise StopIteration

            if not response.has_next:
                self._exhausted = True

    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

        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

            # Guard against pagination loops
            if response.has_next and response.next_cursor == requested_url:
                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=False,  # Loop detected, no more pages
                            )
                        )
                    yield response
                break

            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=response.has_next,
                        )
                    )
                yield response

            if not response.has_next:
                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

    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

        # Guard against pagination loops
        if response.has_next and response.next_cursor == requested_url:
            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=False,  # Loop detected, no more pages
                        )
                    )
                yield response
            break

        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=response.has_next,
                    )
                )
            yield response

        if not response.has_next:
            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)

    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.

__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 uses organizationIds)
    company_ids: list[CompanyId] = Field(default_factory=list, alias="organizationIds")
    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"
    )

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

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."""

    org_monthly: RateLimitInfo = Field(alias="orgMonthly")
    api_key_per_minute: RateLimitInfo = Field(alias="apiKeyPerMinute")

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 attached to an entity.

Source code in affinity/models/secondary.py

class Reminder(AffinityModel):
    """A reminder attached to an entity."""

    id: ReminderIdType
    type: ReminderType
    status: ReminderStatus
    content: str | None = None

    # Due date and recurrence
    due_date: ISODatetime = Field(alias="dueDate")
    reset_type: ReminderResetType | None = Field(None, alias="resetType")
    reminder_days: int | None = Field(None, alias="reminderDays")

    # Persons involved
    creator: PersonSummary | None = None
    owner: PersonSummary | None = None
    completer: PersonSummary | None = None

    # Associated entity (one of these)
    person: PersonSummary | None = None
    company: dict[str, Any] | None = Field(None, alias="organization")
    opportunity: dict[str, Any] | None = None

    # Timestamps
    created_at: ISODatetime = Field(alias="createdAt")
    completed_at: ISODatetime | None = Field(None, alias="completedAt")

ReminderCreate

Bases: AffinityModel

Data for creating a reminder (V1 API).

Source code in affinity/models/secondary.py

class ReminderCreate(AffinityModel):
    """Data for creating a reminder (V1 API)."""

    owner_id: UserId
    type: ReminderType
    content: str | None = None
    due_date: ISODatetime | None = None  # Required for one-time
    reset_type: ReminderResetType | None = None  # Required for recurring
    reminder_days: int | None = None  # Required for recurring

    # Associate with one entity
    person_id: PersonId | None = None
    company_id: CompanyId | None = Field(None, alias="organization_id")
    opportunity_id: OpportunityId | None = None

ReminderUpdate

Bases: AffinityModel

Data for updating a reminder (V1 API).

Source code in affinity/models/secondary.py

class ReminderUpdate(AffinityModel):
    """Data for updating a reminder (V1 API)."""

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

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

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

WebhookCreate

Bases: AffinityModel

Data for creating a webhook subscription (V1 API).

Source code in affinity/models/secondary.py

class WebhookCreate(AffinityModel):
    """Data for creating a webhook subscription (V1 API)."""

    webhook_url: str
    subscriptions: list[WebhookEvent] = Field(default_factory=list)

WebhookSubscription

Bases: AffinityModel

A webhook subscription for real-time events.

Source code in affinity/models/secondary.py

class WebhookSubscription(AffinityModel):
    """A webhook subscription for real-time events."""

    id: WebhookId
    webhook_url: str = Field(alias="webhookUrl")
    subscriptions: list[WebhookEvent] = Field(default_factory=list)
    disabled: bool = False
    created_by: UserId = Field(alias="createdBy")

WebhookUpdate

Bases: AffinityModel

Data for updating a webhook subscription (V1 API).

Source code in affinity/models/secondary.py

class WebhookUpdate(AffinityModel):
    """Data for updating a webhook subscription (V1 API)."""

    webhook_url: str | None = None
    subscriptions: list[WebhookEvent] | None = None
    disabled: bool | None = None

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 | 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
        """
        # Delegate to inner entity for ListEntryWithEntity
        if (
            hasattr(entity, "entity")
            and entity.entity is not None
            and hasattr(entity.entity, "fields")
            and not entity.fields.requested
            and entity.entity.fields.requested
        ):
            entity = entity.entity

        if not entity.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

        name_key = field_name.casefold()
        candidates = self._by_name.get(name_key, [])
        if not candidates:
            return None

        if len(candidates) == 1:
            field = candidates[0]
        else:
            # 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, "
                    f"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 {len(candidates)} fields "
                    f"({sources}). Using first match. {hint}",
                    UserWarning,
                    stacklevel=2,
                )
            field = candidates[0]

        value = entity.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 | 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 _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 | 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.
        """
        # Delegate to inner entity for ListEntryWithEntity
        if (
            hasattr(entity, "entity")
            and entity.entity is not None
            and hasattr(entity.entity, "fields")
            and not entity.fields.requested
            and entity.entity.fields.requested
        ):
            entity = entity.entity

        field_id_str = str(field_id)
        value = entity.fields.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

get(entity: Company | Person | Opportunity | 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 | 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 | 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
    """
    # Delegate to inner entity for ListEntryWithEntity
    if (
        hasattr(entity, "entity")
        and entity.entity is not None
        and hasattr(entity.entity, "fields")
        and not entity.fields.requested
        and entity.entity.fields.requested
    ):
        entity = entity.entity

    if not entity.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

    name_key = field_name.casefold()
    candidates = self._by_name.get(name_key, [])
    if not candidates:
        return None

    if len(candidates) == 1:
        field = candidates[0]
    else:
        # 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, "
                f"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 {len(candidates)} fields "
                f"({sources}). Using first match. {hint}",
                UserWarning,
                stacklevel=2,
            )
        field = candidates[0]

    value = entity.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 | 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 | 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 | 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.
    """
    # Delegate to inner entity for ListEntryWithEntity
    if (
        hasattr(entity, "entity")
        and entity.entity is not None
        and hasattr(entity.entity, "fields")
        and not entity.fields.requested
        and entity.entity.fields.requested
    ):
        entity = entity.entity

    field_id_str = str(field_id)
    value = entity.fields.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 | 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 | 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 | 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