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.

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

    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")
    list_size: int = Field(0, alias="listSize")

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

    @model_validator(mode="before")
    @classmethod
    def _coerce_v2_is_public(cls, value: Any) -> Any:
        # V2 list endpoints use `isPublic`; v1 uses `public`.
        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

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

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

Source code in affinity/models/secondary.py

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

    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 Exception:
            return data

        if entity_type == EntityType.PERSON:
            try:
                data["entity"] = PersonSummary.model_validate(entity)
            except Exception:
                return data
        elif entity_type == EntityType.ORGANIZATION:
            try:
                data["entity"] = CompanySummary.model_validate(entity)
            except Exception:
                return data
        elif entity_type == EntityType.OPPORTUNITY:
            try:
                data["entity"] = OpportunitySummary.model_validate(entity)
            except Exception:
                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 (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)

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

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

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

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.

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

    data: list[T] = Field(default_factory=list)
    pagination: PaginationInfo = Field(default_factory=PaginationInfo)

    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."""
        return self.pagination.next_cursor is not None

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

has_next: bool property

Whether there are more pages.

next_cursor: str | None property

Cursor for the next page, if any.

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

    # V1: only returned when with_interaction_dates=true; preserve shape for forward compatibility.
    interactions: dict[str, Any] | 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 (V1 API).

Source code in affinity/models/entities.py

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

    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