Persons

Interaction Dates

The get() method supports fetching interaction date summaries for a person using the with_interaction_dates parameter. When enabled, the returned Person object will have its interaction_dates and interactions fields populated with:

• Last meeting date: When the last calendar event with this person occurred
• Next meeting date: When the next scheduled calendar event is
• Last email date: When the last email exchange happened
• Last interaction date: The most recent interaction of any type
• Also the first email, first meeting and last chat message dates

from affinity import Affinity
from affinity.types import PersonId

with Affinity(api_key="YOUR_API_KEY") as client:
    # Fetch person with interaction dates
    person = client.persons.get(
        PersonId(456),
        with_interaction_dates=True,
        with_interaction_persons=True,  # Include person IDs for each interaction
    )

    # Access interaction data
    if person.interaction_dates:
        print(f"Last meeting: {person.interaction_dates.last_event_date}")
        print(f"Next meeting: {person.interaction_dates.next_event_date}")
        print(f"Last email: {person.interaction_dates.last_email_date}")

    # Access team member IDs from interactions
    if person.interactions and person.interactions.last_event:
        team_ids = person.interactions.last_event.person_ids
        print(f"Last meeting attendees: {team_ids}")

Many persons

get() with interaction dates makes one V1 call per person, plus one V2 call when field_ids or field_types is given. For many persons, use the async client's batch_get(ids, with_interaction_dates=True, ...): with field filters it reads the fields in one call per 100 ids and the dates in one call per person, so about N + N/100 calls instead of 2N. With on_error="skip" a person that can't be read is left out of the result.

Finding records with Affinity's search

client.persons.find(...), iter_find(...) and find_body(...) filter and sort on Affinity's side (V2 search, beta) and return Person records. They work like the companies methods and need the "Export All Organizations directory" permission. search(term) is the separate name/email search.

from affinity.criteria import Condition

for person in client.persons.iter_find(Condition("field-123", "is-not-empty")):
    print(person.id, person.first_name, person.last_name)

Bases: PersonDiscoveryMixin, SingleFieldMixin, SearchMixin[Person]

Source code in affinity/services/persons.py

class PersonService(PersonDiscoveryMixin, SingleFieldMixin, SearchMixin[Person]):
    _entity_path = "/persons"

    _search_model = Person

    def _search_path(self) -> str:
        return "/persons/search"

    def _search_fields(self) -> builtins.list[FieldMetadata]:
        return self.get_fields(includes=SEARCH_INCLUDES)

    """
    Service for managing persons (contacts).

    Uses V2 API for efficient reading with field selection,
    V1 API for create/update/delete operations.
    """

    def __init__(self, client: HTTPClient):
        self._client = client

    # =========================================================================
    # Read Operations (V2 API)
    # =========================================================================

    def list(
        self,
        *,
        ids: Sequence[PersonId] | None = None,
        field_ids: Sequence[AnyFieldId] | None = None,
        field_types: Sequence[FieldType] | None = None,
        limit: int | None = None,
        cursor: str | None = None,
        **_unsupported: Any,
    ) -> PaginatedResponse[Person]:
        """
        Get a page of persons.

        Args:
            ids: Specific person IDs to fetch (batch lookup)
            field_ids: Specific field IDs to include in response
            field_types: Field types to include
            limit: Maximum number of results
            cursor: Cursor to resume pagination (opaque; obtained from prior responses)

        Returns:
            Paginated response with persons

        Note:
            The V2 /persons endpoint silently ignores any ``filter=`` parameter —
            passing one would return unfiltered results without warning. This method
            therefore rejects ``filter=`` with a ``ValueError``. To search persons
            by name or email use ``client.persons.search_pages(term)``. To filter
            by list-specific fields use ``client.lists.entries(list_id).list(filter=...)``.
        """
        if "filter" in _unsupported:
            raise ValueError(
                "V2 /persons does not support server-side filter. "
                "To search by name/email use client.persons.search_pages(term). "
                "To filter by list-specific fields use "
                "client.lists.entries(list_id).list(filter=...)."
            )
        if _unsupported:
            raise TypeError(f"Unexpected keyword arguments: {list(_unsupported)}")
        validate_entity_field_types(field_types, endpoint="person")
        if cursor is not None:
            if any(p is not None for p in (ids, field_ids, field_types, limit)):
                raise ValueError(
                    "Cannot combine 'cursor' with other parameters; cursor encodes all query "
                    "context. Start a new pagination sequence without a cursor to change "
                    "parameters."
                )
            data = self._client.get_url(cursor)
        else:
            check_positive_limit(limit)
            ids = materialize_ids(ids)
            if no_ids(ids):
                return PaginatedResponse[Person]()
            params: dict[str, Any] = {}
            if ids:
                params["ids"] = [int(id_) for id_ in ids]
            if field_ids:
                params["fieldIds"] = [str(field_id) for field_id in field_ids]
            if field_types:
                params["fieldTypes"] = [field_type.value for field_type in field_types]
            if limit is not None:
                params["limit"] = limit
            data = self._client.get("/persons", params=params or None)

        return PaginatedResponse[Person](
            data=[Person.model_validate(p) for p in data.get("data", [])],
            pagination=PaginationInfo.model_validate(data.get("pagination", {})),
        )

    def get_first(
        self,
        *,
        field_ids: Sequence[AnyFieldId] | None = None,
        field_types: Sequence[FieldType] | None = None,
        **_unsupported: Any,
    ) -> Person | None:
        """
        Get the first person, or None if no results.

        This is a convenience method equivalent to:
            page = client.persons.list(limit=1)
            return page.data[0] if page.data else None

        Args:
            field_ids: Specific field IDs to include
            field_types: Field types to include

        Returns:
            First Person, or None if no results.

        Note:
            The V2 /persons endpoint silently ignores any ``filter=`` parameter —
            passing one would return unfiltered results without warning. This method
            therefore rejects ``filter=`` with a ``ValueError``. To search persons
            by name or email use ``client.persons.search_pages(term)``. To filter
            by list-specific fields use ``client.lists.entries(list_id).list(filter=...)``.
        """
        if "filter" in _unsupported:
            raise ValueError(
                "V2 /persons does not support server-side filter. "
                "To search by name/email use client.persons.search_pages(term). "
                "To filter by list-specific fields use "
                "client.lists.entries(list_id).list(filter=...)."
            )
        if _unsupported:
            raise TypeError(f"Unexpected keyword arguments: {list(_unsupported)}")
        page = self.list(
            field_ids=field_ids,
            field_types=field_types,
            limit=1,
        )
        return page.data[0] if page.data else None

    def pages(
        self,
        *,
        ids: Sequence[PersonId] | None = None,
        field_ids: Sequence[AnyFieldId] | None = None,
        field_types: Sequence[FieldType] | None = None,
        limit: int | None = None,
        cursor: str | None = None,
        **_unsupported: Any,
    ) -> Iterator[PaginatedResponse[Person]]:
        """
        Iterate person pages (not items), yielding `PaginatedResponse[Person]`.

        Useful for ETL scripts that need checkpoint/resume via `page.next_cursor`.

        Args:
            ids: Specific person IDs to fetch (batch lookup)
            field_ids: Specific field IDs to include in response
            field_types: Field types to include
            limit: Maximum results per page
            cursor: Cursor to resume pagination

        Yields:
            PaginatedResponse[Person] for each page

        Note:
            The V2 /persons endpoint silently ignores any ``filter=`` parameter —
            passing one would return unfiltered results without warning. This method
            therefore rejects ``filter=`` with a ``ValueError``. To search persons
            by name or email use ``client.persons.search_pages(term)``. To filter
            by list-specific fields use ``client.lists.entries(list_id).list(filter=...)``.
        """
        if "filter" in _unsupported:
            raise ValueError(
                "V2 /persons does not support server-side filter. "
                "To search by name/email use client.persons.search_pages(term). "
                "To filter by list-specific fields use "
                "client.lists.entries(list_id).list(filter=...)."
            )
        if _unsupported:
            raise TypeError(f"Unexpected keyword arguments: {list(_unsupported)}")
        other_params = (ids, field_ids, field_types, limit)
        if cursor is not None and any(p is not None for p in other_params):
            raise ValueError(
                "Cannot combine 'cursor' with other parameters; cursor encodes all query context. "
                "Start a new pagination sequence without a cursor to change parameters."
            )
        requested_cursor = cursor
        guard = _CursorGuard(cursor)
        page = (
            self.list(cursor=cursor)
            if cursor is not None
            else self.list(ids=ids, field_ids=field_ids, field_types=field_types, limit=limit)
        )
        while True:
            yield page
            # Stop at the last page, or when paging can't progress (see _CursorGuard).
            if guard.last_page(requested_cursor, page):
                return
            requested_cursor = page.next_cursor
            page = self.list(cursor=requested_cursor)

    def all(
        self,
        *,
        ids: Sequence[PersonId] | None = None,
        field_ids: Sequence[AnyFieldId] | None = None,
        field_types: Sequence[FieldType] | None = None,
        **_unsupported: Any,
    ) -> Iterator[Person]:
        """
        Iterate through all persons with automatic pagination.

        Args:
            ids: Specific person IDs to fetch (batch lookup)
            field_ids: Specific field IDs to include
            field_types: Field types to include

        Yields:
            Person objects

        Note:
            The V2 /persons endpoint silently ignores any ``filter=`` parameter —
            passing one would return unfiltered results without warning. This method
            therefore rejects ``filter=`` with a ``ValueError``. To search persons
            by name or email use ``client.persons.search_pages(term)``. To filter
            by list-specific fields use ``client.lists.entries(list_id).list(filter=...)``.
        """
        if "filter" in _unsupported:
            raise ValueError(
                "V2 /persons does not support server-side filter. "
                "To search by name/email use client.persons.search_pages(term). "
                "To filter by list-specific fields use "
                "client.lists.entries(list_id).list(filter=...)."
            )
        if _unsupported:
            raise TypeError(f"Unexpected keyword arguments: {list(_unsupported)}")

        def fetch_page(next_url: str | None) -> PaginatedResponse[Person]:
            if next_url:
                data = self._client.get_url(next_url)
                return PaginatedResponse[Person](
                    data=[Person.model_validate(p) for p in data.get("data", [])],
                    pagination=PaginationInfo.model_validate(data.get("pagination", {})),
                )
            return self.list(
                ids=ids,
                field_ids=field_ids,
                field_types=field_types,
            )

        return PageIterator(fetch_page)

    def iter(
        self,
        *,
        ids: Sequence[PersonId] | None = None,
        field_ids: Sequence[AnyFieldId] | None = None,
        field_types: Sequence[FieldType] | None = None,
        **_unsupported: Any,
    ) -> Iterator[Person]:
        """
        Auto-paginate all persons.

        Alias for `all()` (FR-006 public contract).

        Note:
            The V2 /persons endpoint silently ignores any ``filter=`` parameter —
            passing one would return unfiltered results without warning. This method
            therefore rejects ``filter=`` with a ``ValueError``. To search persons
            by name or email use ``client.persons.search_pages(term)``. To filter
            by list-specific fields use ``client.lists.entries(list_id).list(filter=...)``.
        """
        if "filter" in _unsupported:
            raise ValueError(
                "V2 /persons does not support server-side filter. "
                "To search by name/email use client.persons.search_pages(term). "
                "To filter by list-specific fields use "
                "client.lists.entries(list_id).list(filter=...)."
            )
        if _unsupported:
            raise TypeError(f"Unexpected keyword arguments: {list(_unsupported)}")
        return self.all(ids=ids, field_ids=field_ids, field_types=field_types)

    def get(
        self,
        person_id: PersonId,
        *,
        field_ids: Sequence[AnyFieldId] | None = None,
        field_types: Sequence[FieldType] | None = None,
        include_field_values: bool = False,
        retries: int = 0,
        with_interaction_dates: bool = False,
        with_interaction_persons: bool = False,
    ) -> Person:
        """
        Get a single person by ID.

        Args:
            person_id: The person ID
            field_ids: Specific field IDs to include in response
            field_types: Field types to include (e.g., ["enriched", "global"])
            include_field_values: If True, fetch embedded field values. This saves
                one API call when you need both person info and field values.
                Cannot be combined with field_ids/field_types.
            retries: Number of retries on 404 NotFoundError. Default is 0 (fail fast).
                Set to 2-3 if calling immediately after create() to handle eventual
                consistency lag.
            with_interaction_dates: Include interaction date summaries (last/next
                meeting dates, email dates).
            with_interaction_persons: Include person IDs for each interaction.
                Only applies when with_interaction_dates=True.

        Returns:
            Person object with requested field data.
            When include_field_values=True, the Person will have a `field_values`
            attribute containing the list of FieldValue objects.
            When with_interaction_dates=True, the Person will have interaction_dates
            and interactions populated.

        Raises:
            NotFoundError: If person does not exist after all retries.
            ValueError: If include_field_values is combined with field_ids/field_types.

        Note:
            When combining with_interaction_dates with field_ids/field_types,
            two API calls are made internally and the results are merged.
        """
        return self._get_with_retry(
            person_id,
            field_ids=field_ids,
            field_types=field_types,
            include_field_values=include_field_values,
            retries=retries,
            with_interaction_dates=with_interaction_dates,
            with_interaction_persons=with_interaction_persons,
        )

    def _get_with_retry(
        self,
        person_id: PersonId,
        *,
        field_ids: Sequence[AnyFieldId] | None = None,
        field_types: Sequence[FieldType] | None = None,
        include_field_values: bool = False,
        retries: int = 0,
        with_interaction_dates: bool = False,
        with_interaction_persons: bool = False,
    ) -> Person:
        """Internal: get with retry logic."""
        last_error: NotFoundError | None = None
        attempts = retries + 1  # retries=0 means 1 attempt

        for attempt in range(attempts):
            try:
                return self._get_impl(
                    person_id,
                    field_ids=field_ids,
                    field_types=field_types,
                    include_field_values=include_field_values,
                    with_interaction_dates=with_interaction_dates,
                    with_interaction_persons=with_interaction_persons,
                )
            except NotFoundError as e:
                last_error = e
                if attempt < attempts - 1:  # Don't sleep after last attempt
                    time.sleep(0.5 * (attempt + 1))  # 0.5s, 1s, 1.5s backoff

        # V1 fallback: If V2 returned 404, try V1 API (handles V1→V2 sync delays)
        # Skip if already using V1 path (include_field_values or with_interaction_dates)
        if last_error is not None and not include_field_values and not with_interaction_dates:
            try:
                v1_data = self._client.get(f"/persons/{person_id}", v1=True)
                normalized = _normalize_v1_person_response(v1_data)
                return Person.model_validate(normalized)
            except NotFoundError:
                pass  # V1 also failed, raise original V2 error

        raise last_error  # type: ignore[misc]

    def _get_impl(
        self,
        person_id: PersonId,
        *,
        field_ids: Sequence[AnyFieldId] | None = None,
        field_types: Sequence[FieldType] | None = None,
        include_field_values: bool = False,
        with_interaction_dates: bool = False,
        with_interaction_persons: bool = False,
    ) -> Person:
        """Internal: actual get implementation."""
        has_field_filters = field_ids is not None or field_types is not None

        # include_field_values returns embedded field values, which is incompatible
        # with field_ids/field_types filtering (different data structures)
        if include_field_values and has_field_filters:
            raise ValueError(
                "Cannot combine 'include_field_values' with 'field_ids' or 'field_types'. "
                "Use include_field_values alone for embedded field values, or use "
                "field_ids/field_types alone for filtered fields."
            )

        # Path 1: include_field_values (V1 API, may include interaction dates)
        if include_field_values:
            v1_params: dict[str, Any] = {}
            if with_interaction_dates:
                v1_params["with_interaction_dates"] = True
            if with_interaction_persons:
                v1_params["with_interaction_persons"] = True

            data = self._client.get(
                f"/persons/{person_id}",
                params=v1_params or None,
                v1=True,
            )

            field_values_data = data.get("field_values", [])
            # V1 embedded field_values don't include entityId; add it before validation
            field_values = [
                FieldValue.model_validate({**fv, "entityId": int(person_id)})
                for fv in field_values_data
            ]
            normalized = _normalize_v1_person_response(data)
            person = Person.model_validate(normalized)
            object.__setattr__(person, "field_values", field_values)
            return person

        # Path 2: with_interaction_dates (may need merge if field filters present)
        if with_interaction_dates:
            v1_params = {"with_interaction_dates": True}
            if with_interaction_persons:
                v1_params["with_interaction_persons"] = True

            interaction_data = self._client.get(
                f"/persons/{person_id}",
                params=v1_params,
                v1=True,
            )

            # If field filtering is also requested, fetch filtered fields and merge
            if has_field_filters:
                v2_params: dict[str, Any] = {}
                if field_ids:
                    v2_params["fieldIds"] = [str(fid) for fid in field_ids]
                if field_types:
                    v2_params["fieldTypes"] = [ft.value for ft in field_types]

                filtered_data = self._client.get(
                    f"/persons/{person_id}",
                    params=v2_params,
                )

                # Merge: filtered fields + interaction data
                filtered_data["interaction_dates"] = interaction_data.get("interaction_dates")
                filtered_data["interactions"] = interaction_data.get("interactions")
                return Person.model_validate(filtered_data)

            # No field filtering, normalize and return V1 data
            normalized = _normalize_v1_person_response(interaction_data)
            return Person.model_validate(normalized)

        # Path 3: Standard path (supports field filtering)
        params: dict[str, Any] = {}
        if field_ids:
            params["fieldIds"] = [str(field_id) for field_id in field_ids]
        if field_types:
            params["fieldTypes"] = [field_type.value for field_type in field_types]

        data = self._client.get(
            f"/persons/{person_id}",
            params=params or None,
        )
        return Person.model_validate(data)

    def get_list_entries(
        self,
        person_id: PersonId,
        *,
        limit: int | None = None,
        cursor: str | None = None,
    ) -> PaginatedResponse[ListEntry]:
        """One page of a person's list entries across all lists.

        Each entry's ``entity_id`` is ``person_id`` and its ``entity_type`` is
        ``EntityType.PERSON``; ``entity`` is ``None`` (this endpoint doesn't embed the person).
        Follow ``pagination.next_cursor`` with ``cursor=`` (from this person's entries) for
        the next page.
        """
        path = f"/persons/{person_id}/list-entries"
        if cursor is not None:
            if limit is not None:
                raise ValueError(
                    "Cannot combine 'cursor' with other parameters; cursor encodes all query "
                    "context. Start a new pagination sequence without a cursor to change "
                    "parameters."
                )
            check_list_entries_cursor(cursor, path)
            data = self._client.get_url(cursor)
        else:
            check_positive_limit(limit)
            data = self._client.get(path, params={"limit": limit} if limit is not None else None)

        return entity_list_entries_page(data, entity_id=person_id, entity_type=EntityType.PERSON)

    def get_lists(
        self,
        person_id: PersonId,
    ) -> PaginatedResponse[ListSummary]:
        """Get all lists that contain this person."""
        data = self._client.get(f"/persons/{person_id}/lists")

        return PaginatedResponse[ListSummary](
            data=[ListSummary.model_validate(item) for item in data.get("data", [])],
            pagination=PaginationInfo.model_validate(data.get("pagination", {})),
        )

    def get_fields(
        self,
        *,
        filter: str | None = None,
        includes: Sequence[str] | None = None,
    ) -> builtins.list[FieldMetadata]:
        """
        Get metadata about person fields.

        Cached for performance.

        Args:
            filter: Filter on the field name: ``name="Location"`` (exact) or ``name=~Loc``
                (substring); both case-sensitive. Combine clauses with ``|``. Only ``name``
                is filterable; anything else is rejected with 400.
            includes: Extra properties per field: ``"filterability"``, ``"sortability"``
                (read them via ``is_filterable`` / ``is_sortable``).
        """
        params, cache_key = build_fields_query(
            "person_fields:",
            filter=filter,
            includes=includes,
        )

        data = self._client.get_all_pages(
            "/persons/fields",
            params=params or None,
            cache_key=cache_key,
            cache_ttl=300,
        )

        return [FieldMetadata.model_validate(f) for f in data.get("data", [])]

    def get_field_dropdown_options(
        self, field_id: AnyFieldId, *, with_status_types: bool = False
    ) -> builtins.list[DropdownOption]:
        """All options of a person dropdown, ranked-dropdown or status field (every page).

        Options with ``type``, ``rank``, ``color`` and (status fields) ``status_category`` /
        ``win_rate``. Options of global person fields can only be read; list-specific fields'
        options are managed through ``client.lists``. ``with_status_types=True``: see
        ``lists.get_field_dropdown_options``.
        """
        data = self._client.get_all_pages(
            options_path(field_id, list_id=None, entity="persons"),
            params={"limit": 100},
            min_api_version=DROPDOWN_OPTION_WRITES_MIN_API_VERSION if with_status_types else None,
        )
        return parse_options(data)

    def get_field_values(
        self,
        person_id: PersonId,
        *,
        ids: Sequence[AnyFieldId | str] | None = None,
    ) -> FieldValues:
        """
        Field values of one person, keyed by field id (``GET /v2/persons/{id}/fields``).

        Args:
            person_id: The person.
            ids: Only these fields (server-side filter); all fields when omitted.
        """
        params = {"ids": [str(fid) for fid in ids]} if ids else None
        data = self._client.get_all_pages(f"/persons/{person_id}/fields", params=params)
        return FieldValues.model_validate(data.get("data", []))

    def batch_update_fields(
        self,
        person_id: PersonId,
        updates: Mapping[AnyFieldId | str, Any],
        *,
        value_types: Mapping[AnyFieldId | str, FieldValueType | str] | None = None,
    ) -> BatchOperationResponse:
        """
        Update several field values of one person in one request.

        Affinity applies the whole batch or none of it; each update replaces the field's value
        and ``None`` (with a type) clears it. Global, enriched and Source of Introduction
        fields are all writable. At most 100 updates, and 100 values per multi-value field.

        Needs Affinity API version {FIELD_WRITES_MIN_API_VERSION} (out of beta there): an
        unpinned client sends that version for this call; a client pinned to an older version
        raises ``ApiVersionTooOldError`` without sending anything.

        Args:
            person_id: The person.
            updates: Field id -> new value. Values are auto-typed (str→text, int/float→number,
                datetime/date→datetime) unless the field has an entry in ``value_types``.
            value_types: Field id -> value type (e.g. ``"dropdown-multi"``); needed for lists
                (multi-value fields) and ``None`` (clear).
        """
        items = _batch_update_items(updates, value_types)
        result = self._client.patch(
            f"/persons/{person_id}/fields",
            json={"operation": "update-fields", "updates": items},
            min_api_version=FIELD_WRITES_MIN_API_VERSION,
        )
        return BatchOperationResponse.model_validate(result)

    # =========================================================================
    # Associations (V1 API)
    # =========================================================================

    def get_associated_company_ids(
        self,
        person_id: PersonId,
        *,
        max_results: int | None = None,
    ) -> builtins.list[CompanyId]:
        """
        Get associated company IDs for a person.

        V1-only: V2 does not expose person -> company associations directly.
        Uses GET `/persons/{id}` (V1) and returns `organization_ids`.

        Args:
            person_id: The person ID
            max_results: Maximum number of company IDs to return

        Returns:
            List of CompanyId values associated with this person

        Note:
            The Person model already has `company_ids` populated from V1's
            `organizationIds` field. This method provides API parity with
            `CompanyService.get_associated_person_ids()`.
        """
        data = self._client.get(f"/persons/{person_id}", v1=True)
        # Defensive: handle potential {"person": {...}} wrapper
        # (consistent with CompanyService.get_associated_person_ids pattern)
        person = data.get("person") if isinstance(data, dict) else None
        source = person if isinstance(person, dict) else data
        org_ids = None
        if isinstance(source, dict):
            org_ids = source.get("organization_ids") or source.get("organizationIds")

        if not isinstance(org_ids, list):
            return []

        ids = [CompanyId(int(cid)) for cid in org_ids if cid is not None]
        if max_results is not None and max_results >= 0:
            return ids[:max_results]
        return ids

    def get_associated_company_ids_batch(
        self,
        person_ids: Sequence[PersonId],
        *,
        on_error: Literal["raise", "skip"] = "raise",
    ) -> dict[PersonId, builtins.list[CompanyId]]:
        """
        Get company associations for multiple persons.

        Makes one V1 API call per person.

        Args:
            person_ids: Sequence of person IDs to fetch
            on_error: How to handle errors - "raise" (default) or "skip" failed IDs

        Returns:
            Dict mapping person_id -> list of company_ids

        Raises:
            AffinityError: If on_error="raise" and any fetch fails.

        Example:
            associations = client.persons.get_associated_company_ids_batch(person_ids)
            all_company_ids = set()
            for company_ids in associations.values():
                all_company_ids.update(company_ids)
        """
        result: dict[PersonId, builtins.list[CompanyId]] = {}
        for person_id in person_ids:
            try:
                result[person_id] = self.get_associated_company_ids(person_id)
            except AffinityError:
                if on_error == "raise":
                    raise
                # skip: continue without this person
            except Exception as e:
                if on_error == "raise":
                    raise AffinityError(
                        f"Failed to get associations for person {person_id}: {e}"
                    ) from e
                # skip: continue without this person
        return result

    def get_associated_opportunity_ids(
        self,
        person_id: PersonId,
        *,
        max_results: int | None = None,
    ) -> builtins.list[OpportunityId]:
        """
        Get associated opportunity IDs for a person.

        V1-only: V2 does not expose person -> opportunity associations directly.
        Uses GET `/persons/{id}` (V1) and returns `opportunity_ids`.

        Args:
            person_id: The person ID
            max_results: Maximum number of opportunity IDs to return

        Returns:
            List of OpportunityId values associated with this person

        Note:
            The Person model already has `opportunity_ids` populated from V1's
            `opportunityIds` field. This method provides API parity with
            `OpportunityService.get_associated_person_ids()`.
        """
        data = self._client.get(f"/persons/{person_id}", v1=True)
        # Defensive: handle potential {"person": {...}} wrapper
        person = data.get("person") if isinstance(data, dict) else None
        source = person if isinstance(person, dict) else data
        opp_ids = None
        if isinstance(source, dict):
            opp_ids = source.get("opportunity_ids") or source.get("opportunityIds")

        if not isinstance(opp_ids, list):
            return []

        ids = [OpportunityId(int(oid)) for oid in opp_ids if oid is not None]
        if max_results is not None and max_results >= 0:
            return ids[:max_results]
        return ids

    def get_associated_opportunity_ids_batch(
        self,
        person_ids: Sequence[PersonId],
        *,
        on_error: Literal["raise", "skip"] = "raise",
    ) -> dict[PersonId, builtins.list[OpportunityId]]:
        """
        Get opportunity associations for multiple persons.

        Makes one V1 API call per person.

        Args:
            person_ids: Sequence of person IDs to fetch
            on_error: How to handle errors - "raise" (default) or "skip" failed IDs

        Returns:
            Dict mapping person_id -> list of opportunity_ids

        Raises:
            AffinityError: If on_error="raise" and any fetch fails.
        """
        result: dict[PersonId, builtins.list[OpportunityId]] = {}
        for person_id in person_ids:
            try:
                result[person_id] = self.get_associated_opportunity_ids(person_id)
            except AffinityError:
                if on_error == "raise":
                    raise
                # skip: continue without this person
            except Exception as e:
                if on_error == "raise":
                    raise AffinityError(
                        f"Failed to get associations for person {person_id}: {e}"
                    ) from e
                # skip: continue without this person
        return result

    # =========================================================================
    # Search (V1 API)
    # =========================================================================

    def search(
        self,
        term: str,
        *,
        with_interaction_dates: bool = False,
        with_interaction_persons: bool = False,
        with_opportunities: bool = False,
        page_size: int | None = None,
        page_token: str | None = None,
    ) -> PaginatedResponse[Person]:
        """
        Search for persons by name or email.

        Uses V1 API for search functionality.

        Args:
            term: Search term (name or email)
            with_interaction_dates: Include interaction date data
            with_interaction_persons: Include persons for interactions
            with_opportunities: Include associated opportunity IDs
            page_size: Results per page (max 500)
            page_token: Pagination token

        Returns:
            PaginatedResponse[Person] with matching persons and pagination info
        """
        params: dict[str, Any] = {"term": term}
        if with_interaction_dates:
            params["with_interaction_dates"] = True
        if with_interaction_persons:
            params["with_interaction_persons"] = True
        if with_opportunities:
            params["with_opportunities"] = True
        if page_size:
            params["page_size"] = page_size
        if page_token:
            params["page_token"] = page_token

        data = self._client.get("/persons", params=params, v1=True)
        items = [Person.model_validate(p) for p in data.get("persons", [])]
        return PaginatedResponse[Person](
            data=items,
            next_page_token=data.get("next_page_token"),
        )

    def search_pages(
        self,
        term: str,
        *,
        with_interaction_dates: bool = False,
        with_interaction_persons: bool = False,
        with_opportunities: bool = False,
        page_size: int | None = None,
        page_token: str | None = None,
    ) -> Iterator[PaginatedResponse[Person]]:
        """
        Iterate V1 person-search result pages.

        Useful for scripts that need checkpoint/resume via `next_page_token`.

        Args:
            term: Search term (name or email)
            with_interaction_dates: Include interaction date data
            with_interaction_persons: Include persons for interactions
            with_opportunities: Include associated opportunity IDs
            page_size: Results per page (max 500)
            page_token: Resume from this pagination token

        Yields:
            PaginatedResponse[Person] for each page
        """
        requested_token = page_token
        page = self.search(
            term,
            with_interaction_dates=with_interaction_dates,
            with_interaction_persons=with_interaction_persons,
            with_opportunities=with_opportunities,
            page_size=page_size,
            page_token=page_token,
        )
        while True:
            yield page
            next_token = page.next_page_token
            if not next_token or next_token == requested_token:
                return
            requested_token = next_token
            page = self.search(
                term,
                with_interaction_dates=with_interaction_dates,
                with_interaction_persons=with_interaction_persons,
                with_opportunities=with_opportunities,
                page_size=page_size,
                page_token=next_token,
            )

    def search_all(
        self,
        term: str,
        *,
        with_interaction_dates: bool = False,
        with_interaction_persons: bool = False,
        with_opportunities: bool = False,
        page_size: int | None = None,
        page_token: str | None = None,
    ) -> Iterator[Person]:
        """
        Iterate all V1 person-search results with automatic pagination.

        Args:
            term: Search term (name or email)
            with_interaction_dates: Include interaction date data
            with_interaction_persons: Include persons for interactions
            with_opportunities: Include associated opportunity IDs
            page_size: Results per page (max 500)
            page_token: Resume from this pagination token

        Yields:
            Person objects matching the search term
        """
        for page in self.search_pages(
            term,
            with_interaction_dates=with_interaction_dates,
            with_interaction_persons=with_interaction_persons,
            with_opportunities=with_opportunities,
            page_size=page_size,
            page_token=page_token,
        ):
            yield from page.data

    def resolve(
        self,
        *,
        email: str | None = None,
        name: str | None = None,
    ) -> Person | None:
        """
        Find a single person by email or name.

        This is a convenience helper that searches and returns the first exact match,
        or None if not found. Uses V1 search internally.

        Args:
            email: Email address to search for
            name: Person name to search for (first + last)

        Returns:
            The matching Person, or None if not found

        Raises:
            ValueError: If neither email nor name is provided

        Note:
            This auto-paginates V1 search results until a match is found.
            If multiple matches are found, returns the first one. For full
            disambiguation, use resolve_all() or search() directly.
        """
        if not email and not name:
            raise ValueError("Must provide either email or name")

        term = email or name or ""
        for page in self.search_pages(term, page_size=10):
            for person in page.data:
                if _person_matches(person, email=email, name=name):
                    return person

        return None

    def resolve_all(
        self,
        *,
        email: str | None = None,
        name: str | None = None,
    ) -> builtins.list[Person]:
        """
        Find all persons matching an email or name.

        Notes:
        - This auto-paginates V1 search results to collect exact matches.
        - Unlike resolve(), this returns every match in server-provided order.
        """
        if not email and not name:
            raise ValueError("Must provide either email or name")

        term = email or name or ""
        matches: builtins.list[Person] = []
        for page in self.search_pages(term, page_size=10):
            for person in page.data:
                if _person_matches(person, email=email, name=name):
                    matches.append(person)
        return matches

    # =========================================================================
    # Write Operations (V1 API)
    # =========================================================================

    _DEDUP_PAGE_CAP = 3  # 3 pages x 500 = 1500 fuzzy results scanned per search term

    def create(self, data: PersonCreate, *, if_not_exists: bool = True) -> Person:
        """
        Create a new person.

        Args:
            data: Person creation data
            if_not_exists: If True (default), check for an existing person with
                the same primary email (case-insensitive exact) — falling back
                to exact first+last name match when no email is provided.
                Uses V1 search_pages(term) with page_size=500, capped at
                _DEDUP_PAGE_CAP pages. Set to False to skip the check and
                create unconditionally.

        Returns:
            Created person

        Raises:
            DuplicateEntityError: When if_not_exists=True and an exact-email
                OR exact full-name match already exists.

        Note:
            Creates use V1 API, while reads use V2 API. Due to eventual consistency
            between V1 and V2, a `get()` call immediately after `create()` may return
            404 NotFoundError. If you need to read immediately after creation, either:
            - Use the Person object returned by this method (it contains the created data)
            - Add a short delay (100-500ms) before calling get()
            - Implement retry logic in your application
        """
        if if_not_exists:
            existing = self._find_exact_duplicate(
                first_name=data.first_name,
                last_name=data.last_name,
                emails=list(data.emails or []),
            )
            if existing is not None:
                raise DuplicateEntityError(
                    f"Person with email or name already exists (id={existing.id})",
                    entity_type="person",
                    existing_id=int(existing.id),
                    existing_name=f"{existing.first_name or ''} {existing.last_name or ''}".strip(),
                )

        payload = data.model_dump(by_alias=True, mode="json")
        if not data.company_ids:
            payload.pop("organization_ids", None)

        result = self._client.post("/persons", json=payload, v1=True)

        if self._client.cache:
            self._client.cache.invalidate_prefix("person")

        return Person.model_validate(result)

    def _find_exact_duplicate(
        self,
        *,
        first_name: str | None,
        last_name: str | None,
        emails: builtins.list[str],
    ) -> Person | None:
        """Search V1 for an exact-email or exact full-name match. Returns None if no match.

        Email takes priority — exact case-insensitive match against any entry in the
        person's emails list or primary_email. Falls back to exact first+last name match
        when no emails are provided in the create payload.

        Iterates at most `_DEDUP_PAGE_CAP` pages per search term (1500 fuzzy results).
        """
        emails_lower = {e.strip().lower() for e in emails if e}
        fn_lower = (first_name or "").strip().lower()
        ln_lower = (last_name or "").strip().lower()

        search_terms = list(emails) if emails else [f"{first_name or ''} {last_name or ''}".strip()]
        seen_ids: set[int] = set()

        for term in search_terms:
            if not term:
                continue
            for page_num, page in enumerate(self.search_pages(term, page_size=500)):
                for person in page.data:
                    pid = int(person.id)
                    if pid in seen_ids:
                        continue
                    seen_ids.add(pid)

                    person_emails = {(e or "").strip().lower() for e in (person.emails or [])}
                    if person.primary_email:
                        person_emails.add(person.primary_email.strip().lower())

                    if emails_lower and emails_lower & person_emails:
                        return person

                    if (
                        not emails_lower
                        and fn_lower
                        and ln_lower
                        and (person.first_name or "").lower() == fn_lower
                        and (person.last_name or "").lower() == ln_lower
                    ):
                        return person

                if page_num + 1 >= self._DEDUP_PAGE_CAP:
                    break
                if not page.next_page_token:
                    break

        return None

    def update(
        self,
        person_id: PersonId,
        data: PersonUpdate,
    ) -> Person:
        """
        Update an existing person.

        Note: To add emails/organizations, include existing values plus new ones.
        """
        payload = data.model_dump(
            by_alias=True,
            mode="json",
            exclude_unset=True,
            exclude_none=True,
        )

        result = self._client.put(
            f"/persons/{person_id}",
            json=payload,
            v1=True,
        )

        if self._client.cache:
            self._client.cache.invalidate_prefix("person")

        return Person.model_validate(result)

    def delete(self, person_id: PersonId) -> bool:
        """Delete a person (also deletes associated field values)."""
        result = self._client.delete(f"/persons/{person_id}", v1=True)
        invalidate_all_list_sizes(self._client)

        if self._client.cache:
            self._client.cache.invalidate_prefix("person")

        return bool(result.get("success", True))

    # =========================================================================
    # Merge Operations (V2 BETA)
    # =========================================================================

    def merge(
        self,
        primary_id: PersonId,
        duplicate_id: PersonId,
    ) -> str:
        """
        Merge a duplicate person into a primary person.

        Returns a task URL to check merge status.
        """
        result = self._client.post(
            "/person-merges",
            json={
                "primaryPersonId": int(primary_id),
                "duplicatePersonId": int(duplicate_id),
            },
        )
        return merge_task_url(result, "person")

    def get_merge_status(self, task_id: str) -> MergeTask:
        """Check the status of a merge operation."""
        data = self._client.get(f"/tasks/person-merges/{task_id}")
        return MergeTask.model_validate(data)

    # =========================================================================
    # Merge history and relationships (V2)
    # =========================================================================

    def list_merges(
        self,
        *,
        status: str | None = None,
        task_id: str | None = None,
        limit: int | None = None,
        cursor: str | None = None,
    ) -> PaginatedResponse[PersonMergeState]:
        """One page of past person merges, newest first.

        ``status`` is ``"in-progress"``, ``"success"`` or ``"failed"``; ``task_id`` (or the task
        URL a merge returned) limits it to one merge task. Needs the "Manage duplicates"
        permission and the organization admin role (otherwise ``AuthorizationError``).
        """
        check_cursor_alone(cursor, status, task_id, limit)
        if cursor is not None:
            return page_of(PersonMergeState, self._client.get_url(cursor))
        params = merge_params(status=status, task_id=task_id, limit=limit)
        return page_of(PersonMergeState, self._client.get("/person-merges", params=params or None))

    def iter_merges(
        self, *, status: str | None = None, task_id: str | None = None
    ) -> Iterator[PersonMergeState]:
        """All past person merges matching the filters (every page)."""

        def fetch_page(next_url: str | None) -> PaginatedResponse[PersonMergeState]:
            if next_url:
                return self.list_merges(cursor=next_url)
            return self.list_merges(status=status, task_id=task_id)

        return PageIterator(fetch_page)

    def get_merge_state(self, merge_id: int) -> PersonMergeState:
        """One past person merge (see :meth:`list_merges`)."""
        return PersonMergeState.model_validate(self._client.get(f"/person-merges/{int(merge_id)}"))

    def list_relationships(
        self,
        person_id: PersonId,
        *,
        min_score: float | None = None,
        order: Literal["asc", "desc"] = "desc",
        limit: int | None = None,
        total_count: bool = False,
        cursor: str | None = None,
    ) -> PaginatedResponse[Relationship]:
        """One page of the relationships between your team and this person's people.

        Strongest first (``order="asc"`` for weakest first). ``min_score`` (0.0-1.0) keeps
        relationships at least that strong; relationships known only from LinkedIn have score 0,
        so any ``min_score`` above 0 drops them. ``limit`` is 1-100 (API default 20). Needs
        Affinity API version {RELATIONSHIPS_MIN_API_VERSION} (sent for this call and every page).
        """
        if cursor is not None:
            if min_score is not None or limit is not None or total_count or order != "desc":
                raise ValueError(
                    "Cannot combine 'cursor' with other parameters; cursor encodes all query "
                    "context."
                )
            return page_of(
                Relationship,
                self._client.get_url(cursor, min_api_version=RELATIONSHIPS_MIN_API_VERSION),
            )
        params = relationship_params(
            min_score=min_score, order=order, limit=limit, total_count=total_count
        )
        return page_of(
            Relationship,
            self._client.get(
                f"/persons/{int(person_id)}/relationships",
                params=params or None,
                min_api_version=RELATIONSHIPS_MIN_API_VERSION,
            ),
        )

    def iter_relationships(
        self,
        person_id: PersonId,
        *,
        min_score: float | None = None,
        order: Literal["asc", "desc"] = "desc",
        limit: int | None = None,
    ) -> Iterator[Relationship]:
        """All relationships for this person (every page); see :meth:`list_relationships`."""

        def fetch_page(next_url: str | None) -> PaginatedResponse[Relationship]:
            if next_url:
                return self.list_relationships(person_id, cursor=next_url)
            return self.list_relationships(person_id, min_score=min_score, order=order, limit=limit)

        return PageIterator(fetch_page)

    # =========================================================================
    # Notes (V2)
    # =========================================================================

    def list_notes(
        self,
        person_id: PersonId,
        *,
        creator_id: int | None = None,
        created_after: datetime | None = None,
        created_before: datetime | None = None,
        updated_after: datetime | None = None,
        limit: int | None = None,
        total_count: bool = False,
        cursor: str | None = None,
    ) -> PaginatedResponse[NoteV2]:
        """One page of this person's notes as V2 notes (``NoteV2``; replies not included).

        Filters: ``creator_id``, ``created_after`` (inclusive) / ``created_before`` (exclusive),
        ``updated_after``, rounded outward to whole seconds. ``limit`` is 1-100.
        """
        check_cursor_alone(
            cursor, creator_id, created_after, created_before, updated_after, limit, total_count
        )
        if cursor is not None:
            return page_of(NoteV2, self._client.get_url(cursor, min_api_version=None))
        params = note_params(
            creator_id=creator_id,
            created_after=created_after,
            created_before=created_before,
            updated_after=updated_after,
            limit=limit,
            total_count=total_count,
        )
        return page_of(
            NoteV2,
            self._client.get(
                f"/persons/{int(person_id)}/notes",
                params=params or None,
                min_api_version=None,
            ),
        )

    def iter_notes(
        self,
        person_id: PersonId,
        *,
        creator_id: int | None = None,
        created_after: datetime | None = None,
        created_before: datetime | None = None,
        updated_after: datetime | None = None,
        limit: int | None = None,
    ) -> Iterator[NoteV2]:
        """All of this person's notes (every page); see :meth:`list_notes`."""

        def fetch_page(next_url: str | None) -> PaginatedResponse[NoteV2]:
            if next_url:
                return self.list_notes(person_id, cursor=next_url)
            return self.list_notes(
                person_id,
                creator_id=creator_id,
                created_after=created_after,
                created_before=created_before,
                updated_after=updated_after,
                limit=limit,
            )

        return PageIterator(fetch_page)

all(*, ids: Sequence[PersonId] | None = None, field_ids: Sequence[AnyFieldId] | None = None, field_types: Sequence[FieldType] | None = None, **_unsupported: Any) -> Iterator[Person]

Iterate through all persons with automatic pagination.

Parameters:

Name	Type	Description	Default
ids	Sequence[PersonId] | None	Specific person IDs to fetch (batch lookup)	None
field_ids	Sequence[AnyFieldId] | None	Specific field IDs to include	None
field_types	Sequence[FieldType] | None	Field types to include	None

Yields:

Type	Description
Person	Person objects

Note

The V2 /persons endpoint silently ignores any filter= parameter — passing one would return unfiltered results without warning. This method therefore rejects filter= with a ValueError. To search persons by name or email use client.persons.search_pages(term). To filter by list-specific fields use client.lists.entries(list_id).list(filter=...).

Source code in affinity/services/persons.py

def all(
    self,
    *,
    ids: Sequence[PersonId] | None = None,
    field_ids: Sequence[AnyFieldId] | None = None,
    field_types: Sequence[FieldType] | None = None,
    **_unsupported: Any,
) -> Iterator[Person]:
    """
    Iterate through all persons with automatic pagination.

    Args:
        ids: Specific person IDs to fetch (batch lookup)
        field_ids: Specific field IDs to include
        field_types: Field types to include

    Yields:
        Person objects

    Note:
        The V2 /persons endpoint silently ignores any ``filter=`` parameter —
        passing one would return unfiltered results without warning. This method
        therefore rejects ``filter=`` with a ``ValueError``. To search persons
        by name or email use ``client.persons.search_pages(term)``. To filter
        by list-specific fields use ``client.lists.entries(list_id).list(filter=...)``.
    """
    if "filter" in _unsupported:
        raise ValueError(
            "V2 /persons does not support server-side filter. "
            "To search by name/email use client.persons.search_pages(term). "
            "To filter by list-specific fields use "
            "client.lists.entries(list_id).list(filter=...)."
        )
    if _unsupported:
        raise TypeError(f"Unexpected keyword arguments: {list(_unsupported)}")

    def fetch_page(next_url: str | None) -> PaginatedResponse[Person]:
        if next_url:
            data = self._client.get_url(next_url)
            return PaginatedResponse[Person](
                data=[Person.model_validate(p) for p in data.get("data", [])],
                pagination=PaginationInfo.model_validate(data.get("pagination", {})),
            )
        return self.list(
            ids=ids,
            field_ids=field_ids,
            field_types=field_types,
        )

    return PageIterator(fetch_page)

batch_update_fields(person_id: PersonId, updates: Mapping[AnyFieldId | str, Any], *, value_types: Mapping[AnyFieldId | str, FieldValueType | str] | None = None) -> BatchOperationResponse

Update several field values of one person in one request.

Affinity applies the whole batch or none of it; each update replaces the field's value and None (with a type) clears it. Global, enriched and Source of Introduction fields are all writable. At most 100 updates, and 100 values per multi-value field.

Needs Affinity API version {FIELD_WRITES_MIN_API_VERSION} (out of beta there): an unpinned client sends that version for this call; a client pinned to an older version raises ApiVersionTooOldError without sending anything.

Parameters:

Name	Type	Description	Default
person_id	PersonId	The person.	required
updates	Mapping[AnyFieldId | str, Any]	Field id -> new value. Values are auto-typed (str→text, int/float→number, datetime/date→datetime) unless the field has an entry in value_types.	required
value_types	Mapping[AnyFieldId | str, FieldValueType | str] | None	Field id -> value type (e.g. "dropdown-multi"); needed for lists (multi-value fields) and None (clear).	None

Source code in affinity/services/persons.py

def batch_update_fields(
    self,
    person_id: PersonId,
    updates: Mapping[AnyFieldId | str, Any],
    *,
    value_types: Mapping[AnyFieldId | str, FieldValueType | str] | None = None,
) -> BatchOperationResponse:
    """
    Update several field values of one person in one request.

    Affinity applies the whole batch or none of it; each update replaces the field's value
    and ``None`` (with a type) clears it. Global, enriched and Source of Introduction
    fields are all writable. At most 100 updates, and 100 values per multi-value field.

    Needs Affinity API version {FIELD_WRITES_MIN_API_VERSION} (out of beta there): an
    unpinned client sends that version for this call; a client pinned to an older version
    raises ``ApiVersionTooOldError`` without sending anything.

    Args:
        person_id: The person.
        updates: Field id -> new value. Values are auto-typed (str→text, int/float→number,
            datetime/date→datetime) unless the field has an entry in ``value_types``.
        value_types: Field id -> value type (e.g. ``"dropdown-multi"``); needed for lists
            (multi-value fields) and ``None`` (clear).
    """
    items = _batch_update_items(updates, value_types)
    result = self._client.patch(
        f"/persons/{person_id}/fields",
        json={"operation": "update-fields", "updates": items},
        min_api_version=FIELD_WRITES_MIN_API_VERSION,
    )
    return BatchOperationResponse.model_validate(result)

create(data: PersonCreate, *, if_not_exists: bool = True) -> Person

Create a new person.

Parameters:

Name	Type	Description	Default
data	PersonCreate	Person creation data	required
if_not_exists	bool	If True (default), check for an existing person with the same primary email (case-insensitive exact) — falling back to exact first+last name match when no email is provided. Uses V1 search_pages(term) with page_size=500, capped at _DEDUP_PAGE_CAP pages. Set to False to skip the check and create unconditionally.	True

Returns:

Type	Description
Person	Created person

Raises:

Type	Description
DuplicateEntityError	When if_not_exists=True and an exact-email OR exact full-name match already exists.

Note

Creates use V1 API, while reads use V2 API. Due to eventual consistency between V1 and V2, a get() call immediately after create() may return 404 NotFoundError. If you need to read immediately after creation, either: - Use the Person object returned by this method (it contains the created data) - Add a short delay (100-500ms) before calling get() - Implement retry logic in your application

Source code in affinity/services/persons.py

def create(self, data: PersonCreate, *, if_not_exists: bool = True) -> Person:
    """
    Create a new person.

    Args:
        data: Person creation data
        if_not_exists: If True (default), check for an existing person with
            the same primary email (case-insensitive exact) — falling back
            to exact first+last name match when no email is provided.
            Uses V1 search_pages(term) with page_size=500, capped at
            _DEDUP_PAGE_CAP pages. Set to False to skip the check and
            create unconditionally.

    Returns:
        Created person

    Raises:
        DuplicateEntityError: When if_not_exists=True and an exact-email
            OR exact full-name match already exists.

    Note:
        Creates use V1 API, while reads use V2 API. Due to eventual consistency
        between V1 and V2, a `get()` call immediately after `create()` may return
        404 NotFoundError. If you need to read immediately after creation, either:
        - Use the Person object returned by this method (it contains the created data)
        - Add a short delay (100-500ms) before calling get()
        - Implement retry logic in your application
    """
    if if_not_exists:
        existing = self._find_exact_duplicate(
            first_name=data.first_name,
            last_name=data.last_name,
            emails=list(data.emails or []),
        )
        if existing is not None:
            raise DuplicateEntityError(
                f"Person with email or name already exists (id={existing.id})",
                entity_type="person",
                existing_id=int(existing.id),
                existing_name=f"{existing.first_name or ''} {existing.last_name or ''}".strip(),
            )

    payload = data.model_dump(by_alias=True, mode="json")
    if not data.company_ids:
        payload.pop("organization_ids", None)

    result = self._client.post("/persons", json=payload, v1=True)

    if self._client.cache:
        self._client.cache.invalidate_prefix("person")

    return Person.model_validate(result)

delete(person_id: PersonId) -> bool

Delete a person (also deletes associated field values).

Source code in affinity/services/persons.py

def delete(self, person_id: PersonId) -> bool:
    """Delete a person (also deletes associated field values)."""
    result = self._client.delete(f"/persons/{person_id}", v1=True)
    invalidate_all_list_sizes(self._client)

    if self._client.cache:
        self._client.cache.invalidate_prefix("person")

    return bool(result.get("success", True))

get(person_id: PersonId, *, field_ids: Sequence[AnyFieldId] | None = None, field_types: Sequence[FieldType] | None = None, include_field_values: bool = False, retries: int = 0, with_interaction_dates: bool = False, with_interaction_persons: bool = False) -> Person

Get a single person by ID.

Parameters:

Name	Type	Description	Default
person_id	PersonId	The person ID	required
field_ids	Sequence[AnyFieldId] | None	Specific field IDs to include in response	None
field_types	Sequence[FieldType] | None	Field types to include (e.g., ["enriched", "global"])	None
include_field_values	bool	If True, fetch embedded field values. This saves one API call when you need both person info and field values. Cannot be combined with field_ids/field_types.	False
retries	int	Number of retries on 404 NotFoundError. Default is 0 (fail fast). Set to 2-3 if calling immediately after create() to handle eventual consistency lag.	0
with_interaction_dates	bool	Include interaction date summaries (last/next meeting dates, email dates).	False
with_interaction_persons	bool	Include person IDs for each interaction. Only applies when with_interaction_dates=True.	False

Returns:

Type	Description
Person	Person object with requested field data.
Person	When include_field_values=True, the Person will have a field_values
Person	attribute containing the list of FieldValue objects.
Person	When with_interaction_dates=True, the Person will have interaction_dates
Person	and interactions populated.

Raises:

Type	Description
NotFoundError	If person does not exist after all retries.
ValueError	If include_field_values is combined with field_ids/field_types.

Note

When combining with_interaction_dates with field_ids/field_types, two API calls are made internally and the results are merged.

Source code in affinity/services/persons.py

def get(
    self,
    person_id: PersonId,
    *,
    field_ids: Sequence[AnyFieldId] | None = None,
    field_types: Sequence[FieldType] | None = None,
    include_field_values: bool = False,
    retries: int = 0,
    with_interaction_dates: bool = False,
    with_interaction_persons: bool = False,
) -> Person:
    """
    Get a single person by ID.

    Args:
        person_id: The person ID
        field_ids: Specific field IDs to include in response
        field_types: Field types to include (e.g., ["enriched", "global"])
        include_field_values: If True, fetch embedded field values. This saves
            one API call when you need both person info and field values.
            Cannot be combined with field_ids/field_types.
        retries: Number of retries on 404 NotFoundError. Default is 0 (fail fast).
            Set to 2-3 if calling immediately after create() to handle eventual
            consistency lag.
        with_interaction_dates: Include interaction date summaries (last/next
            meeting dates, email dates).
        with_interaction_persons: Include person IDs for each interaction.
            Only applies when with_interaction_dates=True.

    Returns:
        Person object with requested field data.
        When include_field_values=True, the Person will have a `field_values`
        attribute containing the list of FieldValue objects.
        When with_interaction_dates=True, the Person will have interaction_dates
        and interactions populated.

    Raises:
        NotFoundError: If person does not exist after all retries.
        ValueError: If include_field_values is combined with field_ids/field_types.

    Note:
        When combining with_interaction_dates with field_ids/field_types,
        two API calls are made internally and the results are merged.
    """
    return self._get_with_retry(
        person_id,
        field_ids=field_ids,
        field_types=field_types,
        include_field_values=include_field_values,
        retries=retries,
        with_interaction_dates=with_interaction_dates,
        with_interaction_persons=with_interaction_persons,
    )

get_associated_company_ids(person_id: PersonId, *, max_results: int | None = None) -> builtins.list[CompanyId]

Get associated company IDs for a person.

V1-only: V2 does not expose person -> company associations directly. Uses GET /persons/{id} (V1) and returns organization_ids.

Parameters:

Name	Type	Description	Default
person_id	PersonId	The person ID	required
max_results	int | None	Maximum number of company IDs to return	None

Returns:

Type	Description
list[CompanyId]	List of CompanyId values associated with this person

Note

The Person model already has company_ids populated from V1's organizationIds field. This method provides API parity with CompanyService.get_associated_person_ids().

Source code in affinity/services/persons.py

def get_associated_company_ids(
    self,
    person_id: PersonId,
    *,
    max_results: int | None = None,
) -> builtins.list[CompanyId]:
    """
    Get associated company IDs for a person.

    V1-only: V2 does not expose person -> company associations directly.
    Uses GET `/persons/{id}` (V1) and returns `organization_ids`.

    Args:
        person_id: The person ID
        max_results: Maximum number of company IDs to return

    Returns:
        List of CompanyId values associated with this person

    Note:
        The Person model already has `company_ids` populated from V1's
        `organizationIds` field. This method provides API parity with
        `CompanyService.get_associated_person_ids()`.
    """
    data = self._client.get(f"/persons/{person_id}", v1=True)
    # Defensive: handle potential {"person": {...}} wrapper
    # (consistent with CompanyService.get_associated_person_ids pattern)
    person = data.get("person") if isinstance(data, dict) else None
    source = person if isinstance(person, dict) else data
    org_ids = None
    if isinstance(source, dict):
        org_ids = source.get("organization_ids") or source.get("organizationIds")

    if not isinstance(org_ids, list):
        return []

    ids = [CompanyId(int(cid)) for cid in org_ids if cid is not None]
    if max_results is not None and max_results >= 0:
        return ids[:max_results]
    return ids

get_associated_company_ids_batch(person_ids: Sequence[PersonId], *, on_error: Literal['raise', 'skip'] = 'raise') -> dict[PersonId, builtins.list[CompanyId]]

Get company associations for multiple persons.

Makes one V1 API call per person.

Parameters:

Name	Type	Description	Default
person_ids	Sequence[PersonId]	Sequence of person IDs to fetch	required
on_error	Literal['raise', 'skip']	How to handle errors - "raise" (default) or "skip" failed IDs	'raise'

Returns:

Type	Description
dict[PersonId, list[CompanyId]]	Dict mapping person_id -> list of company_ids

Raises:

Type	Description
AffinityError	If on_error="raise" and any fetch fails.

Example

associations = client.persons.get_associated_company_ids_batch(person_ids) all_company_ids = set() for company_ids in associations.values(): all_company_ids.update(company_ids)

Source code in affinity/services/persons.py

def get_associated_company_ids_batch(
    self,
    person_ids: Sequence[PersonId],
    *,
    on_error: Literal["raise", "skip"] = "raise",
) -> dict[PersonId, builtins.list[CompanyId]]:
    """
    Get company associations for multiple persons.

    Makes one V1 API call per person.

    Args:
        person_ids: Sequence of person IDs to fetch
        on_error: How to handle errors - "raise" (default) or "skip" failed IDs

    Returns:
        Dict mapping person_id -> list of company_ids

    Raises:
        AffinityError: If on_error="raise" and any fetch fails.

    Example:
        associations = client.persons.get_associated_company_ids_batch(person_ids)
        all_company_ids = set()
        for company_ids in associations.values():
            all_company_ids.update(company_ids)
    """
    result: dict[PersonId, builtins.list[CompanyId]] = {}
    for person_id in person_ids:
        try:
            result[person_id] = self.get_associated_company_ids(person_id)
        except AffinityError:
            if on_error == "raise":
                raise
            # skip: continue without this person
        except Exception as e:
            if on_error == "raise":
                raise AffinityError(
                    f"Failed to get associations for person {person_id}: {e}"
                ) from e
            # skip: continue without this person
    return result

get_associated_opportunity_ids(person_id: PersonId, *, max_results: int | None = None) -> builtins.list[OpportunityId]

Get associated opportunity IDs for a person.

V1-only: V2 does not expose person -> opportunity associations directly. Uses GET /persons/{id} (V1) and returns opportunity_ids.

Parameters:

Name	Type	Description	Default
person_id	PersonId	The person ID	required
max_results	int | None	Maximum number of opportunity IDs to return	None

Returns:

Type	Description
list[OpportunityId]	List of OpportunityId values associated with this person

Note

The Person model already has opportunity_ids populated from V1's opportunityIds field. This method provides API parity with OpportunityService.get_associated_person_ids().

Source code in affinity/services/persons.py

def get_associated_opportunity_ids(
    self,
    person_id: PersonId,
    *,
    max_results: int | None = None,
) -> builtins.list[OpportunityId]:
    """
    Get associated opportunity IDs for a person.

    V1-only: V2 does not expose person -> opportunity associations directly.
    Uses GET `/persons/{id}` (V1) and returns `opportunity_ids`.

    Args:
        person_id: The person ID
        max_results: Maximum number of opportunity IDs to return

    Returns:
        List of OpportunityId values associated with this person

    Note:
        The Person model already has `opportunity_ids` populated from V1's
        `opportunityIds` field. This method provides API parity with
        `OpportunityService.get_associated_person_ids()`.
    """
    data = self._client.get(f"/persons/{person_id}", v1=True)
    # Defensive: handle potential {"person": {...}} wrapper
    person = data.get("person") if isinstance(data, dict) else None
    source = person if isinstance(person, dict) else data
    opp_ids = None
    if isinstance(source, dict):
        opp_ids = source.get("opportunity_ids") or source.get("opportunityIds")

    if not isinstance(opp_ids, list):
        return []

    ids = [OpportunityId(int(oid)) for oid in opp_ids if oid is not None]
    if max_results is not None and max_results >= 0:
        return ids[:max_results]
    return ids

get_associated_opportunity_ids_batch(person_ids: Sequence[PersonId], *, on_error: Literal['raise', 'skip'] = 'raise') -> dict[PersonId, builtins.list[OpportunityId]]

Get opportunity associations for multiple persons.

Makes one V1 API call per person.

Parameters:

Name	Type	Description	Default
person_ids	Sequence[PersonId]	Sequence of person IDs to fetch	required
on_error	Literal['raise', 'skip']	How to handle errors - "raise" (default) or "skip" failed IDs	'raise'

Returns:

Type	Description
dict[PersonId, list[OpportunityId]]	Dict mapping person_id -> list of opportunity_ids

Raises:

Type	Description
AffinityError	If on_error="raise" and any fetch fails.

Source code in affinity/services/persons.py

def get_associated_opportunity_ids_batch(
    self,
    person_ids: Sequence[PersonId],
    *,
    on_error: Literal["raise", "skip"] = "raise",
) -> dict[PersonId, builtins.list[OpportunityId]]:
    """
    Get opportunity associations for multiple persons.

    Makes one V1 API call per person.

    Args:
        person_ids: Sequence of person IDs to fetch
        on_error: How to handle errors - "raise" (default) or "skip" failed IDs

    Returns:
        Dict mapping person_id -> list of opportunity_ids

    Raises:
        AffinityError: If on_error="raise" and any fetch fails.
    """
    result: dict[PersonId, builtins.list[OpportunityId]] = {}
    for person_id in person_ids:
        try:
            result[person_id] = self.get_associated_opportunity_ids(person_id)
        except AffinityError:
            if on_error == "raise":
                raise
            # skip: continue without this person
        except Exception as e:
            if on_error == "raise":
                raise AffinityError(
                    f"Failed to get associations for person {person_id}: {e}"
                ) from e
            # skip: continue without this person
    return result

get_field_dropdown_options(field_id: AnyFieldId, *, with_status_types: bool = False) -> builtins.list[DropdownOption]

All options of a person dropdown, ranked-dropdown or status field (every page).

Options with type, rank, color and (status fields) status_category / win_rate. Options of global person fields can only be read; list-specific fields' options are managed through client.lists. with_status_types=True: see lists.get_field_dropdown_options.

Source code in affinity/services/persons.py

def get_field_dropdown_options(
    self, field_id: AnyFieldId, *, with_status_types: bool = False
) -> builtins.list[DropdownOption]:
    """All options of a person dropdown, ranked-dropdown or status field (every page).

    Options with ``type``, ``rank``, ``color`` and (status fields) ``status_category`` /
    ``win_rate``. Options of global person fields can only be read; list-specific fields'
    options are managed through ``client.lists``. ``with_status_types=True``: see
    ``lists.get_field_dropdown_options``.
    """
    data = self._client.get_all_pages(
        options_path(field_id, list_id=None, entity="persons"),
        params={"limit": 100},
        min_api_version=DROPDOWN_OPTION_WRITES_MIN_API_VERSION if with_status_types else None,
    )
    return parse_options(data)

get_field_values(person_id: PersonId, *, ids: Sequence[AnyFieldId | str] | None = None) -> FieldValues

Field values of one person, keyed by field id (GET /v2/persons/{id}/fields).

Parameters:

Name	Type	Description	Default
person_id	PersonId	The person.	required
ids	Sequence[AnyFieldId | str] | None	Only these fields (server-side filter); all fields when omitted.	None

Source code in affinity/services/persons.py

def get_field_values(
    self,
    person_id: PersonId,
    *,
    ids: Sequence[AnyFieldId | str] | None = None,
) -> FieldValues:
    """
    Field values of one person, keyed by field id (``GET /v2/persons/{id}/fields``).

    Args:
        person_id: The person.
        ids: Only these fields (server-side filter); all fields when omitted.
    """
    params = {"ids": [str(fid) for fid in ids]} if ids else None
    data = self._client.get_all_pages(f"/persons/{person_id}/fields", params=params)
    return FieldValues.model_validate(data.get("data", []))

get_fields(*, filter: str | None = None, includes: Sequence[str] | None = None) -> builtins.list[FieldMetadata]

Get metadata about person fields.

Cached for performance.

Parameters:

Name	Type	Description	Default
filter	str | None	Filter on the field name: name="Location" (exact) or name=~Loc (substring); both case-sensitive. Combine clauses with |. Only name is filterable; anything else is rejected with 400.	None
includes	Sequence[str] | None	Extra properties per field: "filterability", "sortability" (read them via is_filterable / is_sortable).	None

Source code in affinity/services/persons.py

def get_fields(
    self,
    *,
    filter: str | None = None,
    includes: Sequence[str] | None = None,
) -> builtins.list[FieldMetadata]:
    """
    Get metadata about person fields.

    Cached for performance.

    Args:
        filter: Filter on the field name: ``name="Location"`` (exact) or ``name=~Loc``
            (substring); both case-sensitive. Combine clauses with ``|``. Only ``name``
            is filterable; anything else is rejected with 400.
        includes: Extra properties per field: ``"filterability"``, ``"sortability"``
            (read them via ``is_filterable`` / ``is_sortable``).
    """
    params, cache_key = build_fields_query(
        "person_fields:",
        filter=filter,
        includes=includes,
    )

    data = self._client.get_all_pages(
        "/persons/fields",
        params=params or None,
        cache_key=cache_key,
        cache_ttl=300,
    )

    return [FieldMetadata.model_validate(f) for f in data.get("data", [])]

get_first(*, field_ids: Sequence[AnyFieldId] | None = None, field_types: Sequence[FieldType] | None = None, **_unsupported: Any) -> Person | None

Get the first person, or None if no results.

This is a convenience method equivalent to

page = client.persons.list(limit=1) return page.data[0] if page.data else None

Parameters:

Name	Type	Description	Default
field_ids	Sequence[AnyFieldId] | None	Specific field IDs to include	None
field_types	Sequence[FieldType] | None	Field types to include	None

Returns:

Type	Description
Person | None	First Person, or None if no results.

Note

The V2 /persons endpoint silently ignores any filter= parameter — passing one would return unfiltered results without warning. This method therefore rejects filter= with a ValueError. To search persons by name or email use client.persons.search_pages(term). To filter by list-specific fields use client.lists.entries(list_id).list(filter=...).

Source code in affinity/services/persons.py

def get_first(
    self,
    *,
    field_ids: Sequence[AnyFieldId] | None = None,
    field_types: Sequence[FieldType] | None = None,
    **_unsupported: Any,
) -> Person | None:
    """
    Get the first person, or None if no results.

    This is a convenience method equivalent to:
        page = client.persons.list(limit=1)
        return page.data[0] if page.data else None

    Args:
        field_ids: Specific field IDs to include
        field_types: Field types to include

    Returns:
        First Person, or None if no results.

    Note:
        The V2 /persons endpoint silently ignores any ``filter=`` parameter —
        passing one would return unfiltered results without warning. This method
        therefore rejects ``filter=`` with a ``ValueError``. To search persons
        by name or email use ``client.persons.search_pages(term)``. To filter
        by list-specific fields use ``client.lists.entries(list_id).list(filter=...)``.
    """
    if "filter" in _unsupported:
        raise ValueError(
            "V2 /persons does not support server-side filter. "
            "To search by name/email use client.persons.search_pages(term). "
            "To filter by list-specific fields use "
            "client.lists.entries(list_id).list(filter=...)."
        )
    if _unsupported:
        raise TypeError(f"Unexpected keyword arguments: {list(_unsupported)}")
    page = self.list(
        field_ids=field_ids,
        field_types=field_types,
        limit=1,
    )
    return page.data[0] if page.data else None

get_list_entries(person_id: PersonId, *, limit: int | None = None, cursor: str | None = None) -> PaginatedResponse[ListEntry]

One page of a person's list entries across all lists.

Each entry's entity_id is person_id and its entity_type is EntityType.PERSON; entity is None (this endpoint doesn't embed the person). Follow pagination.next_cursor with cursor= (from this person's entries) for the next page.

Source code in affinity/services/persons.py

def get_list_entries(
    self,
    person_id: PersonId,
    *,
    limit: int | None = None,
    cursor: str | None = None,
) -> PaginatedResponse[ListEntry]:
    """One page of a person's list entries across all lists.

    Each entry's ``entity_id`` is ``person_id`` and its ``entity_type`` is
    ``EntityType.PERSON``; ``entity`` is ``None`` (this endpoint doesn't embed the person).
    Follow ``pagination.next_cursor`` with ``cursor=`` (from this person's entries) for
    the next page.
    """
    path = f"/persons/{person_id}/list-entries"
    if cursor is not None:
        if limit is not None:
            raise ValueError(
                "Cannot combine 'cursor' with other parameters; cursor encodes all query "
                "context. Start a new pagination sequence without a cursor to change "
                "parameters."
            )
        check_list_entries_cursor(cursor, path)
        data = self._client.get_url(cursor)
    else:
        check_positive_limit(limit)
        data = self._client.get(path, params={"limit": limit} if limit is not None else None)

    return entity_list_entries_page(data, entity_id=person_id, entity_type=EntityType.PERSON)

get_lists(person_id: PersonId) -> PaginatedResponse[ListSummary]

Get all lists that contain this person.

Source code in affinity/services/persons.py

def get_lists(
    self,
    person_id: PersonId,
) -> PaginatedResponse[ListSummary]:
    """Get all lists that contain this person."""
    data = self._client.get(f"/persons/{person_id}/lists")

    return PaginatedResponse[ListSummary](
        data=[ListSummary.model_validate(item) for item in data.get("data", [])],
        pagination=PaginationInfo.model_validate(data.get("pagination", {})),
    )

get_merge_state(merge_id: int) -> PersonMergeState

One past person merge (see :meth:list_merges).

Source code in affinity/services/persons.py

def get_merge_state(self, merge_id: int) -> PersonMergeState:
    """One past person merge (see :meth:`list_merges`)."""
    return PersonMergeState.model_validate(self._client.get(f"/person-merges/{int(merge_id)}"))

get_merge_status(task_id: str) -> MergeTask

Check the status of a merge operation.

Source code in affinity/services/persons.py

def get_merge_status(self, task_id: str) -> MergeTask:
    """Check the status of a merge operation."""
    data = self._client.get(f"/tasks/person-merges/{task_id}")
    return MergeTask.model_validate(data)

iter(*, ids: Sequence[PersonId] | None = None, field_ids: Sequence[AnyFieldId] | None = None, field_types: Sequence[FieldType] | None = None, **_unsupported: Any) -> Iterator[Person]

Auto-paginate all persons.

Alias for all() (FR-006 public contract).

Note

The V2 /persons endpoint silently ignores any filter= parameter — passing one would return unfiltered results without warning. This method therefore rejects filter= with a ValueError. To search persons by name or email use client.persons.search_pages(term). To filter by list-specific fields use client.lists.entries(list_id).list(filter=...).

Source code in affinity/services/persons.py

def iter(
    self,
    *,
    ids: Sequence[PersonId] | None = None,
    field_ids: Sequence[AnyFieldId] | None = None,
    field_types: Sequence[FieldType] | None = None,
    **_unsupported: Any,
) -> Iterator[Person]:
    """
    Auto-paginate all persons.

    Alias for `all()` (FR-006 public contract).

    Note:
        The V2 /persons endpoint silently ignores any ``filter=`` parameter —
        passing one would return unfiltered results without warning. This method
        therefore rejects ``filter=`` with a ``ValueError``. To search persons
        by name or email use ``client.persons.search_pages(term)``. To filter
        by list-specific fields use ``client.lists.entries(list_id).list(filter=...)``.
    """
    if "filter" in _unsupported:
        raise ValueError(
            "V2 /persons does not support server-side filter. "
            "To search by name/email use client.persons.search_pages(term). "
            "To filter by list-specific fields use "
            "client.lists.entries(list_id).list(filter=...)."
        )
    if _unsupported:
        raise TypeError(f"Unexpected keyword arguments: {list(_unsupported)}")
    return self.all(ids=ids, field_ids=field_ids, field_types=field_types)

iter_merges(*, status: str | None = None, task_id: str | None = None) -> Iterator[PersonMergeState]

All past person merges matching the filters (every page).

Source code in affinity/services/persons.py

def iter_merges(
    self, *, status: str | None = None, task_id: str | None = None
) -> Iterator[PersonMergeState]:
    """All past person merges matching the filters (every page)."""

    def fetch_page(next_url: str | None) -> PaginatedResponse[PersonMergeState]:
        if next_url:
            return self.list_merges(cursor=next_url)
        return self.list_merges(status=status, task_id=task_id)

    return PageIterator(fetch_page)

iter_notes(person_id: PersonId, *, creator_id: int | None = None, created_after: datetime | None = None, created_before: datetime | None = None, updated_after: datetime | None = None, limit: int | None = None) -> Iterator[NoteV2]

All of this person's notes (every page); see :meth:list_notes.

Source code in affinity/services/persons.py

def iter_notes(
    self,
    person_id: PersonId,
    *,
    creator_id: int | None = None,
    created_after: datetime | None = None,
    created_before: datetime | None = None,
    updated_after: datetime | None = None,
    limit: int | None = None,
) -> Iterator[NoteV2]:
    """All of this person's notes (every page); see :meth:`list_notes`."""

    def fetch_page(next_url: str | None) -> PaginatedResponse[NoteV2]:
        if next_url:
            return self.list_notes(person_id, cursor=next_url)
        return self.list_notes(
            person_id,
            creator_id=creator_id,
            created_after=created_after,
            created_before=created_before,
            updated_after=updated_after,
            limit=limit,
        )

    return PageIterator(fetch_page)

iter_relationships(person_id: PersonId, *, min_score: float | None = None, order: Literal['asc', 'desc'] = 'desc', limit: int | None = None) -> Iterator[Relationship]

All relationships for this person (every page); see :meth:list_relationships.

Source code in affinity/services/persons.py

def iter_relationships(
    self,
    person_id: PersonId,
    *,
    min_score: float | None = None,
    order: Literal["asc", "desc"] = "desc",
    limit: int | None = None,
) -> Iterator[Relationship]:
    """All relationships for this person (every page); see :meth:`list_relationships`."""

    def fetch_page(next_url: str | None) -> PaginatedResponse[Relationship]:
        if next_url:
            return self.list_relationships(person_id, cursor=next_url)
        return self.list_relationships(person_id, min_score=min_score, order=order, limit=limit)

    return PageIterator(fetch_page)

list(*, ids: Sequence[PersonId] | None = None, field_ids: Sequence[AnyFieldId] | None = None, field_types: Sequence[FieldType] | None = None, limit: int | None = None, cursor: str | None = None, **_unsupported: Any) -> PaginatedResponse[Person]

Get a page of persons.

Parameters:

Name	Type	Description	Default
ids	Sequence[PersonId] | None	Specific person IDs to fetch (batch lookup)	None
field_ids	Sequence[AnyFieldId] | None	Specific field IDs to include in response	None
field_types	Sequence[FieldType] | None	Field types to include	None
limit	int | None	Maximum number of results	None
cursor	str | None	Cursor to resume pagination (opaque; obtained from prior responses)	None

Returns:

Type	Description
PaginatedResponse[Person]	Paginated response with persons

Note

The V2 /persons endpoint silently ignores any filter= parameter — passing one would return unfiltered results without warning. This method therefore rejects filter= with a ValueError. To search persons by name or email use client.persons.search_pages(term). To filter by list-specific fields use client.lists.entries(list_id).list(filter=...).

Source code in affinity/services/persons.py

def list(
    self,
    *,
    ids: Sequence[PersonId] | None = None,
    field_ids: Sequence[AnyFieldId] | None = None,
    field_types: Sequence[FieldType] | None = None,
    limit: int | None = None,
    cursor: str | None = None,
    **_unsupported: Any,
) -> PaginatedResponse[Person]:
    """
    Get a page of persons.

    Args:
        ids: Specific person IDs to fetch (batch lookup)
        field_ids: Specific field IDs to include in response
        field_types: Field types to include
        limit: Maximum number of results
        cursor: Cursor to resume pagination (opaque; obtained from prior responses)

    Returns:
        Paginated response with persons

    Note:
        The V2 /persons endpoint silently ignores any ``filter=`` parameter —
        passing one would return unfiltered results without warning. This method
        therefore rejects ``filter=`` with a ``ValueError``. To search persons
        by name or email use ``client.persons.search_pages(term)``. To filter
        by list-specific fields use ``client.lists.entries(list_id).list(filter=...)``.
    """
    if "filter" in _unsupported:
        raise ValueError(
            "V2 /persons does not support server-side filter. "
            "To search by name/email use client.persons.search_pages(term). "
            "To filter by list-specific fields use "
            "client.lists.entries(list_id).list(filter=...)."
        )
    if _unsupported:
        raise TypeError(f"Unexpected keyword arguments: {list(_unsupported)}")
    validate_entity_field_types(field_types, endpoint="person")
    if cursor is not None:
        if any(p is not None for p in (ids, field_ids, field_types, limit)):
            raise ValueError(
                "Cannot combine 'cursor' with other parameters; cursor encodes all query "
                "context. Start a new pagination sequence without a cursor to change "
                "parameters."
            )
        data = self._client.get_url(cursor)
    else:
        check_positive_limit(limit)
        ids = materialize_ids(ids)
        if no_ids(ids):
            return PaginatedResponse[Person]()
        params: dict[str, Any] = {}
        if ids:
            params["ids"] = [int(id_) for id_ in ids]
        if field_ids:
            params["fieldIds"] = [str(field_id) for field_id in field_ids]
        if field_types:
            params["fieldTypes"] = [field_type.value for field_type in field_types]
        if limit is not None:
            params["limit"] = limit
        data = self._client.get("/persons", params=params or None)

    return PaginatedResponse[Person](
        data=[Person.model_validate(p) for p in data.get("data", [])],
        pagination=PaginationInfo.model_validate(data.get("pagination", {})),
    )

list_merges(*, status: str | None = None, task_id: str | None = None, limit: int | None = None, cursor: str | None = None) -> PaginatedResponse[PersonMergeState]

One page of past person merges, newest first.

status is "in-progress", "success" or "failed"; task_id (or the task URL a merge returned) limits it to one merge task. Needs the "Manage duplicates" permission and the organization admin role (otherwise AuthorizationError).

Source code in affinity/services/persons.py

def list_merges(
    self,
    *,
    status: str | None = None,
    task_id: str | None = None,
    limit: int | None = None,
    cursor: str | None = None,
) -> PaginatedResponse[PersonMergeState]:
    """One page of past person merges, newest first.

    ``status`` is ``"in-progress"``, ``"success"`` or ``"failed"``; ``task_id`` (or the task
    URL a merge returned) limits it to one merge task. Needs the "Manage duplicates"
    permission and the organization admin role (otherwise ``AuthorizationError``).
    """
    check_cursor_alone(cursor, status, task_id, limit)
    if cursor is not None:
        return page_of(PersonMergeState, self._client.get_url(cursor))
    params = merge_params(status=status, task_id=task_id, limit=limit)
    return page_of(PersonMergeState, self._client.get("/person-merges", params=params or None))

list_notes(person_id: PersonId, *, creator_id: int | None = None, created_after: datetime | None = None, created_before: datetime | None = None, updated_after: datetime | None = None, limit: int | None = None, total_count: bool = False, cursor: str | None = None) -> PaginatedResponse[NoteV2]

One page of this person's notes as V2 notes (NoteV2; replies not included).

Filters: creator_id, created_after (inclusive) / created_before (exclusive), updated_after, rounded outward to whole seconds. limit is 1-100.

Source code in affinity/services/persons.py

def list_notes(
    self,
    person_id: PersonId,
    *,
    creator_id: int | None = None,
    created_after: datetime | None = None,
    created_before: datetime | None = None,
    updated_after: datetime | None = None,
    limit: int | None = None,
    total_count: bool = False,
    cursor: str | None = None,
) -> PaginatedResponse[NoteV2]:
    """One page of this person's notes as V2 notes (``NoteV2``; replies not included).

    Filters: ``creator_id``, ``created_after`` (inclusive) / ``created_before`` (exclusive),
    ``updated_after``, rounded outward to whole seconds. ``limit`` is 1-100.
    """
    check_cursor_alone(
        cursor, creator_id, created_after, created_before, updated_after, limit, total_count
    )
    if cursor is not None:
        return page_of(NoteV2, self._client.get_url(cursor, min_api_version=None))
    params = note_params(
        creator_id=creator_id,
        created_after=created_after,
        created_before=created_before,
        updated_after=updated_after,
        limit=limit,
        total_count=total_count,
    )
    return page_of(
        NoteV2,
        self._client.get(
            f"/persons/{int(person_id)}/notes",
            params=params or None,
            min_api_version=None,
        ),
    )

list_relationships(person_id: PersonId, *, min_score: float | None = None, order: Literal['asc', 'desc'] = 'desc', limit: int | None = None, total_count: bool = False, cursor: str | None = None) -> PaginatedResponse[Relationship]

One page of the relationships between your team and this person's people.

Strongest first (order="asc" for weakest first). min_score (0.0-1.0) keeps relationships at least that strong; relationships known only from LinkedIn have score 0, so any min_score above 0 drops them. limit is 1-100 (API default 20). Needs Affinity API version {RELATIONSHIPS_MIN_API_VERSION} (sent for this call and every page).

Source code in affinity/services/persons.py

def list_relationships(
    self,
    person_id: PersonId,
    *,
    min_score: float | None = None,
    order: Literal["asc", "desc"] = "desc",
    limit: int | None = None,
    total_count: bool = False,
    cursor: str | None = None,
) -> PaginatedResponse[Relationship]:
    """One page of the relationships between your team and this person's people.

    Strongest first (``order="asc"`` for weakest first). ``min_score`` (0.0-1.0) keeps
    relationships at least that strong; relationships known only from LinkedIn have score 0,
    so any ``min_score`` above 0 drops them. ``limit`` is 1-100 (API default 20). Needs
    Affinity API version {RELATIONSHIPS_MIN_API_VERSION} (sent for this call and every page).
    """
    if cursor is not None:
        if min_score is not None or limit is not None or total_count or order != "desc":
            raise ValueError(
                "Cannot combine 'cursor' with other parameters; cursor encodes all query "
                "context."
            )
        return page_of(
            Relationship,
            self._client.get_url(cursor, min_api_version=RELATIONSHIPS_MIN_API_VERSION),
        )
    params = relationship_params(
        min_score=min_score, order=order, limit=limit, total_count=total_count
    )
    return page_of(
        Relationship,
        self._client.get(
            f"/persons/{int(person_id)}/relationships",
            params=params or None,
            min_api_version=RELATIONSHIPS_MIN_API_VERSION,
        ),
    )

merge(primary_id: PersonId, duplicate_id: PersonId) -> str

Merge a duplicate person into a primary person.

Returns a task URL to check merge status.

Source code in affinity/services/persons.py

def merge(
    self,
    primary_id: PersonId,
    duplicate_id: PersonId,
) -> str:
    """
    Merge a duplicate person into a primary person.

    Returns a task URL to check merge status.
    """
    result = self._client.post(
        "/person-merges",
        json={
            "primaryPersonId": int(primary_id),
            "duplicatePersonId": int(duplicate_id),
        },
    )
    return merge_task_url(result, "person")

pages(*, ids: Sequence[PersonId] | None = None, field_ids: Sequence[AnyFieldId] | None = None, field_types: Sequence[FieldType] | None = None, limit: int | None = None, cursor: str | None = None, **_unsupported: Any) -> Iterator[PaginatedResponse[Person]]

Iterate person pages (not items), yielding PaginatedResponse[Person].

Useful for ETL scripts that need checkpoint/resume via page.next_cursor.

Parameters:

Name	Type	Description	Default
ids	Sequence[PersonId] | None	Specific person IDs to fetch (batch lookup)	None
field_ids	Sequence[AnyFieldId] | None	Specific field IDs to include in response	None
field_types	Sequence[FieldType] | None	Field types to include	None
limit	int | None	Maximum results per page	None
cursor	str | None	Cursor to resume pagination	None

Yields:

Type	Description
PaginatedResponse[Person]	PaginatedResponse[Person] for each page

Note

The V2 /persons endpoint silently ignores any filter= parameter — passing one would return unfiltered results without warning. This method therefore rejects filter= with a ValueError. To search persons by name or email use client.persons.search_pages(term). To filter by list-specific fields use client.lists.entries(list_id).list(filter=...).

Source code in affinity/services/persons.py

def pages(
    self,
    *,
    ids: Sequence[PersonId] | None = None,
    field_ids: Sequence[AnyFieldId] | None = None,
    field_types: Sequence[FieldType] | None = None,
    limit: int | None = None,
    cursor: str | None = None,
    **_unsupported: Any,
) -> Iterator[PaginatedResponse[Person]]:
    """
    Iterate person pages (not items), yielding `PaginatedResponse[Person]`.

    Useful for ETL scripts that need checkpoint/resume via `page.next_cursor`.

    Args:
        ids: Specific person IDs to fetch (batch lookup)
        field_ids: Specific field IDs to include in response
        field_types: Field types to include
        limit: Maximum results per page
        cursor: Cursor to resume pagination

    Yields:
        PaginatedResponse[Person] for each page

    Note:
        The V2 /persons endpoint silently ignores any ``filter=`` parameter —
        passing one would return unfiltered results without warning. This method
        therefore rejects ``filter=`` with a ``ValueError``. To search persons
        by name or email use ``client.persons.search_pages(term)``. To filter
        by list-specific fields use ``client.lists.entries(list_id).list(filter=...)``.
    """
    if "filter" in _unsupported:
        raise ValueError(
            "V2 /persons does not support server-side filter. "
            "To search by name/email use client.persons.search_pages(term). "
            "To filter by list-specific fields use "
            "client.lists.entries(list_id).list(filter=...)."
        )
    if _unsupported:
        raise TypeError(f"Unexpected keyword arguments: {list(_unsupported)}")
    other_params = (ids, field_ids, field_types, limit)
    if cursor is not None and any(p is not None for p in other_params):
        raise ValueError(
            "Cannot combine 'cursor' with other parameters; cursor encodes all query context. "
            "Start a new pagination sequence without a cursor to change parameters."
        )
    requested_cursor = cursor
    guard = _CursorGuard(cursor)
    page = (
        self.list(cursor=cursor)
        if cursor is not None
        else self.list(ids=ids, field_ids=field_ids, field_types=field_types, limit=limit)
    )
    while True:
        yield page
        # Stop at the last page, or when paging can't progress (see _CursorGuard).
        if guard.last_page(requested_cursor, page):
            return
        requested_cursor = page.next_cursor
        page = self.list(cursor=requested_cursor)

resolve(*, email: str | None = None, name: str | None = None) -> Person | None

Find a single person by email or name.

This is a convenience helper that searches and returns the first exact match, or None if not found. Uses V1 search internally.

Parameters:

Name	Type	Description	Default
email	str | None	Email address to search for	None
name	str | None	Person name to search for (first + last)	None

Returns:

Type	Description
Person | None	The matching Person, or None if not found

Raises:

Type	Description
ValueError	If neither email nor name is provided

Note

This auto-paginates V1 search results until a match is found. If multiple matches are found, returns the first one. For full disambiguation, use resolve_all() or search() directly.

Source code in affinity/services/persons.py

def resolve(
    self,
    *,
    email: str | None = None,
    name: str | None = None,
) -> Person | None:
    """
    Find a single person by email or name.

    This is a convenience helper that searches and returns the first exact match,
    or None if not found. Uses V1 search internally.

    Args:
        email: Email address to search for
        name: Person name to search for (first + last)

    Returns:
        The matching Person, or None if not found

    Raises:
        ValueError: If neither email nor name is provided

    Note:
        This auto-paginates V1 search results until a match is found.
        If multiple matches are found, returns the first one. For full
        disambiguation, use resolve_all() or search() directly.
    """
    if not email and not name:
        raise ValueError("Must provide either email or name")

    term = email or name or ""
    for page in self.search_pages(term, page_size=10):
        for person in page.data:
            if _person_matches(person, email=email, name=name):
                return person

    return None

resolve_all(*, email: str | None = None, name: str | None = None) -> builtins.list[Person]

Find all persons matching an email or name.

Notes: - This auto-paginates V1 search results to collect exact matches. - Unlike resolve(), this returns every match in server-provided order.

Source code in affinity/services/persons.py

def resolve_all(
    self,
    *,
    email: str | None = None,
    name: str | None = None,
) -> builtins.list[Person]:
    """
    Find all persons matching an email or name.

    Notes:
    - This auto-paginates V1 search results to collect exact matches.
    - Unlike resolve(), this returns every match in server-provided order.
    """
    if not email and not name:
        raise ValueError("Must provide either email or name")

    term = email or name or ""
    matches: builtins.list[Person] = []
    for page in self.search_pages(term, page_size=10):
        for person in page.data:
            if _person_matches(person, email=email, name=name):
                matches.append(person)
    return matches

search(term: str, *, with_interaction_dates: bool = False, with_interaction_persons: bool = False, with_opportunities: bool = False, page_size: int | None = None, page_token: str | None = None) -> PaginatedResponse[Person]

Search for persons by name or email.

Uses V1 API for search functionality.

Parameters:

Name	Type	Description	Default
term	str	Search term (name or email)	required
with_interaction_dates	bool	Include interaction date data	False
with_interaction_persons	bool	Include persons for interactions	False
with_opportunities	bool	Include associated opportunity IDs	False
page_size	int | None	Results per page (max 500)	None
page_token	str | None	Pagination token	None

Returns:

Type	Description
PaginatedResponse[Person]	PaginatedResponse[Person] with matching persons and pagination info

Source code in affinity/services/persons.py

def search(
    self,
    term: str,
    *,
    with_interaction_dates: bool = False,
    with_interaction_persons: bool = False,
    with_opportunities: bool = False,
    page_size: int | None = None,
    page_token: str | None = None,
) -> PaginatedResponse[Person]:
    """
    Search for persons by name or email.

    Uses V1 API for search functionality.

    Args:
        term: Search term (name or email)
        with_interaction_dates: Include interaction date data
        with_interaction_persons: Include persons for interactions
        with_opportunities: Include associated opportunity IDs
        page_size: Results per page (max 500)
        page_token: Pagination token

    Returns:
        PaginatedResponse[Person] with matching persons and pagination info
    """
    params: dict[str, Any] = {"term": term}
    if with_interaction_dates:
        params["with_interaction_dates"] = True
    if with_interaction_persons:
        params["with_interaction_persons"] = True
    if with_opportunities:
        params["with_opportunities"] = True
    if page_size:
        params["page_size"] = page_size
    if page_token:
        params["page_token"] = page_token

    data = self._client.get("/persons", params=params, v1=True)
    items = [Person.model_validate(p) for p in data.get("persons", [])]
    return PaginatedResponse[Person](
        data=items,
        next_page_token=data.get("next_page_token"),
    )

search_all(term: str, *, with_interaction_dates: bool = False, with_interaction_persons: bool = False, with_opportunities: bool = False, page_size: int | None = None, page_token: str | None = None) -> Iterator[Person]

Iterate all V1 person-search results with automatic pagination.

Parameters:

Name	Type	Description	Default
term	str	Search term (name or email)	required
with_interaction_dates	bool	Include interaction date data	False
with_interaction_persons	bool	Include persons for interactions	False
with_opportunities	bool	Include associated opportunity IDs	False
page_size	int | None	Results per page (max 500)	None
page_token	str | None	Resume from this pagination token	None

Yields:

Type	Description
Person	Person objects matching the search term

Source code in affinity/services/persons.py

def search_all(
    self,
    term: str,
    *,
    with_interaction_dates: bool = False,
    with_interaction_persons: bool = False,
    with_opportunities: bool = False,
    page_size: int | None = None,
    page_token: str | None = None,
) -> Iterator[Person]:
    """
    Iterate all V1 person-search results with automatic pagination.

    Args:
        term: Search term (name or email)
        with_interaction_dates: Include interaction date data
        with_interaction_persons: Include persons for interactions
        with_opportunities: Include associated opportunity IDs
        page_size: Results per page (max 500)
        page_token: Resume from this pagination token

    Yields:
        Person objects matching the search term
    """
    for page in self.search_pages(
        term,
        with_interaction_dates=with_interaction_dates,
        with_interaction_persons=with_interaction_persons,
        with_opportunities=with_opportunities,
        page_size=page_size,
        page_token=page_token,
    ):
        yield from page.data

search_pages(term: str, *, with_interaction_dates: bool = False, with_interaction_persons: bool = False, with_opportunities: bool = False, page_size: int | None = None, page_token: str | None = None) -> Iterator[PaginatedResponse[Person]]

Iterate V1 person-search result pages.

Useful for scripts that need checkpoint/resume via next_page_token.

Parameters:

Name	Type	Description	Default
term	str	Search term (name or email)	required
with_interaction_dates	bool	Include interaction date data	False
with_interaction_persons	bool	Include persons for interactions	False
with_opportunities	bool	Include associated opportunity IDs	False
page_size	int | None	Results per page (max 500)	None
page_token	str | None	Resume from this pagination token	None

Yields:

Type	Description
PaginatedResponse[Person]	PaginatedResponse[Person] for each page

Source code in affinity/services/persons.py

def search_pages(
    self,
    term: str,
    *,
    with_interaction_dates: bool = False,
    with_interaction_persons: bool = False,
    with_opportunities: bool = False,
    page_size: int | None = None,
    page_token: str | None = None,
) -> Iterator[PaginatedResponse[Person]]:
    """
    Iterate V1 person-search result pages.

    Useful for scripts that need checkpoint/resume via `next_page_token`.

    Args:
        term: Search term (name or email)
        with_interaction_dates: Include interaction date data
        with_interaction_persons: Include persons for interactions
        with_opportunities: Include associated opportunity IDs
        page_size: Results per page (max 500)
        page_token: Resume from this pagination token

    Yields:
        PaginatedResponse[Person] for each page
    """
    requested_token = page_token
    page = self.search(
        term,
        with_interaction_dates=with_interaction_dates,
        with_interaction_persons=with_interaction_persons,
        with_opportunities=with_opportunities,
        page_size=page_size,
        page_token=page_token,
    )
    while True:
        yield page
        next_token = page.next_page_token
        if not next_token or next_token == requested_token:
            return
        requested_token = next_token
        page = self.search(
            term,
            with_interaction_dates=with_interaction_dates,
            with_interaction_persons=with_interaction_persons,
            with_opportunities=with_opportunities,
            page_size=page_size,
            page_token=next_token,
        )

update(person_id: PersonId, data: PersonUpdate) -> Person

Update an existing person.

Note: To add emails/organizations, include existing values plus new ones.

Source code in affinity/services/persons.py

def update(
    self,
    person_id: PersonId,
    data: PersonUpdate,
) -> Person:
    """
    Update an existing person.

    Note: To add emails/organizations, include existing values plus new ones.
    """
    payload = data.model_dump(
        by_alias=True,
        mode="json",
        exclude_unset=True,
        exclude_none=True,
    )

    result = self._client.put(
        f"/persons/{person_id}",
        json=payload,
        v1=True,
    )

    if self._client.cache:
        self._client.cache.invalidate_prefix("person")

    return Person.model_validate(result)

Bases: AsyncPersonDiscoveryMixin, AsyncSingleFieldMixin, AsyncSearchMixin[Person]

Source code in affinity/services/persons.py

class AsyncPersonService(
    AsyncPersonDiscoveryMixin, AsyncSingleFieldMixin, AsyncSearchMixin[Person]
):
    _entity_path = "/persons"

    _search_model = Person

    def _search_path(self) -> str:
        return "/persons/search"

    async def _search_fields(self) -> builtins.list[FieldMetadata]:
        return await self.get_fields(includes=SEARCH_INCLUDES)

    """Async version of PersonService."""

    def __init__(self, client: AsyncHTTPClient):
        self._client = client

    async def list(
        self,
        *,
        ids: Sequence[PersonId] | None = None,
        field_ids: Sequence[AnyFieldId] | None = None,
        field_types: Sequence[FieldType] | None = None,
        limit: int | None = None,
        cursor: str | None = None,
        **_unsupported: Any,
    ) -> PaginatedResponse[Person]:
        """
        Get a page of persons.

        Args:
            ids: Specific person IDs to fetch (batch lookup)
            field_ids: Specific field IDs to include in response
            field_types: Field types to include
            limit: Maximum number of results
            cursor: Cursor to resume pagination (opaque; obtained from prior responses)

        Returns:
            Paginated response with persons

        Note:
            The V2 /persons endpoint silently ignores any ``filter=`` parameter —
            passing one would return unfiltered results without warning. This method
            therefore rejects ``filter=`` with a ``ValueError``. To search persons
            by name or email use ``client.persons.search_pages(term)``. To filter
            by list-specific fields use ``client.lists.entries(list_id).list(filter=...)``.
        """
        if "filter" in _unsupported:
            raise ValueError(
                "V2 /persons does not support server-side filter. "
                "To search by name/email use client.persons.search_pages(term). "
                "To filter by list-specific fields use "
                "client.lists.entries(list_id).list(filter=...)."
            )
        if _unsupported:
            raise TypeError(f"Unexpected keyword arguments: {list(_unsupported)}")
        validate_entity_field_types(field_types, endpoint="person")
        if cursor is not None:
            if any(p is not None for p in (ids, field_ids, field_types, limit)):
                raise ValueError(
                    "Cannot combine 'cursor' with other parameters; cursor encodes all query "
                    "context. Start a new pagination sequence without a cursor to change "
                    "parameters."
                )
            data = await self._client.get_url(cursor)
        else:
            check_positive_limit(limit)
            ids = materialize_ids(ids)
            if no_ids(ids):
                return PaginatedResponse[Person]()
            params: dict[str, Any] = {}
            if ids:
                params["ids"] = [int(id_) for id_ in ids]
            if field_ids:
                params["fieldIds"] = [str(field_id) for field_id in field_ids]
            if field_types:
                params["fieldTypes"] = [field_type.value for field_type in field_types]
            if limit is not None:
                params["limit"] = limit
            data = await self._client.get("/persons", params=params or None)

        return PaginatedResponse[Person](
            data=[Person.model_validate(p) for p in data.get("data", [])],
            pagination=PaginationInfo.model_validate(data.get("pagination", {})),
        )

    async def get_first(
        self,
        *,
        field_ids: Sequence[AnyFieldId] | None = None,
        field_types: Sequence[FieldType] | None = None,
        **_unsupported: Any,
    ) -> Person | None:
        """
        Get the first person, or None if no results.

        See PersonService.get_first() for details.

        Note:
            The V2 /persons endpoint silently ignores any ``filter=`` parameter —
            passing one would return unfiltered results without warning. This method
            therefore rejects ``filter=`` with a ``ValueError``. To search persons
            by name or email use ``client.persons.search_pages(term)``. To filter
            by list-specific fields use ``client.lists.entries(list_id).list(filter=...)``.
        """
        if "filter" in _unsupported:
            raise ValueError(
                "V2 /persons does not support server-side filter. "
                "To search by name/email use client.persons.search_pages(term). "
                "To filter by list-specific fields use "
                "client.lists.entries(list_id).list(filter=...)."
            )
        if _unsupported:
            raise TypeError(f"Unexpected keyword arguments: {list(_unsupported)}")
        page = await self.list(
            field_ids=field_ids,
            field_types=field_types,
            limit=1,
        )
        return page.data[0] if page.data else None

    async def batch_get(
        self,
        person_ids: Sequence[PersonId],
        *,
        field_ids: Sequence[AnyFieldId] | None = None,
        field_types: Sequence[FieldType] | None = None,
        include_field_values: bool = False,
        with_interaction_dates: bool = False,
        with_interaction_persons: bool = False,
        max_concurrent: int = 10,
        on_error: Literal["raise", "skip"] = "raise",
    ) -> dict[PersonId, Person]:
        """
        Fetch multiple persons with controlled concurrency.

        Makes individual get() calls with bounded concurrency: one call per id. With
        with_interaction_dates and field_ids/field_types, the fields come from one call per
        100 ids and the dates from one V1 call per id (about N + N/100 calls, not 2N).

        Args:
            person_ids: Person IDs to fetch
            field_ids: Specific field IDs to include
            field_types: Field types to include
            include_field_values: If True, fetch embedded field values
            with_interaction_dates: Include interaction date summaries
            with_interaction_persons: Include person IDs for interactions
            max_concurrent: Maximum concurrent API calls (default: 10)
            on_error: How to handle AffinityError exceptions:
                - "raise": Raise on first AffinityError (default)
                - "skip": Skip failed IDs, return partial results

        Returns:
            Dict mapping person_id -> Person for successfully fetched persons.

        Raises:
            AffinityError: If on_error="raise" and any fetch fails.
        """
        if not person_ids:
            return {}
        if max_concurrent < 1:
            raise ValueError("max_concurrent must be at least 1")

        unique_ids = list(dict.fromkeys(person_ids))
        if (
            with_interaction_dates
            and (field_ids is not None or field_types is not None)
            and not include_field_values
        ):
            # One V2 call per 100 ids for the fields, one V1 call per id for the dates
            v1_params: dict[str, Any] = {"with_interaction_dates": True}
            if with_interaction_persons:
                v1_params["with_interaction_persons"] = True
            v2_params: dict[str, Any] = {}
            if field_ids:
                v2_params["fieldIds"] = [str(fid) for fid in field_ids]
            if field_types:
                v2_params["fieldTypes"] = [ft.value for ft in field_types]
            return await batch_get_with_dates(
                self._client,
                unique_ids,
                v1_path="/persons",
                v2_path="/persons",
                kind="Person",
                v1_params=v1_params,
                v2_params=v2_params,
                max_concurrent=max_concurrent,
                on_error=on_error,
                get_one=lambda record_id: self.get(
                    record_id,
                    field_ids=field_ids,
                    field_types=field_types,
                    with_interaction_dates=True,
                    with_interaction_persons=with_interaction_persons,
                ),
                build=Person.model_validate,
            )
        results: dict[PersonId, Person] = {}

        async def fetch_one(pid: PersonId) -> tuple[PersonId, Person | None]:
            try:
                person = await self.get(
                    pid,
                    field_ids=field_ids,
                    field_types=field_types,
                    include_field_values=include_field_values,
                    with_interaction_dates=with_interaction_dates,
                    with_interaction_persons=with_interaction_persons,
                )
                return (pid, person)
            except AffinityError:
                if on_error == "raise":
                    raise
                return (pid, None)

        for i in range(0, len(unique_ids), max_concurrent):
            chunk = unique_ids[i : i + max_concurrent]
            tasks = [asyncio.create_task(fetch_one(pid)) for pid in chunk]
            try:
                for coro in asyncio.as_completed(tasks):
                    pid, person = await coro
                    if person is not None:
                        results[pid] = person
            except BaseException:
                for task in tasks:
                    if not task.done():
                        task.cancel()
                await asyncio.gather(*tasks, return_exceptions=True)
                raise

        return results

    def pages(
        self,
        *,
        ids: Sequence[PersonId] | None = None,
        field_ids: Sequence[AnyFieldId] | None = None,
        field_types: Sequence[FieldType] | None = None,
        limit: int | None = None,
        cursor: str | None = None,
        **_unsupported: Any,
    ) -> AsyncIterator[PaginatedResponse[Person]]:
        """
        Iterate person pages (not items), yielding `PaginatedResponse[Person]`.

        Useful for ETL scripts that need checkpoint/resume via `page.next_cursor`.

        Args:
            ids: Specific person IDs to fetch (batch lookup)
            field_ids: Specific field IDs to include in response
            field_types: Field types to include
            limit: Maximum results per page
            cursor: Cursor to resume pagination

        Returns:
            AsyncIterator of PaginatedResponse[Person]

        Note:
            The V2 /persons endpoint silently ignores any ``filter=`` parameter —
            passing one would return unfiltered results without warning. This method
            therefore rejects ``filter=`` with a ``ValueError``. To search persons
            by name or email use ``client.persons.search_pages(term)``. To filter
            by list-specific fields use ``client.lists.entries(list_id).list(filter=...)``.
        """
        if "filter" in _unsupported:
            raise ValueError(
                "V2 /persons does not support server-side filter. "
                "To search by name/email use client.persons.search_pages(term). "
                "To filter by list-specific fields use "
                "client.lists.entries(list_id).list(filter=...)."
            )
        if _unsupported:
            raise TypeError(f"Unexpected keyword arguments: {list(_unsupported)}")
        return self._pages_iter(
            ids=ids,
            field_ids=field_ids,
            field_types=field_types,
            limit=limit,
            cursor=cursor,
        )

    async def _pages_iter(
        self,
        *,
        ids: Sequence[PersonId] | None = None,
        field_ids: Sequence[AnyFieldId] | None = None,
        field_types: Sequence[FieldType] | None = None,
        limit: int | None = None,
        cursor: str | None = None,
    ) -> AsyncIterator[PaginatedResponse[Person]]:
        other_params = (ids, field_ids, field_types, limit)
        if cursor is not None and any(p is not None for p in other_params):
            raise ValueError(
                "Cannot combine 'cursor' with other parameters; cursor encodes all query context. "
                "Start a new pagination sequence without a cursor to change parameters."
            )
        requested_cursor = cursor
        guard = _CursorGuard(cursor)
        if cursor is not None:
            page = await self.list(cursor=cursor)
        else:
            page = await self.list(
                ids=ids,
                field_ids=field_ids,
                field_types=field_types,
                limit=limit,
            )
        while True:
            yield page
            # Stop at the last page, or when paging can't progress (see _CursorGuard).
            if guard.last_page(requested_cursor, page):
                return
            requested_cursor = page.next_cursor
            page = await self.list(cursor=requested_cursor)

    def all(
        self,
        *,
        ids: Sequence[PersonId] | None = None,
        field_ids: Sequence[AnyFieldId] | None = None,
        field_types: Sequence[FieldType] | None = None,
        **_unsupported: Any,
    ) -> AsyncIterator[Person]:
        """
        Iterate through all persons with automatic pagination.

        Args:
            ids: Specific person IDs to fetch (batch lookup)
            field_ids: Specific field IDs to include
            field_types: Field types to include

        Yields:
            Person objects

        Note:
            The V2 /persons endpoint silently ignores any ``filter=`` parameter —
            passing one would return unfiltered results without warning. This method
            therefore rejects ``filter=`` with a ``ValueError``. To search persons
            by name or email use ``client.persons.search_pages(term)``. To filter
            by list-specific fields use ``client.lists.entries(list_id).list(filter=...)``.
        """
        if "filter" in _unsupported:
            raise ValueError(
                "V2 /persons does not support server-side filter. "
                "To search by name/email use client.persons.search_pages(term). "
                "To filter by list-specific fields use "
                "client.lists.entries(list_id).list(filter=...)."
            )
        if _unsupported:
            raise TypeError(f"Unexpected keyword arguments: {list(_unsupported)}")

        async def fetch_page(next_url: str | None) -> PaginatedResponse[Person]:
            if next_url:
                data = await self._client.get_url(next_url)
                return PaginatedResponse[Person](
                    data=[Person.model_validate(p) for p in data.get("data", [])],
                    pagination=PaginationInfo.model_validate(data.get("pagination", {})),
                )
            return await self.list(ids=ids, field_ids=field_ids, field_types=field_types)

        return AsyncPageIterator(fetch_page)

    def iter(
        self,
        *,
        ids: Sequence[PersonId] | None = None,
        field_ids: Sequence[AnyFieldId] | None = None,
        field_types: Sequence[FieldType] | None = None,
        **_unsupported: Any,
    ) -> AsyncIterator[Person]:
        """
        Auto-paginate all persons.

        Alias for `all()` (FR-006 public contract).

        Note:
            The V2 /persons endpoint silently ignores any ``filter=`` parameter —
            passing one would return unfiltered results without warning. This method
            therefore rejects ``filter=`` with a ``ValueError``. To search persons
            by name or email use ``client.persons.search_pages(term)``. To filter
            by list-specific fields use ``client.lists.entries(list_id).list(filter=...)``.
        """
        if "filter" in _unsupported:
            raise ValueError(
                "V2 /persons does not support server-side filter. "
                "To search by name/email use client.persons.search_pages(term). "
                "To filter by list-specific fields use "
                "client.lists.entries(list_id).list(filter=...)."
            )
        if _unsupported:
            raise TypeError(f"Unexpected keyword arguments: {list(_unsupported)}")
        return self.all(ids=ids, field_ids=field_ids, field_types=field_types)

    async def get(
        self,
        person_id: PersonId,
        *,
        field_ids: Sequence[AnyFieldId] | None = None,
        field_types: Sequence[FieldType] | None = None,
        include_field_values: bool = False,
        retries: int = 0,
        with_interaction_dates: bool = False,
        with_interaction_persons: bool = False,
    ) -> Person:
        """
        Get a single person by ID.

        Args:
            person_id: The person ID
            field_ids: Specific field IDs to include in response
            field_types: Field types to include (e.g., ["enriched", "global"])
            include_field_values: If True, fetch embedded field values. This saves
                one API call when you need both person info and field values.
                Cannot be combined with field_ids/field_types.
            retries: Number of retries on 404 NotFoundError. Default is 0 (fail fast).
                Set to 2-3 if calling immediately after create() to handle eventual
                consistency lag.
            with_interaction_dates: Include interaction date summaries (last/next
                meeting dates, email dates).
            with_interaction_persons: Include person IDs for each interaction.
                Only applies when with_interaction_dates=True.

        Returns:
            Person object with requested field data.
            When include_field_values=True, the Person will have a `field_values`
            attribute containing the list of FieldValue objects.
            When with_interaction_dates=True, the Person will have interaction_dates
            and interactions populated.

        Raises:
            NotFoundError: If person does not exist after all retries.
            ValueError: If include_field_values is combined with field_ids/field_types.

        Note:
            When combining with_interaction_dates with field_ids/field_types,
            two API calls are made internally and the results are merged.
        """
        return await self._get_with_retry(
            person_id,
            field_ids=field_ids,
            field_types=field_types,
            include_field_values=include_field_values,
            retries=retries,
            with_interaction_dates=with_interaction_dates,
            with_interaction_persons=with_interaction_persons,
        )

    async def _get_with_retry(
        self,
        person_id: PersonId,
        *,
        field_ids: Sequence[AnyFieldId] | None = None,
        field_types: Sequence[FieldType] | None = None,
        include_field_values: bool = False,
        retries: int = 0,
        with_interaction_dates: bool = False,
        with_interaction_persons: bool = False,
    ) -> Person:
        """Internal: get with retry logic."""
        last_error: NotFoundError | None = None
        attempts = retries + 1  # retries=0 means 1 attempt

        for attempt in range(attempts):
            try:
                return await self._get_impl(
                    person_id,
                    field_ids=field_ids,
                    field_types=field_types,
                    include_field_values=include_field_values,
                    with_interaction_dates=with_interaction_dates,
                    with_interaction_persons=with_interaction_persons,
                )
            except NotFoundError as e:
                last_error = e
                if attempt < attempts - 1:  # Don't sleep after last attempt
                    await asyncio.sleep(0.5 * (attempt + 1))  # 0.5s, 1s, 1.5s backoff

        # V1 fallback: If V2 returned 404, try V1 API (handles V1→V2 sync delays)
        # Skip if already using V1 path (include_field_values or with_interaction_dates)
        if last_error is not None and not include_field_values and not with_interaction_dates:
            try:
                v1_data = await self._client.get(f"/persons/{person_id}", v1=True)
                normalized = _normalize_v1_person_response(v1_data)
                return Person.model_validate(normalized)
            except NotFoundError:
                pass  # V1 also failed, raise original V2 error

        raise last_error  # type: ignore[misc]

    async def _get_impl(
        self,
        person_id: PersonId,
        *,
        field_ids: Sequence[AnyFieldId] | None = None,
        field_types: Sequence[FieldType] | None = None,
        include_field_values: bool = False,
        with_interaction_dates: bool = False,
        with_interaction_persons: bool = False,
    ) -> Person:
        """Internal: actual get implementation."""
        has_field_filters = field_ids is not None or field_types is not None

        # include_field_values returns embedded field values, which is incompatible
        # with field_ids/field_types filtering (different data structures)
        if include_field_values and has_field_filters:
            raise ValueError(
                "Cannot combine 'include_field_values' with 'field_ids' or 'field_types'. "
                "Use include_field_values alone for embedded field values, or use "
                "field_ids/field_types alone for filtered fields."
            )

        # Path 1: include_field_values (V1 API, may include interaction dates)
        if include_field_values:
            v1_params: dict[str, Any] = {}
            if with_interaction_dates:
                v1_params["with_interaction_dates"] = True
            if with_interaction_persons:
                v1_params["with_interaction_persons"] = True

            data = await self._client.get(
                f"/persons/{person_id}",
                params=v1_params or None,
                v1=True,
            )

            field_values_data = data.get("field_values", [])
            # V1 embedded field_values don't include entityId; add it before validation
            field_values = [
                FieldValue.model_validate({**fv, "entityId": int(person_id)})
                for fv in field_values_data
            ]
            normalized = _normalize_v1_person_response(data)
            person = Person.model_validate(normalized)
            object.__setattr__(person, "field_values", field_values)
            return person

        # Path 2: with_interaction_dates (may need merge if field filters present)
        if with_interaction_dates:
            v1_params = {"with_interaction_dates": True}
            if with_interaction_persons:
                v1_params["with_interaction_persons"] = True

            interaction_data = await self._client.get(
                f"/persons/{person_id}",
                params=v1_params,
                v1=True,
            )

            # If field filtering is also requested, fetch filtered fields and merge
            if has_field_filters:
                v2_params: dict[str, Any] = {}
                if field_ids:
                    v2_params["fieldIds"] = [str(fid) for fid in field_ids]
                if field_types:
                    v2_params["fieldTypes"] = [ft.value for ft in field_types]

                filtered_data = await self._client.get(
                    f"/persons/{person_id}",
                    params=v2_params,
                )

                # Merge: filtered fields + interaction data
                filtered_data["interaction_dates"] = interaction_data.get("interaction_dates")
                filtered_data["interactions"] = interaction_data.get("interactions")
                return Person.model_validate(filtered_data)

            # No field filtering, normalize and return V1 data
            normalized = _normalize_v1_person_response(interaction_data)
            return Person.model_validate(normalized)

        # Path 3: Standard path (supports field filtering)
        params: dict[str, Any] = {}
        if field_ids:
            params["fieldIds"] = [str(field_id) for field_id in field_ids]
        if field_types:
            params["fieldTypes"] = [field_type.value for field_type in field_types]

        data = await self._client.get(f"/persons/{person_id}", params=params or None)
        return Person.model_validate(data)

    async def get_list_entries(
        self,
        person_id: PersonId,
        *,
        limit: int | None = None,
        cursor: str | None = None,
    ) -> PaginatedResponse[ListEntry]:
        """One page of a person's list entries across all lists.

        Each entry's ``entity_id`` is ``person_id`` and its ``entity_type`` is
        ``EntityType.PERSON``; ``entity`` is ``None`` (this endpoint doesn't embed the person).
        Follow ``pagination.next_cursor`` with ``cursor=`` (from this person's entries) for
        the next page.
        """
        path = f"/persons/{person_id}/list-entries"
        if cursor is not None:
            if limit is not None:
                raise ValueError(
                    "Cannot combine 'cursor' with other parameters; cursor encodes all query "
                    "context. Start a new pagination sequence without a cursor to change "
                    "parameters."
                )
            check_list_entries_cursor(cursor, path)
            data = await self._client.get_url(cursor)
        else:
            check_positive_limit(limit)
            data = await self._client.get(
                path, params={"limit": limit} if limit is not None else None
            )

        return entity_list_entries_page(data, entity_id=person_id, entity_type=EntityType.PERSON)

    async def get_lists(
        self,
        person_id: PersonId,
    ) -> PaginatedResponse[ListSummary]:
        """Get all lists that contain this person."""
        data = await self._client.get(f"/persons/{person_id}/lists")

        return PaginatedResponse[ListSummary](
            data=[ListSummary.model_validate(item) for item in data.get("data", [])],
            pagination=PaginationInfo.model_validate(data.get("pagination", {})),
        )

    async def get_fields(
        self,
        *,
        filter: str | None = None,
        includes: Sequence[str] | None = None,
    ) -> builtins.list[FieldMetadata]:
        """
        Get metadata about person fields.

        Cached for performance.

        Args:
            filter: Filter on the field name: ``name="Location"`` (exact) or ``name=~Loc``
                (substring); both case-sensitive. Combine clauses with ``|``. Only ``name``
                is filterable; anything else is rejected with 400.
            includes: Extra properties per field: ``"filterability"``, ``"sortability"``
                (read them via ``is_filterable`` / ``is_sortable``).
        """
        params, cache_key = build_fields_query(
            "person_fields:",
            filter=filter,
            includes=includes,
        )

        data = await self._client.get_all_pages(
            "/persons/fields",
            params=params or None,
            cache_key=cache_key,
            cache_ttl=300,
        )

        return [FieldMetadata.model_validate(f) for f in data.get("data", [])]

    async def get_field_dropdown_options(
        self, field_id: AnyFieldId, *, with_status_types: bool = False
    ) -> builtins.list[DropdownOption]:
        """All options of a person dropdown, ranked-dropdown or status field (every page).

        Options with ``type``, ``rank``, ``color`` and (status fields) ``status_category`` /
        ``win_rate``. Options of global person fields can only be read; list-specific fields'
        options are managed through ``client.lists``. ``with_status_types=True``: see
        ``lists.get_field_dropdown_options``.
        """
        data = await self._client.get_all_pages(
            options_path(field_id, list_id=None, entity="persons"),
            params={"limit": 100},
            min_api_version=DROPDOWN_OPTION_WRITES_MIN_API_VERSION if with_status_types else None,
        )
        return parse_options(data)

    async def get_field_values(
        self,
        person_id: PersonId,
        *,
        ids: Sequence[AnyFieldId | str] | None = None,
    ) -> FieldValues:
        """
        Field values of one person, keyed by field id (``GET /v2/persons/{id}/fields``).

        Args:
            person_id: The person.
            ids: Only these fields (server-side filter); all fields when omitted.
        """
        params = {"ids": [str(fid) for fid in ids]} if ids else None
        data = await self._client.get_all_pages(f"/persons/{person_id}/fields", params=params)
        return FieldValues.model_validate(data.get("data", []))

    async def batch_update_fields(
        self,
        person_id: PersonId,
        updates: Mapping[AnyFieldId | str, Any],
        *,
        value_types: Mapping[AnyFieldId | str, FieldValueType | str] | None = None,
    ) -> BatchOperationResponse:
        """
        Update several field values of one person in one request.

        Affinity applies the whole batch or none of it; each update replaces the field's value
        and ``None`` (with a type) clears it. Global, enriched and Source of Introduction
        fields are all writable. At most 100 updates, and 100 values per multi-value field.

        Needs Affinity API version {FIELD_WRITES_MIN_API_VERSION} (out of beta there): an
        unpinned client sends that version for this call; a client pinned to an older version
        raises ``ApiVersionTooOldError`` without sending anything.

        Args:
            person_id: The person.
            updates: Field id -> new value. Values are auto-typed (str→text, int/float→number,
                datetime/date→datetime) unless the field has an entry in ``value_types``.
            value_types: Field id -> value type (e.g. ``"dropdown-multi"``); needed for lists
                (multi-value fields) and ``None`` (clear).
        """
        items = _batch_update_items(updates, value_types)
        result = await self._client.patch(
            f"/persons/{person_id}/fields",
            json={"operation": "update-fields", "updates": items},
            min_api_version=FIELD_WRITES_MIN_API_VERSION,
        )
        return BatchOperationResponse.model_validate(result)

    # =========================================================================
    # Associations (V1 API)
    # =========================================================================

    async def get_associated_company_ids(
        self,
        person_id: PersonId,
        *,
        max_results: int | None = None,
    ) -> builtins.list[CompanyId]:
        """
        Get associated company IDs for a person.

        V1-only: V2 does not expose person -> company associations directly.
        Uses GET `/persons/{id}` (V1) and returns `organization_ids`.

        Args:
            person_id: The person ID
            max_results: Maximum number of company IDs to return

        Returns:
            List of CompanyId values associated with this person

        Note:
            The Person model already has `company_ids` populated from V1's
            `organizationIds` field. This method provides API parity with
            `CompanyService.get_associated_person_ids()`.
        """
        data = await self._client.get(f"/persons/{person_id}", v1=True)
        # Defensive: handle potential {"person": {...}} wrapper
        # (consistent with CompanyService.get_associated_person_ids pattern)
        person = data.get("person") if isinstance(data, dict) else None
        source = person if isinstance(person, dict) else data
        org_ids = None
        if isinstance(source, dict):
            org_ids = source.get("organization_ids") or source.get("organizationIds")

        if not isinstance(org_ids, list):
            return []

        ids = [CompanyId(int(cid)) for cid in org_ids if cid is not None]
        if max_results is not None and max_results >= 0:
            return ids[:max_results]
        return ids

    async def get_associated_company_ids_batch(
        self,
        person_ids: Sequence[PersonId],
        *,
        on_error: Literal["raise", "skip"] = "raise",
    ) -> dict[PersonId, builtins.list[CompanyId]]:
        """
        Get company associations for multiple persons.

        Makes one V1 API call per person.

        Args:
            person_ids: Sequence of person IDs to fetch
            on_error: How to handle errors - "raise" (default) or "skip" failed IDs

        Returns:
            Dict mapping person_id -> list of company_ids

        Raises:
            AffinityError: If on_error="raise" and any fetch fails.
        """
        result: dict[PersonId, builtins.list[CompanyId]] = {}
        for person_id in person_ids:
            try:
                result[person_id] = await self.get_associated_company_ids(person_id)
            except AffinityError:
                if on_error == "raise":
                    raise
                # skip: continue without this person
            except Exception as e:
                if on_error == "raise":
                    raise AffinityError(
                        f"Failed to get associations for person {person_id}: {e}"
                    ) from e
                # skip: continue without this person
        return result

    async def get_associated_opportunity_ids(
        self,
        person_id: PersonId,
        *,
        max_results: int | None = None,
    ) -> builtins.list[OpportunityId]:
        """
        Get associated opportunity IDs for a person.

        V1-only: V2 does not expose person -> opportunity associations directly.
        Uses GET `/persons/{id}` (V1) and returns `opportunity_ids`.

        Args:
            person_id: The person ID
            max_results: Maximum number of opportunity IDs to return

        Returns:
            List of OpportunityId values associated with this person

        Note:
            The Person model already has `opportunity_ids` populated from V1's
            `opportunityIds` field. This method provides API parity with
            `OpportunityService.get_associated_person_ids()`.
        """
        data = await self._client.get(f"/persons/{person_id}", v1=True)
        # Defensive: handle potential {"person": {...}} wrapper
        person = data.get("person") if isinstance(data, dict) else None
        source = person if isinstance(person, dict) else data
        opp_ids = None
        if isinstance(source, dict):
            opp_ids = source.get("opportunity_ids") or source.get("opportunityIds")

        if not isinstance(opp_ids, list):
            return []

        ids = [OpportunityId(int(oid)) for oid in opp_ids if oid is not None]
        if max_results is not None and max_results >= 0:
            return ids[:max_results]
        return ids

    async def get_associated_opportunity_ids_batch(
        self,
        person_ids: Sequence[PersonId],
        *,
        on_error: Literal["raise", "skip"] = "raise",
    ) -> dict[PersonId, builtins.list[OpportunityId]]:
        """
        Get opportunity associations for multiple persons.

        Makes one V1 API call per person.

        Args:
            person_ids: Sequence of person IDs to fetch
            on_error: How to handle errors - "raise" (default) or "skip" failed IDs

        Returns:
            Dict mapping person_id -> list of opportunity_ids

        Raises:
            AffinityError: If on_error="raise" and any fetch fails.
        """
        result: dict[PersonId, builtins.list[OpportunityId]] = {}
        for person_id in person_ids:
            try:
                result[person_id] = await self.get_associated_opportunity_ids(person_id)
            except AffinityError:
                if on_error == "raise":
                    raise
                # skip: continue without this person
            except Exception as e:
                if on_error == "raise":
                    raise AffinityError(
                        f"Failed to get associations for person {person_id}: {e}"
                    ) from e
                # skip: continue without this person
        return result

    # =========================================================================
    # Search (V1 API)
    # =========================================================================

    async def search(
        self,
        term: str,
        *,
        with_interaction_dates: bool = False,
        with_interaction_persons: bool = False,
        with_opportunities: bool = False,
        page_size: int | None = None,
        page_token: str | None = None,
    ) -> PaginatedResponse[Person]:
        """
        Search for persons by name or email.

        Uses V1 API for search functionality.
        """
        params: dict[str, Any] = {"term": term}
        if with_interaction_dates:
            params["with_interaction_dates"] = True
        if with_interaction_persons:
            params["with_interaction_persons"] = True
        if with_opportunities:
            params["with_opportunities"] = True
        if page_size:
            params["page_size"] = page_size
        if page_token:
            params["page_token"] = page_token

        data = await self._client.get("/persons", params=params, v1=True)
        items = [Person.model_validate(p) for p in data.get("persons", [])]
        return PaginatedResponse[Person](
            data=items,
            next_page_token=data.get("next_page_token"),
        )

    async def search_pages(
        self,
        term: str,
        *,
        with_interaction_dates: bool = False,
        with_interaction_persons: bool = False,
        with_opportunities: bool = False,
        page_size: int | None = None,
        page_token: str | None = None,
    ) -> AsyncIterator[PaginatedResponse[Person]]:
        """
        Iterate V1 person-search result pages.

        Useful for scripts that need checkpoint/resume via `next_page_token`.

        Args:
            term: Search term (name or email)
            with_interaction_dates: Include interaction date data
            with_interaction_persons: Include persons for interactions
            with_opportunities: Include associated opportunity IDs
            page_size: Results per page (max 500)
            page_token: Resume from this pagination token

        Yields:
            PaginatedResponse[Person] for each page
        """
        requested_token = page_token
        page = await self.search(
            term,
            with_interaction_dates=with_interaction_dates,
            with_interaction_persons=with_interaction_persons,
            with_opportunities=with_opportunities,
            page_size=page_size,
            page_token=page_token,
        )
        while True:
            yield page
            next_token = page.next_page_token
            if not next_token or next_token == requested_token:
                return
            requested_token = next_token
            page = await self.search(
                term,
                with_interaction_dates=with_interaction_dates,
                with_interaction_persons=with_interaction_persons,
                with_opportunities=with_opportunities,
                page_size=page_size,
                page_token=next_token,
            )

    async def search_all(
        self,
        term: str,
        *,
        with_interaction_dates: bool = False,
        with_interaction_persons: bool = False,
        with_opportunities: bool = False,
        page_size: int | None = None,
        page_token: str | None = None,
    ) -> AsyncIterator[Person]:
        """
        Iterate all V1 person-search results with automatic pagination.

        Args:
            term: Search term (name or email)
            with_interaction_dates: Include interaction date data
            with_interaction_persons: Include persons for interactions
            with_opportunities: Include associated opportunity IDs
            page_size: Results per page (max 500)
            page_token: Resume from this pagination token

        Yields:
            Person objects matching the search term
        """
        async for page in self.search_pages(
            term,
            with_interaction_dates=with_interaction_dates,
            with_interaction_persons=with_interaction_persons,
            with_opportunities=with_opportunities,
            page_size=page_size,
            page_token=page_token,
        ):
            for person in page.data:
                yield person

    async def resolve(
        self,
        *,
        email: str | None = None,
        name: str | None = None,
    ) -> Person | None:
        """
        Find a single person by email or name.

        This is a convenience helper that searches and returns the first exact match,
        or None if not found. Uses V1 search internally.
        """
        if not email and not name:
            raise ValueError("Must provide either email or name")

        term = email or name or ""
        async for page in self.search_pages(term, page_size=10):
            for person in page.data:
                if _person_matches(person, email=email, name=name):
                    return person

        return None

    async def resolve_all(
        self,
        *,
        email: str | None = None,
        name: str | None = None,
    ) -> builtins.list[Person]:
        """
        Find all persons matching an email or name.

        Notes:
        - This auto-paginates V1 search results to collect exact matches.
        - Unlike resolve(), this returns every match in server-provided order.
        """
        if not email and not name:
            raise ValueError("Must provide either email or name")

        term = email or name or ""
        matches: builtins.list[Person] = []
        async for page in self.search_pages(term, page_size=10):
            for person in page.data:
                if _person_matches(person, email=email, name=name):
                    matches.append(person)
        return matches

    # =========================================================================
    # Write Operations (V1 API)
    # =========================================================================

    _DEDUP_PAGE_CAP = 3  # 3 pages x 500 = 1500 fuzzy results scanned per search term

    async def create(self, data: PersonCreate, *, if_not_exists: bool = True) -> Person:
        """
        Create a new person.

        Args:
            data: Person creation data
            if_not_exists: If True (default), check for an existing person with
                the same primary email (case-insensitive exact) — falling back
                to exact first+last name match when no email is provided.
                Uses V1 search_pages(term) with page_size=500, capped at
                _DEDUP_PAGE_CAP pages. Set to False to skip the check and
                create unconditionally.

        Returns:
            Created person

        Raises:
            DuplicateEntityError: When if_not_exists=True and an exact-email
                OR exact full-name match already exists.

        Note:
            Creates use V1 API, while reads use V2 API. Due to eventual consistency
            between V1 and V2, a `get()` call immediately after `create()` may return
            404 NotFoundError. If you need to read immediately after creation, either:
            - Use the Person object returned by this method (it contains the created data)
            - Add a short delay (100-500ms) before calling get()
            - Implement retry logic in your application
        """
        if if_not_exists:
            existing = await self._find_exact_duplicate(
                first_name=data.first_name,
                last_name=data.last_name,
                emails=list(data.emails or []),
            )
            if existing is not None:
                raise DuplicateEntityError(
                    f"Person with email or name already exists (id={existing.id})",
                    entity_type="person",
                    existing_id=int(existing.id),
                    existing_name=f"{existing.first_name or ''} {existing.last_name or ''}".strip(),
                )

        payload = data.model_dump(by_alias=True, mode="json")
        if not data.company_ids:
            payload.pop("organization_ids", None)

        result = await self._client.post("/persons", json=payload, v1=True)

        if self._client.cache:
            self._client.cache.invalidate_prefix("person")

        return Person.model_validate(result)

    async def _find_exact_duplicate(
        self,
        *,
        first_name: str | None,
        last_name: str | None,
        emails: builtins.list[str],
    ) -> Person | None:
        """Search V1 for an exact-email or exact full-name match. Returns None if no match.

        Email takes priority — exact case-insensitive match against any entry in the
        person's emails list or primary_email. Falls back to exact first+last name match
        when no emails are provided in the create payload.

        Iterates at most `_DEDUP_PAGE_CAP` pages per search term (1500 fuzzy results).
        """
        emails_lower = {e.strip().lower() for e in emails if e}
        fn_lower = (first_name or "").strip().lower()
        ln_lower = (last_name or "").strip().lower()

        search_terms = list(emails) if emails else [f"{first_name or ''} {last_name or ''}".strip()]
        seen_ids: set[int] = set()

        for term in search_terms:
            if not term:
                continue
            page_num = 0
            async for page in self.search_pages(term, page_size=500):
                for person in page.data:
                    pid = int(person.id)
                    if pid in seen_ids:
                        continue
                    seen_ids.add(pid)

                    person_emails = {(e or "").strip().lower() for e in (person.emails or [])}
                    if person.primary_email:
                        person_emails.add(person.primary_email.strip().lower())

                    if emails_lower and emails_lower & person_emails:
                        return person

                    if (
                        not emails_lower
                        and fn_lower
                        and ln_lower
                        and (person.first_name or "").lower() == fn_lower
                        and (person.last_name or "").lower() == ln_lower
                    ):
                        return person

                page_num += 1
                if page_num >= self._DEDUP_PAGE_CAP:
                    break
                if not page.next_page_token:
                    break

        return None

    async def update(
        self,
        person_id: PersonId,
        data: PersonUpdate,
    ) -> Person:
        """
        Update an existing person.

        Note: To add emails/organizations, include existing values plus new ones.
        """
        payload = data.model_dump(
            by_alias=True,
            mode="json",
            exclude_unset=True,
            exclude_none=True,
        )

        result = await self._client.put(
            f"/persons/{person_id}",
            json=payload,
            v1=True,
        )

        if self._client.cache:
            self._client.cache.invalidate_prefix("person")

        return Person.model_validate(result)

    async def delete(self, person_id: PersonId) -> bool:
        """Delete a person (also deletes associated field values)."""
        result = await self._client.delete(f"/persons/{person_id}", v1=True)
        invalidate_all_list_sizes(self._client)

        if self._client.cache:
            self._client.cache.invalidate_prefix("person")

        return bool(result.get("success", True))

    # =========================================================================
    # Merge Operations (V2 BETA)
    # =========================================================================

    async def merge(
        self,
        primary_id: PersonId,
        duplicate_id: PersonId,
    ) -> str:
        """
        Merge a duplicate person into a primary person.

        Returns a task URL to check merge status.
        """
        result = await self._client.post(
            "/person-merges",
            json={
                "primaryPersonId": int(primary_id),
                "duplicatePersonId": int(duplicate_id),
            },
        )
        return merge_task_url(result, "person")

    async def get_merge_status(self, task_id: str) -> MergeTask:
        """Check the status of a merge operation."""
        data = await self._client.get(f"/tasks/person-merges/{task_id}")
        return MergeTask.model_validate(data)

    # =========================================================================
    # Merge history and relationships (V2)
    # =========================================================================

    async def list_merges(
        self,
        *,
        status: str | None = None,
        task_id: str | None = None,
        limit: int | None = None,
        cursor: str | None = None,
    ) -> PaginatedResponse[PersonMergeState]:
        """One page of past person merges, newest first.

        ``status`` is ``"in-progress"``, ``"success"`` or ``"failed"``; ``task_id`` (or the task
        URL a merge returned) limits it to one merge task. Needs the "Manage duplicates"
        permission and the organization admin role (otherwise ``AuthorizationError``).
        """
        check_cursor_alone(cursor, status, task_id, limit)
        if cursor is not None:
            return page_of(PersonMergeState, await self._client.get_url(cursor))
        params = merge_params(status=status, task_id=task_id, limit=limit)
        return page_of(
            PersonMergeState, await self._client.get("/person-merges", params=params or None)
        )

    def iter_merges(
        self, *, status: str | None = None, task_id: str | None = None
    ) -> AsyncIterator[PersonMergeState]:
        """All past person merges matching the filters (every page)."""

        async def fetch_page(next_url: str | None) -> PaginatedResponse[PersonMergeState]:
            if next_url:
                return await self.list_merges(cursor=next_url)
            return await self.list_merges(status=status, task_id=task_id)

        return AsyncPageIterator(fetch_page)

    async def get_merge_state(self, merge_id: int) -> PersonMergeState:
        """One past person merge (see :meth:`list_merges`)."""
        return PersonMergeState.model_validate(
            await self._client.get(f"/person-merges/{int(merge_id)}")
        )

    async def list_relationships(
        self,
        person_id: PersonId,
        *,
        min_score: float | None = None,
        order: Literal["asc", "desc"] = "desc",
        limit: int | None = None,
        total_count: bool = False,
        cursor: str | None = None,
    ) -> PaginatedResponse[Relationship]:
        """One page of the relationships between your team and this person's people.

        Strongest first (``order="asc"`` for weakest first). ``min_score`` (0.0-1.0) keeps
        relationships at least that strong; relationships known only from LinkedIn have score 0,
        so any ``min_score`` above 0 drops them. ``limit`` is 1-100 (API default 20). Needs
        Affinity API version {RELATIONSHIPS_MIN_API_VERSION} (sent for this call and every page).
        """
        if cursor is not None:
            if min_score is not None or limit is not None or total_count or order != "desc":
                raise ValueError(
                    "Cannot combine 'cursor' with other parameters; cursor encodes all query "
                    "context."
                )
            return page_of(
                Relationship,
                await self._client.get_url(cursor, min_api_version=RELATIONSHIPS_MIN_API_VERSION),
            )
        params = relationship_params(
            min_score=min_score, order=order, limit=limit, total_count=total_count
        )
        return page_of(
            Relationship,
            await self._client.get(
                f"/persons/{int(person_id)}/relationships",
                params=params or None,
                min_api_version=RELATIONSHIPS_MIN_API_VERSION,
            ),
        )

    def iter_relationships(
        self,
        person_id: PersonId,
        *,
        min_score: float | None = None,
        order: Literal["asc", "desc"] = "desc",
        limit: int | None = None,
    ) -> AsyncIterator[Relationship]:
        """All relationships for this person (every page); see :meth:`list_relationships`."""

        async def fetch_page(next_url: str | None) -> PaginatedResponse[Relationship]:
            if next_url:
                return await self.list_relationships(person_id, cursor=next_url)
            return await self.list_relationships(
                person_id, min_score=min_score, order=order, limit=limit
            )

        return AsyncPageIterator(fetch_page)

    # =========================================================================
    # Notes (V2)
    # =========================================================================

    async def list_notes(
        self,
        person_id: PersonId,
        *,
        creator_id: int | None = None,
        created_after: datetime | None = None,
        created_before: datetime | None = None,
        updated_after: datetime | None = None,
        limit: int | None = None,
        total_count: bool = False,
        cursor: str | None = None,
    ) -> PaginatedResponse[NoteV2]:
        """One page of this person's notes as V2 notes (``NoteV2``; replies not included).

        Filters: ``creator_id``, ``created_after`` (inclusive) / ``created_before`` (exclusive),
        ``updated_after``, rounded outward to whole seconds. ``limit`` is 1-100.
        """
        check_cursor_alone(
            cursor, creator_id, created_after, created_before, updated_after, limit, total_count
        )
        if cursor is not None:
            return page_of(NoteV2, await self._client.get_url(cursor, min_api_version=None))
        params = note_params(
            creator_id=creator_id,
            created_after=created_after,
            created_before=created_before,
            updated_after=updated_after,
            limit=limit,
            total_count=total_count,
        )
        return page_of(
            NoteV2,
            await self._client.get(
                f"/persons/{int(person_id)}/notes",
                params=params or None,
                min_api_version=None,
            ),
        )

    def iter_notes(
        self,
        person_id: PersonId,
        *,
        creator_id: int | None = None,
        created_after: datetime | None = None,
        created_before: datetime | None = None,
        updated_after: datetime | None = None,
        limit: int | None = None,
    ) -> AsyncIterator[NoteV2]:
        """All of this person's notes (every page); see :meth:`list_notes`."""

        async def fetch_page(next_url: str | None) -> PaginatedResponse[NoteV2]:
            if next_url:
                return await self.list_notes(person_id, cursor=next_url)
            return await self.list_notes(
                person_id,
                creator_id=creator_id,
                created_after=created_after,
                created_before=created_before,
                updated_after=updated_after,
                limit=limit,
            )

        return AsyncPageIterator(fetch_page)

all(*, ids: Sequence[PersonId] | None = None, field_ids: Sequence[AnyFieldId] | None = None, field_types: Sequence[FieldType] | None = None, **_unsupported: Any) -> AsyncIterator[Person]

Iterate through all persons with automatic pagination.

Parameters:

Name	Type	Description	Default
ids	Sequence[PersonId] | None	Specific person IDs to fetch (batch lookup)	None
field_ids	Sequence[AnyFieldId] | None	Specific field IDs to include	None
field_types	Sequence[FieldType] | None	Field types to include	None

Yields:

Type	Description
AsyncIterator[Person]	Person objects

Note

The V2 /persons endpoint silently ignores any filter= parameter — passing one would return unfiltered results without warning. This method therefore rejects filter= with a ValueError. To search persons by name or email use client.persons.search_pages(term). To filter by list-specific fields use client.lists.entries(list_id).list(filter=...).

Source code in affinity/services/persons.py

def all(
    self,
    *,
    ids: Sequence[PersonId] | None = None,
    field_ids: Sequence[AnyFieldId] | None = None,
    field_types: Sequence[FieldType] | None = None,
    **_unsupported: Any,
) -> AsyncIterator[Person]:
    """
    Iterate through all persons with automatic pagination.

    Args:
        ids: Specific person IDs to fetch (batch lookup)
        field_ids: Specific field IDs to include
        field_types: Field types to include

    Yields:
        Person objects

    Note:
        The V2 /persons endpoint silently ignores any ``filter=`` parameter —
        passing one would return unfiltered results without warning. This method
        therefore rejects ``filter=`` with a ``ValueError``. To search persons
        by name or email use ``client.persons.search_pages(term)``. To filter
        by list-specific fields use ``client.lists.entries(list_id).list(filter=...)``.
    """
    if "filter" in _unsupported:
        raise ValueError(
            "V2 /persons does not support server-side filter. "
            "To search by name/email use client.persons.search_pages(term). "
            "To filter by list-specific fields use "
            "client.lists.entries(list_id).list(filter=...)."
        )
    if _unsupported:
        raise TypeError(f"Unexpected keyword arguments: {list(_unsupported)}")

    async def fetch_page(next_url: str | None) -> PaginatedResponse[Person]:
        if next_url:
            data = await self._client.get_url(next_url)
            return PaginatedResponse[Person](
                data=[Person.model_validate(p) for p in data.get("data", [])],
                pagination=PaginationInfo.model_validate(data.get("pagination", {})),
            )
        return await self.list(ids=ids, field_ids=field_ids, field_types=field_types)

    return AsyncPageIterator(fetch_page)

batch_get(person_ids: Sequence[PersonId], *, field_ids: Sequence[AnyFieldId] | None = None, field_types: Sequence[FieldType] | None = None, include_field_values: bool = False, with_interaction_dates: bool = False, with_interaction_persons: bool = False, max_concurrent: int = 10, on_error: Literal['raise', 'skip'] = 'raise') -> dict[PersonId, Person] async

Fetch multiple persons with controlled concurrency.

Makes individual get() calls with bounded concurrency: one call per id. With with_interaction_dates and field_ids/field_types, the fields come from one call per 100 ids and the dates from one V1 call per id (about N + N/100 calls, not 2N).

Parameters:

Name	Type	Description	Default
person_ids	Sequence[PersonId]	Person IDs to fetch	required
field_ids	Sequence[AnyFieldId] | None	Specific field IDs to include	None
field_types	Sequence[FieldType] | None	Field types to include	None
include_field_values	bool	If True, fetch embedded field values	False
with_interaction_dates	bool	Include interaction date summaries	False
with_interaction_persons	bool	Include person IDs for interactions	False
max_concurrent	int	Maximum concurrent API calls (default: 10)	10
on_error	Literal['raise', 'skip']	How to handle AffinityError exceptions: - "raise": Raise on first AffinityError (default) - "skip": Skip failed IDs, return partial results	'raise'

Returns:

Type	Description
dict[PersonId, Person]	Dict mapping person_id -> Person for successfully fetched persons.

Raises:

Type	Description
AffinityError	If on_error="raise" and any fetch fails.

Source code in affinity/services/persons.py

async def batch_get(
    self,
    person_ids: Sequence[PersonId],
    *,
    field_ids: Sequence[AnyFieldId] | None = None,
    field_types: Sequence[FieldType] | None = None,
    include_field_values: bool = False,
    with_interaction_dates: bool = False,
    with_interaction_persons: bool = False,
    max_concurrent: int = 10,
    on_error: Literal["raise", "skip"] = "raise",
) -> dict[PersonId, Person]:
    """
    Fetch multiple persons with controlled concurrency.

    Makes individual get() calls with bounded concurrency: one call per id. With
    with_interaction_dates and field_ids/field_types, the fields come from one call per
    100 ids and the dates from one V1 call per id (about N + N/100 calls, not 2N).

    Args:
        person_ids: Person IDs to fetch
        field_ids: Specific field IDs to include
        field_types: Field types to include
        include_field_values: If True, fetch embedded field values
        with_interaction_dates: Include interaction date summaries
        with_interaction_persons: Include person IDs for interactions
        max_concurrent: Maximum concurrent API calls (default: 10)
        on_error: How to handle AffinityError exceptions:
            - "raise": Raise on first AffinityError (default)
            - "skip": Skip failed IDs, return partial results

    Returns:
        Dict mapping person_id -> Person for successfully fetched persons.

    Raises:
        AffinityError: If on_error="raise" and any fetch fails.
    """
    if not person_ids:
        return {}
    if max_concurrent < 1:
        raise ValueError("max_concurrent must be at least 1")

    unique_ids = list(dict.fromkeys(person_ids))
    if (
        with_interaction_dates
        and (field_ids is not None or field_types is not None)
        and not include_field_values
    ):
        # One V2 call per 100 ids for the fields, one V1 call per id for the dates
        v1_params: dict[str, Any] = {"with_interaction_dates": True}
        if with_interaction_persons:
            v1_params["with_interaction_persons"] = True
        v2_params: dict[str, Any] = {}
        if field_ids:
            v2_params["fieldIds"] = [str(fid) for fid in field_ids]
        if field_types:
            v2_params["fieldTypes"] = [ft.value for ft in field_types]
        return await batch_get_with_dates(
            self._client,
            unique_ids,
            v1_path="/persons",
            v2_path="/persons",
            kind="Person",
            v1_params=v1_params,
            v2_params=v2_params,
            max_concurrent=max_concurrent,
            on_error=on_error,
            get_one=lambda record_id: self.get(
                record_id,
                field_ids=field_ids,
                field_types=field_types,
                with_interaction_dates=True,
                with_interaction_persons=with_interaction_persons,
            ),
            build=Person.model_validate,
        )
    results: dict[PersonId, Person] = {}

    async def fetch_one(pid: PersonId) -> tuple[PersonId, Person | None]:
        try:
            person = await self.get(
                pid,
                field_ids=field_ids,
                field_types=field_types,
                include_field_values=include_field_values,
                with_interaction_dates=with_interaction_dates,
                with_interaction_persons=with_interaction_persons,
            )
            return (pid, person)
        except AffinityError:
            if on_error == "raise":
                raise
            return (pid, None)

    for i in range(0, len(unique_ids), max_concurrent):
        chunk = unique_ids[i : i + max_concurrent]
        tasks = [asyncio.create_task(fetch_one(pid)) for pid in chunk]
        try:
            for coro in asyncio.as_completed(tasks):
                pid, person = await coro
                if person is not None:
                    results[pid] = person
        except BaseException:
            for task in tasks:
                if not task.done():
                    task.cancel()
            await asyncio.gather(*tasks, return_exceptions=True)
            raise

    return results

batch_update_fields(person_id: PersonId, updates: Mapping[AnyFieldId | str, Any], *, value_types: Mapping[AnyFieldId | str, FieldValueType | str] | None = None) -> BatchOperationResponse async

Update several field values of one person in one request.

Affinity applies the whole batch or none of it; each update replaces the field's value and None (with a type) clears it. Global, enriched and Source of Introduction fields are all writable. At most 100 updates, and 100 values per multi-value field.

Needs Affinity API version {FIELD_WRITES_MIN_API_VERSION} (out of beta there): an unpinned client sends that version for this call; a client pinned to an older version raises ApiVersionTooOldError without sending anything.

Parameters:

Name	Type	Description	Default
person_id	PersonId	The person.	required
updates	Mapping[AnyFieldId | str, Any]	Field id -> new value. Values are auto-typed (str→text, int/float→number, datetime/date→datetime) unless the field has an entry in value_types.	required
value_types	Mapping[AnyFieldId | str, FieldValueType | str] | None	Field id -> value type (e.g. "dropdown-multi"); needed for lists (multi-value fields) and None (clear).	None

Source code in affinity/services/persons.py

async def batch_update_fields(
    self,
    person_id: PersonId,
    updates: Mapping[AnyFieldId | str, Any],
    *,
    value_types: Mapping[AnyFieldId | str, FieldValueType | str] | None = None,
) -> BatchOperationResponse:
    """
    Update several field values of one person in one request.

    Affinity applies the whole batch or none of it; each update replaces the field's value
    and ``None`` (with a type) clears it. Global, enriched and Source of Introduction
    fields are all writable. At most 100 updates, and 100 values per multi-value field.

    Needs Affinity API version {FIELD_WRITES_MIN_API_VERSION} (out of beta there): an
    unpinned client sends that version for this call; a client pinned to an older version
    raises ``ApiVersionTooOldError`` without sending anything.

    Args:
        person_id: The person.
        updates: Field id -> new value. Values are auto-typed (str→text, int/float→number,
            datetime/date→datetime) unless the field has an entry in ``value_types``.
        value_types: Field id -> value type (e.g. ``"dropdown-multi"``); needed for lists
            (multi-value fields) and ``None`` (clear).
    """
    items = _batch_update_items(updates, value_types)
    result = await self._client.patch(
        f"/persons/{person_id}/fields",
        json={"operation": "update-fields", "updates": items},
        min_api_version=FIELD_WRITES_MIN_API_VERSION,
    )
    return BatchOperationResponse.model_validate(result)

create(data: PersonCreate, *, if_not_exists: bool = True) -> Person async

Create a new person.

Parameters:

Name	Type	Description	Default
data	PersonCreate	Person creation data	required
if_not_exists	bool	If True (default), check for an existing person with the same primary email (case-insensitive exact) — falling back to exact first+last name match when no email is provided. Uses V1 search_pages(term) with page_size=500, capped at _DEDUP_PAGE_CAP pages. Set to False to skip the check and create unconditionally.	True

Returns:

Type	Description
Person	Created person

Raises:

Type	Description
DuplicateEntityError	When if_not_exists=True and an exact-email OR exact full-name match already exists.

Note

Creates use V1 API, while reads use V2 API. Due to eventual consistency between V1 and V2, a get() call immediately after create() may return 404 NotFoundError. If you need to read immediately after creation, either: - Use the Person object returned by this method (it contains the created data) - Add a short delay (100-500ms) before calling get() - Implement retry logic in your application

Source code in affinity/services/persons.py

async def create(self, data: PersonCreate, *, if_not_exists: bool = True) -> Person:
    """
    Create a new person.

    Args:
        data: Person creation data
        if_not_exists: If True (default), check for an existing person with
            the same primary email (case-insensitive exact) — falling back
            to exact first+last name match when no email is provided.
            Uses V1 search_pages(term) with page_size=500, capped at
            _DEDUP_PAGE_CAP pages. Set to False to skip the check and
            create unconditionally.

    Returns:
        Created person

    Raises:
        DuplicateEntityError: When if_not_exists=True and an exact-email
            OR exact full-name match already exists.

    Note:
        Creates use V1 API, while reads use V2 API. Due to eventual consistency
        between V1 and V2, a `get()` call immediately after `create()` may return
        404 NotFoundError. If you need to read immediately after creation, either:
        - Use the Person object returned by this method (it contains the created data)
        - Add a short delay (100-500ms) before calling get()
        - Implement retry logic in your application
    """
    if if_not_exists:
        existing = await self._find_exact_duplicate(
            first_name=data.first_name,
            last_name=data.last_name,
            emails=list(data.emails or []),
        )
        if existing is not None:
            raise DuplicateEntityError(
                f"Person with email or name already exists (id={existing.id})",
                entity_type="person",
                existing_id=int(existing.id),
                existing_name=f"{existing.first_name or ''} {existing.last_name or ''}".strip(),
            )

    payload = data.model_dump(by_alias=True, mode="json")
    if not data.company_ids:
        payload.pop("organization_ids", None)

    result = await self._client.post("/persons", json=payload, v1=True)

    if self._client.cache:
        self._client.cache.invalidate_prefix("person")

    return Person.model_validate(result)

delete(person_id: PersonId) -> bool async

Delete a person (also deletes associated field values).

Source code in affinity/services/persons.py

async def delete(self, person_id: PersonId) -> bool:
    """Delete a person (also deletes associated field values)."""
    result = await self._client.delete(f"/persons/{person_id}", v1=True)
    invalidate_all_list_sizes(self._client)

    if self._client.cache:
        self._client.cache.invalidate_prefix("person")

    return bool(result.get("success", True))

get(person_id: PersonId, *, field_ids: Sequence[AnyFieldId] | None = None, field_types: Sequence[FieldType] | None = None, include_field_values: bool = False, retries: int = 0, with_interaction_dates: bool = False, with_interaction_persons: bool = False) -> Person async

Get a single person by ID.

Parameters:

Name	Type	Description	Default
person_id	PersonId	The person ID	required
field_ids	Sequence[AnyFieldId] | None	Specific field IDs to include in response	None
field_types	Sequence[FieldType] | None	Field types to include (e.g., ["enriched", "global"])	None
include_field_values	bool	If True, fetch embedded field values. This saves one API call when you need both person info and field values. Cannot be combined with field_ids/field_types.	False
retries	int	Number of retries on 404 NotFoundError. Default is 0 (fail fast). Set to 2-3 if calling immediately after create() to handle eventual consistency lag.	0
with_interaction_dates	bool	Include interaction date summaries (last/next meeting dates, email dates).	False
with_interaction_persons	bool	Include person IDs for each interaction. Only applies when with_interaction_dates=True.	False

Returns:

Type	Description
Person	Person object with requested field data.
Person	When include_field_values=True, the Person will have a field_values
Person	attribute containing the list of FieldValue objects.
Person	When with_interaction_dates=True, the Person will have interaction_dates
Person	and interactions populated.

Raises:

Type	Description
NotFoundError	If person does not exist after all retries.
ValueError	If include_field_values is combined with field_ids/field_types.

Note

When combining with_interaction_dates with field_ids/field_types, two API calls are made internally and the results are merged.

Source code in affinity/services/persons.py

async def get(
    self,
    person_id: PersonId,
    *,
    field_ids: Sequence[AnyFieldId] | None = None,
    field_types: Sequence[FieldType] | None = None,
    include_field_values: bool = False,
    retries: int = 0,
    with_interaction_dates: bool = False,
    with_interaction_persons: bool = False,
) -> Person:
    """
    Get a single person by ID.

    Args:
        person_id: The person ID
        field_ids: Specific field IDs to include in response
        field_types: Field types to include (e.g., ["enriched", "global"])
        include_field_values: If True, fetch embedded field values. This saves
            one API call when you need both person info and field values.
            Cannot be combined with field_ids/field_types.
        retries: Number of retries on 404 NotFoundError. Default is 0 (fail fast).
            Set to 2-3 if calling immediately after create() to handle eventual
            consistency lag.
        with_interaction_dates: Include interaction date summaries (last/next
            meeting dates, email dates).
        with_interaction_persons: Include person IDs for each interaction.
            Only applies when with_interaction_dates=True.

    Returns:
        Person object with requested field data.
        When include_field_values=True, the Person will have a `field_values`
        attribute containing the list of FieldValue objects.
        When with_interaction_dates=True, the Person will have interaction_dates
        and interactions populated.

    Raises:
        NotFoundError: If person does not exist after all retries.
        ValueError: If include_field_values is combined with field_ids/field_types.

    Note:
        When combining with_interaction_dates with field_ids/field_types,
        two API calls are made internally and the results are merged.
    """
    return await self._get_with_retry(
        person_id,
        field_ids=field_ids,
        field_types=field_types,
        include_field_values=include_field_values,
        retries=retries,
        with_interaction_dates=with_interaction_dates,
        with_interaction_persons=with_interaction_persons,
    )

get_associated_company_ids(person_id: PersonId, *, max_results: int | None = None) -> builtins.list[CompanyId] async

Get associated company IDs for a person.

V1-only: V2 does not expose person -> company associations directly. Uses GET /persons/{id} (V1) and returns organization_ids.

Parameters:

Name	Type	Description	Default
person_id	PersonId	The person ID	required
max_results	int | None	Maximum number of company IDs to return	None

Returns:

Type	Description
list[CompanyId]	List of CompanyId values associated with this person

Note

The Person model already has company_ids populated from V1's organizationIds field. This method provides API parity with CompanyService.get_associated_person_ids().

Source code in affinity/services/persons.py

async def get_associated_company_ids(
    self,
    person_id: PersonId,
    *,
    max_results: int | None = None,
) -> builtins.list[CompanyId]:
    """
    Get associated company IDs for a person.

    V1-only: V2 does not expose person -> company associations directly.
    Uses GET `/persons/{id}` (V1) and returns `organization_ids`.

    Args:
        person_id: The person ID
        max_results: Maximum number of company IDs to return

    Returns:
        List of CompanyId values associated with this person

    Note:
        The Person model already has `company_ids` populated from V1's
        `organizationIds` field. This method provides API parity with
        `CompanyService.get_associated_person_ids()`.
    """
    data = await self._client.get(f"/persons/{person_id}", v1=True)
    # Defensive: handle potential {"person": {...}} wrapper
    # (consistent with CompanyService.get_associated_person_ids pattern)
    person = data.get("person") if isinstance(data, dict) else None
    source = person if isinstance(person, dict) else data
    org_ids = None
    if isinstance(source, dict):
        org_ids = source.get("organization_ids") or source.get("organizationIds")

    if not isinstance(org_ids, list):
        return []

    ids = [CompanyId(int(cid)) for cid in org_ids if cid is not None]
    if max_results is not None and max_results >= 0:
        return ids[:max_results]
    return ids

get_associated_company_ids_batch(person_ids: Sequence[PersonId], *, on_error: Literal['raise', 'skip'] = 'raise') -> dict[PersonId, builtins.list[CompanyId]] async

Get company associations for multiple persons.

Makes one V1 API call per person.

Parameters:

Name	Type	Description	Default
person_ids	Sequence[PersonId]	Sequence of person IDs to fetch	required
on_error	Literal['raise', 'skip']	How to handle errors - "raise" (default) or "skip" failed IDs	'raise'

Returns:

Type	Description
dict[PersonId, list[CompanyId]]	Dict mapping person_id -> list of company_ids

Raises:

Type	Description
AffinityError	If on_error="raise" and any fetch fails.

Source code in affinity/services/persons.py

async def get_associated_company_ids_batch(
    self,
    person_ids: Sequence[PersonId],
    *,
    on_error: Literal["raise", "skip"] = "raise",
) -> dict[PersonId, builtins.list[CompanyId]]:
    """
    Get company associations for multiple persons.

    Makes one V1 API call per person.

    Args:
        person_ids: Sequence of person IDs to fetch
        on_error: How to handle errors - "raise" (default) or "skip" failed IDs

    Returns:
        Dict mapping person_id -> list of company_ids

    Raises:
        AffinityError: If on_error="raise" and any fetch fails.
    """
    result: dict[PersonId, builtins.list[CompanyId]] = {}
    for person_id in person_ids:
        try:
            result[person_id] = await self.get_associated_company_ids(person_id)
        except AffinityError:
            if on_error == "raise":
                raise
            # skip: continue without this person
        except Exception as e:
            if on_error == "raise":
                raise AffinityError(
                    f"Failed to get associations for person {person_id}: {e}"
                ) from e
            # skip: continue without this person
    return result

get_associated_opportunity_ids(person_id: PersonId, *, max_results: int | None = None) -> builtins.list[OpportunityId] async

Get associated opportunity IDs for a person.

V1-only: V2 does not expose person -> opportunity associations directly. Uses GET /persons/{id} (V1) and returns opportunity_ids.

Parameters:

Name	Type	Description	Default
person_id	PersonId	The person ID	required
max_results	int | None	Maximum number of opportunity IDs to return	None

Returns:

Type	Description
list[OpportunityId]	List of OpportunityId values associated with this person

Note

The Person model already has opportunity_ids populated from V1's opportunityIds field. This method provides API parity with OpportunityService.get_associated_person_ids().

Source code in affinity/services/persons.py

async def get_associated_opportunity_ids(
    self,
    person_id: PersonId,
    *,
    max_results: int | None = None,
) -> builtins.list[OpportunityId]:
    """
    Get associated opportunity IDs for a person.

    V1-only: V2 does not expose person -> opportunity associations directly.
    Uses GET `/persons/{id}` (V1) and returns `opportunity_ids`.

    Args:
        person_id: The person ID
        max_results: Maximum number of opportunity IDs to return

    Returns:
        List of OpportunityId values associated with this person

    Note:
        The Person model already has `opportunity_ids` populated from V1's
        `opportunityIds` field. This method provides API parity with
        `OpportunityService.get_associated_person_ids()`.
    """
    data = await self._client.get(f"/persons/{person_id}", v1=True)
    # Defensive: handle potential {"person": {...}} wrapper
    person = data.get("person") if isinstance(data, dict) else None
    source = person if isinstance(person, dict) else data
    opp_ids = None
    if isinstance(source, dict):
        opp_ids = source.get("opportunity_ids") or source.get("opportunityIds")

    if not isinstance(opp_ids, list):
        return []

    ids = [OpportunityId(int(oid)) for oid in opp_ids if oid is not None]
    if max_results is not None and max_results >= 0:
        return ids[:max_results]
    return ids

get_associated_opportunity_ids_batch(person_ids: Sequence[PersonId], *, on_error: Literal['raise', 'skip'] = 'raise') -> dict[PersonId, builtins.list[OpportunityId]] async

Get opportunity associations for multiple persons.

Makes one V1 API call per person.

Parameters:

Name	Type	Description	Default
person_ids	Sequence[PersonId]	Sequence of person IDs to fetch	required
on_error	Literal['raise', 'skip']	How to handle errors - "raise" (default) or "skip" failed IDs	'raise'

Returns:

Type	Description
dict[PersonId, list[OpportunityId]]	Dict mapping person_id -> list of opportunity_ids

Raises:

Type	Description
AffinityError	If on_error="raise" and any fetch fails.

Source code in affinity/services/persons.py

async def get_associated_opportunity_ids_batch(
    self,
    person_ids: Sequence[PersonId],
    *,
    on_error: Literal["raise", "skip"] = "raise",
) -> dict[PersonId, builtins.list[OpportunityId]]:
    """
    Get opportunity associations for multiple persons.

    Makes one V1 API call per person.

    Args:
        person_ids: Sequence of person IDs to fetch
        on_error: How to handle errors - "raise" (default) or "skip" failed IDs

    Returns:
        Dict mapping person_id -> list of opportunity_ids

    Raises:
        AffinityError: If on_error="raise" and any fetch fails.
    """
    result: dict[PersonId, builtins.list[OpportunityId]] = {}
    for person_id in person_ids:
        try:
            result[person_id] = await self.get_associated_opportunity_ids(person_id)
        except AffinityError:
            if on_error == "raise":
                raise
            # skip: continue without this person
        except Exception as e:
            if on_error == "raise":
                raise AffinityError(
                    f"Failed to get associations for person {person_id}: {e}"
                ) from e
            # skip: continue without this person
    return result

get_field_dropdown_options(field_id: AnyFieldId, *, with_status_types: bool = False) -> builtins.list[DropdownOption] async

All options of a person dropdown, ranked-dropdown or status field (every page).

Options with type, rank, color and (status fields) status_category / win_rate. Options of global person fields can only be read; list-specific fields' options are managed through client.lists. with_status_types=True: see lists.get_field_dropdown_options.

Source code in affinity/services/persons.py

async def get_field_dropdown_options(
    self, field_id: AnyFieldId, *, with_status_types: bool = False
) -> builtins.list[DropdownOption]:
    """All options of a person dropdown, ranked-dropdown or status field (every page).

    Options with ``type``, ``rank``, ``color`` and (status fields) ``status_category`` /
    ``win_rate``. Options of global person fields can only be read; list-specific fields'
    options are managed through ``client.lists``. ``with_status_types=True``: see
    ``lists.get_field_dropdown_options``.
    """
    data = await self._client.get_all_pages(
        options_path(field_id, list_id=None, entity="persons"),
        params={"limit": 100},
        min_api_version=DROPDOWN_OPTION_WRITES_MIN_API_VERSION if with_status_types else None,
    )
    return parse_options(data)

get_field_values(person_id: PersonId, *, ids: Sequence[AnyFieldId | str] | None = None) -> FieldValues async

Field values of one person, keyed by field id (GET /v2/persons/{id}/fields).

Parameters:

Name	Type	Description	Default
person_id	PersonId	The person.	required
ids	Sequence[AnyFieldId | str] | None	Only these fields (server-side filter); all fields when omitted.	None

Source code in affinity/services/persons.py

async def get_field_values(
    self,
    person_id: PersonId,
    *,
    ids: Sequence[AnyFieldId | str] | None = None,
) -> FieldValues:
    """
    Field values of one person, keyed by field id (``GET /v2/persons/{id}/fields``).

    Args:
        person_id: The person.
        ids: Only these fields (server-side filter); all fields when omitted.
    """
    params = {"ids": [str(fid) for fid in ids]} if ids else None
    data = await self._client.get_all_pages(f"/persons/{person_id}/fields", params=params)
    return FieldValues.model_validate(data.get("data", []))

get_fields(*, filter: str | None = None, includes: Sequence[str] | None = None) -> builtins.list[FieldMetadata] async

Get metadata about person fields.

Cached for performance.

Parameters:

Name	Type	Description	Default
filter	str | None	Filter on the field name: name="Location" (exact) or name=~Loc (substring); both case-sensitive. Combine clauses with |. Only name is filterable; anything else is rejected with 400.	None
includes	Sequence[str] | None	Extra properties per field: "filterability", "sortability" (read them via is_filterable / is_sortable).	None

Source code in affinity/services/persons.py

async def get_fields(
    self,
    *,
    filter: str | None = None,
    includes: Sequence[str] | None = None,
) -> builtins.list[FieldMetadata]:
    """
    Get metadata about person fields.

    Cached for performance.

    Args:
        filter: Filter on the field name: ``name="Location"`` (exact) or ``name=~Loc``
            (substring); both case-sensitive. Combine clauses with ``|``. Only ``name``
            is filterable; anything else is rejected with 400.
        includes: Extra properties per field: ``"filterability"``, ``"sortability"``
            (read them via ``is_filterable`` / ``is_sortable``).
    """
    params, cache_key = build_fields_query(
        "person_fields:",
        filter=filter,
        includes=includes,
    )

    data = await self._client.get_all_pages(
        "/persons/fields",
        params=params or None,
        cache_key=cache_key,
        cache_ttl=300,
    )

    return [FieldMetadata.model_validate(f) for f in data.get("data", [])]

get_first(*, field_ids: Sequence[AnyFieldId] | None = None, field_types: Sequence[FieldType] | None = None, **_unsupported: Any) -> Person | None async

Get the first person, or None if no results.

See PersonService.get_first() for details.

Note

The V2 /persons endpoint silently ignores any filter= parameter — passing one would return unfiltered results without warning. This method therefore rejects filter= with a ValueError. To search persons by name or email use client.persons.search_pages(term). To filter by list-specific fields use client.lists.entries(list_id).list(filter=...).

Source code in affinity/services/persons.py

async def get_first(
    self,
    *,
    field_ids: Sequence[AnyFieldId] | None = None,
    field_types: Sequence[FieldType] | None = None,
    **_unsupported: Any,
) -> Person | None:
    """
    Get the first person, or None if no results.

    See PersonService.get_first() for details.

    Note:
        The V2 /persons endpoint silently ignores any ``filter=`` parameter —
        passing one would return unfiltered results without warning. This method
        therefore rejects ``filter=`` with a ``ValueError``. To search persons
        by name or email use ``client.persons.search_pages(term)``. To filter
        by list-specific fields use ``client.lists.entries(list_id).list(filter=...)``.
    """
    if "filter" in _unsupported:
        raise ValueError(
            "V2 /persons does not support server-side filter. "
            "To search by name/email use client.persons.search_pages(term). "
            "To filter by list-specific fields use "
            "client.lists.entries(list_id).list(filter=...)."
        )
    if _unsupported:
        raise TypeError(f"Unexpected keyword arguments: {list(_unsupported)}")
    page = await self.list(
        field_ids=field_ids,
        field_types=field_types,
        limit=1,
    )
    return page.data[0] if page.data else None

get_list_entries(person_id: PersonId, *, limit: int | None = None, cursor: str | None = None) -> PaginatedResponse[ListEntry] async

One page of a person's list entries across all lists.

Each entry's entity_id is person_id and its entity_type is EntityType.PERSON; entity is None (this endpoint doesn't embed the person). Follow pagination.next_cursor with cursor= (from this person's entries) for the next page.

Source code in affinity/services/persons.py

async def get_list_entries(
    self,
    person_id: PersonId,
    *,
    limit: int | None = None,
    cursor: str | None = None,
) -> PaginatedResponse[ListEntry]:
    """One page of a person's list entries across all lists.

    Each entry's ``entity_id`` is ``person_id`` and its ``entity_type`` is
    ``EntityType.PERSON``; ``entity`` is ``None`` (this endpoint doesn't embed the person).
    Follow ``pagination.next_cursor`` with ``cursor=`` (from this person's entries) for
    the next page.
    """
    path = f"/persons/{person_id}/list-entries"
    if cursor is not None:
        if limit is not None:
            raise ValueError(
                "Cannot combine 'cursor' with other parameters; cursor encodes all query "
                "context. Start a new pagination sequence without a cursor to change "
                "parameters."
            )
        check_list_entries_cursor(cursor, path)
        data = await self._client.get_url(cursor)
    else:
        check_positive_limit(limit)
        data = await self._client.get(
            path, params={"limit": limit} if limit is not None else None
        )

    return entity_list_entries_page(data, entity_id=person_id, entity_type=EntityType.PERSON)

get_lists(person_id: PersonId) -> PaginatedResponse[ListSummary] async

Get all lists that contain this person.

Source code in affinity/services/persons.py

async def get_lists(
    self,
    person_id: PersonId,
) -> PaginatedResponse[ListSummary]:
    """Get all lists that contain this person."""
    data = await self._client.get(f"/persons/{person_id}/lists")

    return PaginatedResponse[ListSummary](
        data=[ListSummary.model_validate(item) for item in data.get("data", [])],
        pagination=PaginationInfo.model_validate(data.get("pagination", {})),
    )

get_merge_state(merge_id: int) -> PersonMergeState async

One past person merge (see :meth:list_merges).

Source code in affinity/services/persons.py

async def get_merge_state(self, merge_id: int) -> PersonMergeState:
    """One past person merge (see :meth:`list_merges`)."""
    return PersonMergeState.model_validate(
        await self._client.get(f"/person-merges/{int(merge_id)}")
    )

get_merge_status(task_id: str) -> MergeTask async

Check the status of a merge operation.

Source code in affinity/services/persons.py

async def get_merge_status(self, task_id: str) -> MergeTask:
    """Check the status of a merge operation."""
    data = await self._client.get(f"/tasks/person-merges/{task_id}")
    return MergeTask.model_validate(data)

iter(*, ids: Sequence[PersonId] | None = None, field_ids: Sequence[AnyFieldId] | None = None, field_types: Sequence[FieldType] | None = None, **_unsupported: Any) -> AsyncIterator[Person]

Auto-paginate all persons.

Alias for all() (FR-006 public contract).

Note

The V2 /persons endpoint silently ignores any filter= parameter — passing one would return unfiltered results without warning. This method therefore rejects filter= with a ValueError. To search persons by name or email use client.persons.search_pages(term). To filter by list-specific fields use client.lists.entries(list_id).list(filter=...).

Source code in affinity/services/persons.py

def iter(
    self,
    *,
    ids: Sequence[PersonId] | None = None,
    field_ids: Sequence[AnyFieldId] | None = None,
    field_types: Sequence[FieldType] | None = None,
    **_unsupported: Any,
) -> AsyncIterator[Person]:
    """
    Auto-paginate all persons.

    Alias for `all()` (FR-006 public contract).

    Note:
        The V2 /persons endpoint silently ignores any ``filter=`` parameter —
        passing one would return unfiltered results without warning. This method
        therefore rejects ``filter=`` with a ``ValueError``. To search persons
        by name or email use ``client.persons.search_pages(term)``. To filter
        by list-specific fields use ``client.lists.entries(list_id).list(filter=...)``.
    """
    if "filter" in _unsupported:
        raise ValueError(
            "V2 /persons does not support server-side filter. "
            "To search by name/email use client.persons.search_pages(term). "
            "To filter by list-specific fields use "
            "client.lists.entries(list_id).list(filter=...)."
        )
    if _unsupported:
        raise TypeError(f"Unexpected keyword arguments: {list(_unsupported)}")
    return self.all(ids=ids, field_ids=field_ids, field_types=field_types)

iter_merges(*, status: str | None = None, task_id: str | None = None) -> AsyncIterator[PersonMergeState]

All past person merges matching the filters (every page).

Source code in affinity/services/persons.py

def iter_merges(
    self, *, status: str | None = None, task_id: str | None = None
) -> AsyncIterator[PersonMergeState]:
    """All past person merges matching the filters (every page)."""

    async def fetch_page(next_url: str | None) -> PaginatedResponse[PersonMergeState]:
        if next_url:
            return await self.list_merges(cursor=next_url)
        return await self.list_merges(status=status, task_id=task_id)

    return AsyncPageIterator(fetch_page)

iter_notes(person_id: PersonId, *, creator_id: int | None = None, created_after: datetime | None = None, created_before: datetime | None = None, updated_after: datetime | None = None, limit: int | None = None) -> AsyncIterator[NoteV2]

All of this person's notes (every page); see :meth:list_notes.

Source code in affinity/services/persons.py

def iter_notes(
    self,
    person_id: PersonId,
    *,
    creator_id: int | None = None,
    created_after: datetime | None = None,
    created_before: datetime | None = None,
    updated_after: datetime | None = None,
    limit: int | None = None,
) -> AsyncIterator[NoteV2]:
    """All of this person's notes (every page); see :meth:`list_notes`."""

    async def fetch_page(next_url: str | None) -> PaginatedResponse[NoteV2]:
        if next_url:
            return await self.list_notes(person_id, cursor=next_url)
        return await self.list_notes(
            person_id,
            creator_id=creator_id,
            created_after=created_after,
            created_before=created_before,
            updated_after=updated_after,
            limit=limit,
        )

    return AsyncPageIterator(fetch_page)

iter_relationships(person_id: PersonId, *, min_score: float | None = None, order: Literal['asc', 'desc'] = 'desc', limit: int | None = None) -> AsyncIterator[Relationship]

All relationships for this person (every page); see :meth:list_relationships.

Source code in affinity/services/persons.py

def iter_relationships(
    self,
    person_id: PersonId,
    *,
    min_score: float | None = None,
    order: Literal["asc", "desc"] = "desc",
    limit: int | None = None,
) -> AsyncIterator[Relationship]:
    """All relationships for this person (every page); see :meth:`list_relationships`."""

    async def fetch_page(next_url: str | None) -> PaginatedResponse[Relationship]:
        if next_url:
            return await self.list_relationships(person_id, cursor=next_url)
        return await self.list_relationships(
            person_id, min_score=min_score, order=order, limit=limit
        )

    return AsyncPageIterator(fetch_page)

list(*, ids: Sequence[PersonId] | None = None, field_ids: Sequence[AnyFieldId] | None = None, field_types: Sequence[FieldType] | None = None, limit: int | None = None, cursor: str | None = None, **_unsupported: Any) -> PaginatedResponse[Person] async

Get a page of persons.

Parameters:

Name	Type	Description	Default
ids	Sequence[PersonId] | None	Specific person IDs to fetch (batch lookup)	None
field_ids	Sequence[AnyFieldId] | None	Specific field IDs to include in response	None
field_types	Sequence[FieldType] | None	Field types to include	None
limit	int | None	Maximum number of results	None
cursor	str | None	Cursor to resume pagination (opaque; obtained from prior responses)	None

Returns:

Type	Description
PaginatedResponse[Person]	Paginated response with persons

Note

The V2 /persons endpoint silently ignores any filter= parameter — passing one would return unfiltered results without warning. This method therefore rejects filter= with a ValueError. To search persons by name or email use client.persons.search_pages(term). To filter by list-specific fields use client.lists.entries(list_id).list(filter=...).

Source code in affinity/services/persons.py

async def list(
    self,
    *,
    ids: Sequence[PersonId] | None = None,
    field_ids: Sequence[AnyFieldId] | None = None,
    field_types: Sequence[FieldType] | None = None,
    limit: int | None = None,
    cursor: str | None = None,
    **_unsupported: Any,
) -> PaginatedResponse[Person]:
    """
    Get a page of persons.

    Args:
        ids: Specific person IDs to fetch (batch lookup)
        field_ids: Specific field IDs to include in response
        field_types: Field types to include
        limit: Maximum number of results
        cursor: Cursor to resume pagination (opaque; obtained from prior responses)

    Returns:
        Paginated response with persons

    Note:
        The V2 /persons endpoint silently ignores any ``filter=`` parameter —
        passing one would return unfiltered results without warning. This method
        therefore rejects ``filter=`` with a ``ValueError``. To search persons
        by name or email use ``client.persons.search_pages(term)``. To filter
        by list-specific fields use ``client.lists.entries(list_id).list(filter=...)``.
    """
    if "filter" in _unsupported:
        raise ValueError(
            "V2 /persons does not support server-side filter. "
            "To search by name/email use client.persons.search_pages(term). "
            "To filter by list-specific fields use "
            "client.lists.entries(list_id).list(filter=...)."
        )
    if _unsupported:
        raise TypeError(f"Unexpected keyword arguments: {list(_unsupported)}")
    validate_entity_field_types(field_types, endpoint="person")
    if cursor is not None:
        if any(p is not None for p in (ids, field_ids, field_types, limit)):
            raise ValueError(
                "Cannot combine 'cursor' with other parameters; cursor encodes all query "
                "context. Start a new pagination sequence without a cursor to change "
                "parameters."
            )
        data = await self._client.get_url(cursor)
    else:
        check_positive_limit(limit)
        ids = materialize_ids(ids)
        if no_ids(ids):
            return PaginatedResponse[Person]()
        params: dict[str, Any] = {}
        if ids:
            params["ids"] = [int(id_) for id_ in ids]
        if field_ids:
            params["fieldIds"] = [str(field_id) for field_id in field_ids]
        if field_types:
            params["fieldTypes"] = [field_type.value for field_type in field_types]
        if limit is not None:
            params["limit"] = limit
        data = await self._client.get("/persons", params=params or None)

    return PaginatedResponse[Person](
        data=[Person.model_validate(p) for p in data.get("data", [])],
        pagination=PaginationInfo.model_validate(data.get("pagination", {})),
    )

list_merges(*, status: str | None = None, task_id: str | None = None, limit: int | None = None, cursor: str | None = None) -> PaginatedResponse[PersonMergeState] async

One page of past person merges, newest first.

status is "in-progress", "success" or "failed"; task_id (or the task URL a merge returned) limits it to one merge task. Needs the "Manage duplicates" permission and the organization admin role (otherwise AuthorizationError).

Source code in affinity/services/persons.py

async def list_merges(
    self,
    *,
    status: str | None = None,
    task_id: str | None = None,
    limit: int | None = None,
    cursor: str | None = None,
) -> PaginatedResponse[PersonMergeState]:
    """One page of past person merges, newest first.

    ``status`` is ``"in-progress"``, ``"success"`` or ``"failed"``; ``task_id`` (or the task
    URL a merge returned) limits it to one merge task. Needs the "Manage duplicates"
    permission and the organization admin role (otherwise ``AuthorizationError``).
    """
    check_cursor_alone(cursor, status, task_id, limit)
    if cursor is not None:
        return page_of(PersonMergeState, await self._client.get_url(cursor))
    params = merge_params(status=status, task_id=task_id, limit=limit)
    return page_of(
        PersonMergeState, await self._client.get("/person-merges", params=params or None)
    )

list_notes(person_id: PersonId, *, creator_id: int | None = None, created_after: datetime | None = None, created_before: datetime | None = None, updated_after: datetime | None = None, limit: int | None = None, total_count: bool = False, cursor: str | None = None) -> PaginatedResponse[NoteV2] async

One page of this person's notes as V2 notes (NoteV2; replies not included).

Filters: creator_id, created_after (inclusive) / created_before (exclusive), updated_after, rounded outward to whole seconds. limit is 1-100.

Source code in affinity/services/persons.py

async def list_notes(
    self,
    person_id: PersonId,
    *,
    creator_id: int | None = None,
    created_after: datetime | None = None,
    created_before: datetime | None = None,
    updated_after: datetime | None = None,
    limit: int | None = None,
    total_count: bool = False,
    cursor: str | None = None,
) -> PaginatedResponse[NoteV2]:
    """One page of this person's notes as V2 notes (``NoteV2``; replies not included).

    Filters: ``creator_id``, ``created_after`` (inclusive) / ``created_before`` (exclusive),
    ``updated_after``, rounded outward to whole seconds. ``limit`` is 1-100.
    """
    check_cursor_alone(
        cursor, creator_id, created_after, created_before, updated_after, limit, total_count
    )
    if cursor is not None:
        return page_of(NoteV2, await self._client.get_url(cursor, min_api_version=None))
    params = note_params(
        creator_id=creator_id,
        created_after=created_after,
        created_before=created_before,
        updated_after=updated_after,
        limit=limit,
        total_count=total_count,
    )
    return page_of(
        NoteV2,
        await self._client.get(
            f"/persons/{int(person_id)}/notes",
            params=params or None,
            min_api_version=None,
        ),
    )

list_relationships(person_id: PersonId, *, min_score: float | None = None, order: Literal['asc', 'desc'] = 'desc', limit: int | None = None, total_count: bool = False, cursor: str | None = None) -> PaginatedResponse[Relationship] async

One page of the relationships between your team and this person's people.

Strongest first (order="asc" for weakest first). min_score (0.0-1.0) keeps relationships at least that strong; relationships known only from LinkedIn have score 0, so any min_score above 0 drops them. limit is 1-100 (API default 20). Needs Affinity API version {RELATIONSHIPS_MIN_API_VERSION} (sent for this call and every page).

Source code in affinity/services/persons.py

async def list_relationships(
    self,
    person_id: PersonId,
    *,
    min_score: float | None = None,
    order: Literal["asc", "desc"] = "desc",
    limit: int | None = None,
    total_count: bool = False,
    cursor: str | None = None,
) -> PaginatedResponse[Relationship]:
    """One page of the relationships between your team and this person's people.

    Strongest first (``order="asc"`` for weakest first). ``min_score`` (0.0-1.0) keeps
    relationships at least that strong; relationships known only from LinkedIn have score 0,
    so any ``min_score`` above 0 drops them. ``limit`` is 1-100 (API default 20). Needs
    Affinity API version {RELATIONSHIPS_MIN_API_VERSION} (sent for this call and every page).
    """
    if cursor is not None:
        if min_score is not None or limit is not None or total_count or order != "desc":
            raise ValueError(
                "Cannot combine 'cursor' with other parameters; cursor encodes all query "
                "context."
            )
        return page_of(
            Relationship,
            await self._client.get_url(cursor, min_api_version=RELATIONSHIPS_MIN_API_VERSION),
        )
    params = relationship_params(
        min_score=min_score, order=order, limit=limit, total_count=total_count
    )
    return page_of(
        Relationship,
        await self._client.get(
            f"/persons/{int(person_id)}/relationships",
            params=params or None,
            min_api_version=RELATIONSHIPS_MIN_API_VERSION,
        ),
    )

merge(primary_id: PersonId, duplicate_id: PersonId) -> str async

Merge a duplicate person into a primary person.

Returns a task URL to check merge status.

Source code in affinity/services/persons.py

async def merge(
    self,
    primary_id: PersonId,
    duplicate_id: PersonId,
) -> str:
    """
    Merge a duplicate person into a primary person.

    Returns a task URL to check merge status.
    """
    result = await self._client.post(
        "/person-merges",
        json={
            "primaryPersonId": int(primary_id),
            "duplicatePersonId": int(duplicate_id),
        },
    )
    return merge_task_url(result, "person")

pages(*, ids: Sequence[PersonId] | None = None, field_ids: Sequence[AnyFieldId] | None = None, field_types: Sequence[FieldType] | None = None, limit: int | None = None, cursor: str | None = None, **_unsupported: Any) -> AsyncIterator[PaginatedResponse[Person]]

Iterate person pages (not items), yielding PaginatedResponse[Person].

Useful for ETL scripts that need checkpoint/resume via page.next_cursor.

Parameters:

Name	Type	Description	Default
ids	Sequence[PersonId] | None	Specific person IDs to fetch (batch lookup)	None
field_ids	Sequence[AnyFieldId] | None	Specific field IDs to include in response	None
field_types	Sequence[FieldType] | None	Field types to include	None
limit	int | None	Maximum results per page	None
cursor	str | None	Cursor to resume pagination	None

Returns:

Type	Description
AsyncIterator[PaginatedResponse[Person]]	AsyncIterator of PaginatedResponse[Person]

Note

The V2 /persons endpoint silently ignores any filter= parameter — passing one would return unfiltered results without warning. This method therefore rejects filter= with a ValueError. To search persons by name or email use client.persons.search_pages(term). To filter by list-specific fields use client.lists.entries(list_id).list(filter=...).

Source code in affinity/services/persons.py

def pages(
    self,
    *,
    ids: Sequence[PersonId] | None = None,
    field_ids: Sequence[AnyFieldId] | None = None,
    field_types: Sequence[FieldType] | None = None,
    limit: int | None = None,
    cursor: str | None = None,
    **_unsupported: Any,
) -> AsyncIterator[PaginatedResponse[Person]]:
    """
    Iterate person pages (not items), yielding `PaginatedResponse[Person]`.

    Useful for ETL scripts that need checkpoint/resume via `page.next_cursor`.

    Args:
        ids: Specific person IDs to fetch (batch lookup)
        field_ids: Specific field IDs to include in response
        field_types: Field types to include
        limit: Maximum results per page
        cursor: Cursor to resume pagination

    Returns:
        AsyncIterator of PaginatedResponse[Person]

    Note:
        The V2 /persons endpoint silently ignores any ``filter=`` parameter —
        passing one would return unfiltered results without warning. This method
        therefore rejects ``filter=`` with a ``ValueError``. To search persons
        by name or email use ``client.persons.search_pages(term)``. To filter
        by list-specific fields use ``client.lists.entries(list_id).list(filter=...)``.
    """
    if "filter" in _unsupported:
        raise ValueError(
            "V2 /persons does not support server-side filter. "
            "To search by name/email use client.persons.search_pages(term). "
            "To filter by list-specific fields use "
            "client.lists.entries(list_id).list(filter=...)."
        )
    if _unsupported:
        raise TypeError(f"Unexpected keyword arguments: {list(_unsupported)}")
    return self._pages_iter(
        ids=ids,
        field_ids=field_ids,
        field_types=field_types,
        limit=limit,
        cursor=cursor,
    )

resolve(*, email: str | None = None, name: str | None = None) -> Person | None async

Find a single person by email or name.

This is a convenience helper that searches and returns the first exact match, or None if not found. Uses V1 search internally.

Source code in affinity/services/persons.py

async def resolve(
    self,
    *,
    email: str | None = None,
    name: str | None = None,
) -> Person | None:
    """
    Find a single person by email or name.

    This is a convenience helper that searches and returns the first exact match,
    or None if not found. Uses V1 search internally.
    """
    if not email and not name:
        raise ValueError("Must provide either email or name")

    term = email or name or ""
    async for page in self.search_pages(term, page_size=10):
        for person in page.data:
            if _person_matches(person, email=email, name=name):
                return person

    return None

resolve_all(*, email: str | None = None, name: str | None = None) -> builtins.list[Person] async

Find all persons matching an email or name.

Notes: - This auto-paginates V1 search results to collect exact matches. - Unlike resolve(), this returns every match in server-provided order.

Source code in affinity/services/persons.py

async def resolve_all(
    self,
    *,
    email: str | None = None,
    name: str | None = None,
) -> builtins.list[Person]:
    """
    Find all persons matching an email or name.

    Notes:
    - This auto-paginates V1 search results to collect exact matches.
    - Unlike resolve(), this returns every match in server-provided order.
    """
    if not email and not name:
        raise ValueError("Must provide either email or name")

    term = email or name or ""
    matches: builtins.list[Person] = []
    async for page in self.search_pages(term, page_size=10):
        for person in page.data:
            if _person_matches(person, email=email, name=name):
                matches.append(person)
    return matches

search(term: str, *, with_interaction_dates: bool = False, with_interaction_persons: bool = False, with_opportunities: bool = False, page_size: int | None = None, page_token: str | None = None) -> PaginatedResponse[Person] async

Search for persons by name or email.

Uses V1 API for search functionality.

Source code in affinity/services/persons.py

async def search(
    self,
    term: str,
    *,
    with_interaction_dates: bool = False,
    with_interaction_persons: bool = False,
    with_opportunities: bool = False,
    page_size: int | None = None,
    page_token: str | None = None,
) -> PaginatedResponse[Person]:
    """
    Search for persons by name or email.

    Uses V1 API for search functionality.
    """
    params: dict[str, Any] = {"term": term}
    if with_interaction_dates:
        params["with_interaction_dates"] = True
    if with_interaction_persons:
        params["with_interaction_persons"] = True
    if with_opportunities:
        params["with_opportunities"] = True
    if page_size:
        params["page_size"] = page_size
    if page_token:
        params["page_token"] = page_token

    data = await self._client.get("/persons", params=params, v1=True)
    items = [Person.model_validate(p) for p in data.get("persons", [])]
    return PaginatedResponse[Person](
        data=items,
        next_page_token=data.get("next_page_token"),
    )

search_all(term: str, *, with_interaction_dates: bool = False, with_interaction_persons: bool = False, with_opportunities: bool = False, page_size: int | None = None, page_token: str | None = None) -> AsyncIterator[Person] async

Iterate all V1 person-search results with automatic pagination.

Parameters:

Name	Type	Description	Default
term	str	Search term (name or email)	required
with_interaction_dates	bool	Include interaction date data	False
with_interaction_persons	bool	Include persons for interactions	False
with_opportunities	bool	Include associated opportunity IDs	False
page_size	int | None	Results per page (max 500)	None
page_token	str | None	Resume from this pagination token	None

Yields:

Type	Description
AsyncIterator[Person]	Person objects matching the search term

Source code in affinity/services/persons.py

async def search_all(
    self,
    term: str,
    *,
    with_interaction_dates: bool = False,
    with_interaction_persons: bool = False,
    with_opportunities: bool = False,
    page_size: int | None = None,
    page_token: str | None = None,
) -> AsyncIterator[Person]:
    """
    Iterate all V1 person-search results with automatic pagination.

    Args:
        term: Search term (name or email)
        with_interaction_dates: Include interaction date data
        with_interaction_persons: Include persons for interactions
        with_opportunities: Include associated opportunity IDs
        page_size: Results per page (max 500)
        page_token: Resume from this pagination token

    Yields:
        Person objects matching the search term
    """
    async for page in self.search_pages(
        term,
        with_interaction_dates=with_interaction_dates,
        with_interaction_persons=with_interaction_persons,
        with_opportunities=with_opportunities,
        page_size=page_size,
        page_token=page_token,
    ):
        for person in page.data:
            yield person

search_pages(term: str, *, with_interaction_dates: bool = False, with_interaction_persons: bool = False, with_opportunities: bool = False, page_size: int | None = None, page_token: str | None = None) -> AsyncIterator[PaginatedResponse[Person]] async

Iterate V1 person-search result pages.

Useful for scripts that need checkpoint/resume via next_page_token.

Parameters:

Name	Type	Description	Default
term	str	Search term (name or email)	required
with_interaction_dates	bool	Include interaction date data	False
with_interaction_persons	bool	Include persons for interactions	False
with_opportunities	bool	Include associated opportunity IDs	False
page_size	int | None	Results per page (max 500)	None
page_token	str | None	Resume from this pagination token	None

Yields:

Type	Description
AsyncIterator[PaginatedResponse[Person]]	PaginatedResponse[Person] for each page

Source code in affinity/services/persons.py

async def search_pages(
    self,
    term: str,
    *,
    with_interaction_dates: bool = False,
    with_interaction_persons: bool = False,
    with_opportunities: bool = False,
    page_size: int | None = None,
    page_token: str | None = None,
) -> AsyncIterator[PaginatedResponse[Person]]:
    """
    Iterate V1 person-search result pages.

    Useful for scripts that need checkpoint/resume via `next_page_token`.

    Args:
        term: Search term (name or email)
        with_interaction_dates: Include interaction date data
        with_interaction_persons: Include persons for interactions
        with_opportunities: Include associated opportunity IDs
        page_size: Results per page (max 500)
        page_token: Resume from this pagination token

    Yields:
        PaginatedResponse[Person] for each page
    """
    requested_token = page_token
    page = await self.search(
        term,
        with_interaction_dates=with_interaction_dates,
        with_interaction_persons=with_interaction_persons,
        with_opportunities=with_opportunities,
        page_size=page_size,
        page_token=page_token,
    )
    while True:
        yield page
        next_token = page.next_page_token
        if not next_token or next_token == requested_token:
            return
        requested_token = next_token
        page = await self.search(
            term,
            with_interaction_dates=with_interaction_dates,
            with_interaction_persons=with_interaction_persons,
            with_opportunities=with_opportunities,
            page_size=page_size,
            page_token=next_token,
        )

update(person_id: PersonId, data: PersonUpdate) -> Person async

Update an existing person.

Note: To add emails/organizations, include existing values plus new ones.

Source code in affinity/services/persons.py

async def update(
    self,
    person_id: PersonId,
    data: PersonUpdate,
) -> Person:
    """
    Update an existing person.

    Note: To add emails/organizations, include existing values plus new ones.
    """
    payload = data.model_dump(
        by_alias=True,
        mode="json",
        exclude_unset=True,
        exclude_none=True,
    )

    result = await self._client.put(
        f"/persons/{person_id}",
        json=payload,
        v1=True,
    )

    if self._client.cache:
        self._client.cache.invalidate_prefix("person")

    return Person.model_validate(result)