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

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

Service for managing persons (contacts).

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

Source code in affinity/services/persons.py

class PersonService:
    """
    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,
        filter: str | FilterExpression | None = None,
        limit: int | None = None,
        cursor: str | None = None,
    ) -> 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
            filter: V2 filter expression string, or a FilterExpression built via `affinity.F`
            limit: Maximum number of results
            cursor: Cursor to resume pagination (opaque; obtained from prior responses)

        Returns:
            Paginated response with persons
        """
        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, filter, 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:
            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 filter is not None:
                filter_text = str(filter).strip()
                if filter_text:
                    params["filter"] = filter_text
            if limit:
                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,
        *,
        filter: str | FilterExpression | None = None,
        field_ids: Sequence[AnyFieldId] | None = None,
        field_types: Sequence[FieldType] | None = None,
    ) -> Person | None:
        """
        Get the first person matching the filter, or None if no match.

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

        Args:
            filter: V2 filter expression
            field_ids: Specific field IDs to include
            field_types: Field types to include

        Returns:
            First matching Person, or None if no results.
        """
        page = self.list(
            filter=filter,
            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,
        filter: str | FilterExpression | None = None,
        limit: int | None = None,
        cursor: str | None = None,
    ) -> 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
            filter: V2 filter expression string or FilterExpression
            limit: Maximum results per page
            cursor: Cursor to resume pagination

        Yields:
            PaginatedResponse[Person] for each page
        """
        other_params = (ids, field_ids, field_types, filter, 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
        page = (
            self.list(cursor=cursor)
            if cursor is not None
            else self.list(
                ids=ids, field_ids=field_ids, field_types=field_types, filter=filter, limit=limit
            )
        )
        while True:
            yield page
            if not page.has_next:
                return
            next_cursor = page.next_cursor
            if next_cursor is None or next_cursor == requested_cursor:
                return
            requested_cursor = next_cursor
            page = self.list(cursor=next_cursor)

    def all(
        self,
        *,
        ids: Sequence[PersonId] | None = None,
        field_ids: Sequence[AnyFieldId] | None = None,
        field_types: Sequence[FieldType] | None = None,
        filter: str | FilterExpression | None = None,
    ) -> 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
            filter: V2 filter expression

        Yields:
            Person objects
        """

        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,
                filter=filter,
            )

        return PageIterator(fetch_page)

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

        Alias for `all()` (FR-006 public contract).
        """
        return self.all(ids=ids, field_ids=field_ids, field_types=field_types, filter=filter)

    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,
    ) -> PaginatedResponse[ListEntry]:
        """Get all list entries for a person across all lists."""
        data = self._client.get(f"/persons/{person_id}/list-entries")

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

    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,
        *,
        field_types: Sequence[FieldType] | None = None,
    ) -> builtins.list[FieldMetadata]:
        """
        Get metadata about person fields.

        Cached for performance.
        """
        params: dict[str, Any] = {}
        if field_types:
            params["fieldTypes"] = [field_type.value for field_type in field_types]

        data = self._client.get(
            "/persons/fields",
            params=params or None,
            cache_key=f"person_fields:{','.join(field_types or [])}",
            cache_ttl=300,
        )

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

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

    def create(self, data: PersonCreate) -> Person:
        """
        Create a new person.

        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

        Raises:
            ValidationError: If email conflicts with existing person
        """
        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 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)

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

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

    # =========================================================================
    # 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.
        """
        if not self._client.enable_beta_endpoints:
            raise BetaEndpointDisabledError(
                "Person merge is a beta endpoint; set enable_beta_endpoints=True to use it."
            )
        result = self._client.post(
            "/person-merges",
            json={
                "primaryPersonId": int(primary_id),
                "duplicatePersonId": int(duplicate_id),
            },
        )
        return str(result.get("taskUrl", ""))

    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)

all(*, ids: Sequence[PersonId] | None = None, field_ids: Sequence[AnyFieldId] | None = None, field_types: Sequence[FieldType] | None = None, filter: str | FilterExpression | None = None) -> 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
filter	str | FilterExpression | None	V2 filter expression	None

Yields:

Type	Description
Person	Person objects

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,
    filter: str | FilterExpression | None = None,
) -> 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
        filter: V2 filter expression

    Yields:
        Person objects
    """

    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,
            filter=filter,
        )

    return PageIterator(fetch_page)

create(data: PersonCreate) -> Person

Create a new person.

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

Raises:

Type	Description
ValidationError	If email conflicts with existing person

Source code in affinity/services/persons.py

def create(self, data: PersonCreate) -> Person:
    """
    Create a new person.

    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

    Raises:
        ValidationError: If email conflicts with existing person
    """
    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)

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

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

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_fields(*, field_types: Sequence[FieldType] | None = None) -> builtins.list[FieldMetadata]

Get metadata about person fields.

Cached for performance.

Source code in affinity/services/persons.py

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

    Cached for performance.
    """
    params: dict[str, Any] = {}
    if field_types:
        params["fieldTypes"] = [field_type.value for field_type in field_types]

    data = self._client.get(
        "/persons/fields",
        params=params or None,
        cache_key=f"person_fields:{','.join(field_types or [])}",
        cache_ttl=300,
    )

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

get_first(*, filter: str | FilterExpression | None = None, field_ids: Sequence[AnyFieldId] | None = None, field_types: Sequence[FieldType] | None = None) -> Person | None

Get the first person matching the filter, or None if no match.

This is a convenience method equivalent to

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

Parameters:

Name	Type	Description	Default
filter	str | FilterExpression | None	V2 filter expression	None
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 matching Person, or None if no results.

Source code in affinity/services/persons.py

def get_first(
    self,
    *,
    filter: str | FilterExpression | None = None,
    field_ids: Sequence[AnyFieldId] | None = None,
    field_types: Sequence[FieldType] | None = None,
) -> Person | None:
    """
    Get the first person matching the filter, or None if no match.

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

    Args:
        filter: V2 filter expression
        field_ids: Specific field IDs to include
        field_types: Field types to include

    Returns:
        First matching Person, or None if no results.
    """
    page = self.list(
        filter=filter,
        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) -> PaginatedResponse[ListEntry]

Get all list entries for a person across all lists.

Source code in affinity/services/persons.py

def get_list_entries(
    self,
    person_id: PersonId,
) -> PaginatedResponse[ListEntry]:
    """Get all list entries for a person across all lists."""
    data = self._client.get(f"/persons/{person_id}/list-entries")

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

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_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, filter: str | FilterExpression | None = None) -> Iterator[Person]

Auto-paginate all persons.

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

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,
    filter: str | FilterExpression | None = None,
) -> Iterator[Person]:
    """
    Auto-paginate all persons.

    Alias for `all()` (FR-006 public contract).
    """
    return self.all(ids=ids, field_ids=field_ids, field_types=field_types, filter=filter)

list(*, ids: Sequence[PersonId] | None = None, field_ids: Sequence[AnyFieldId] | None = None, field_types: Sequence[FieldType] | None = None, filter: str | FilterExpression | None = None, limit: int | None = None, cursor: str | None = None) -> 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
filter	str | FilterExpression | None	V2 filter expression string, or a FilterExpression built via affinity.F	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

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,
    filter: str | FilterExpression | None = None,
    limit: int | None = None,
    cursor: str | None = None,
) -> 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
        filter: V2 filter expression string, or a FilterExpression built via `affinity.F`
        limit: Maximum number of results
        cursor: Cursor to resume pagination (opaque; obtained from prior responses)

    Returns:
        Paginated response with persons
    """
    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, filter, 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:
        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 filter is not None:
            filter_text = str(filter).strip()
            if filter_text:
                params["filter"] = filter_text
        if limit:
            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", {})),
    )

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.
    """
    if not self._client.enable_beta_endpoints:
        raise BetaEndpointDisabledError(
            "Person merge is a beta endpoint; set enable_beta_endpoints=True to use it."
        )
    result = self._client.post(
        "/person-merges",
        json={
            "primaryPersonId": int(primary_id),
            "duplicatePersonId": int(duplicate_id),
        },
    )
    return str(result.get("taskUrl", ""))

pages(*, ids: Sequence[PersonId] | None = None, field_ids: Sequence[AnyFieldId] | None = None, field_types: Sequence[FieldType] | None = None, filter: str | FilterExpression | None = None, limit: int | None = None, cursor: str | None = None) -> 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
filter	str | FilterExpression | None	V2 filter expression string or FilterExpression	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

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,
    filter: str | FilterExpression | None = None,
    limit: int | None = None,
    cursor: str | None = None,
) -> 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
        filter: V2 filter expression string or FilterExpression
        limit: Maximum results per page
        cursor: Cursor to resume pagination

    Yields:
        PaginatedResponse[Person] for each page
    """
    other_params = (ids, field_ids, field_types, filter, 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
    page = (
        self.list(cursor=cursor)
        if cursor is not None
        else self.list(
            ids=ids, field_ids=field_ids, field_types=field_types, filter=filter, limit=limit
        )
    )
    while True:
        yield page
        if not page.has_next:
            return
        next_cursor = page.next_cursor
        if next_cursor is None or next_cursor == requested_cursor:
            return
        requested_cursor = next_cursor
        page = self.list(cursor=next_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)

Async version of PersonService.

Source code in affinity/services/persons.py

class AsyncPersonService:
    """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,
        filter: str | FilterExpression | None = None,
        limit: int | None = None,
        cursor: str | None = None,
    ) -> 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
            filter: V2 filter expression string, or a FilterExpression built via `affinity.F`
            limit: Maximum number of results
            cursor: Cursor to resume pagination (opaque; obtained from prior responses)

        Returns:
            Paginated response with persons
        """
        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, filter, 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:
            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 filter is not None:
                filter_text = str(filter).strip()
                if filter_text:
                    params["filter"] = filter_text
            if limit:
                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,
        *,
        filter: str | FilterExpression | None = None,
        field_ids: Sequence[AnyFieldId] | None = None,
        field_types: Sequence[FieldType] | None = None,
    ) -> Person | None:
        """
        Get the first person matching the filter, or None if no match.

        See PersonService.get_first() for details.
        """
        page = await self.list(
            filter=filter,
            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.

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

    async def pages(
        self,
        *,
        ids: Sequence[PersonId] | None = None,
        field_ids: Sequence[AnyFieldId] | None = None,
        field_types: Sequence[FieldType] | None = None,
        filter: str | FilterExpression | None = None,
        limit: int | None = None,
        cursor: str | None = None,
    ) -> 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
            filter: V2 filter expression string or FilterExpression
            limit: Maximum results per page
            cursor: Cursor to resume pagination

        Yields:
            PaginatedResponse[Person] for each page
        """
        other_params = (ids, field_ids, field_types, filter, 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
        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,
                filter=filter,
                limit=limit,
            )
        while True:
            yield page
            if not page.has_next:
                return
            next_cursor = page.next_cursor
            if next_cursor is None or next_cursor == requested_cursor:
                return
            requested_cursor = next_cursor
            page = await self.list(cursor=next_cursor)

    def all(
        self,
        *,
        ids: Sequence[PersonId] | None = None,
        field_ids: Sequence[AnyFieldId] | None = None,
        field_types: Sequence[FieldType] | None = None,
        filter: str | FilterExpression | None = None,
    ) -> 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
            filter: V2 filter expression

        Yields:
            Person objects
        """

        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, filter=filter
            )

        return AsyncPageIterator(fetch_page)

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

        Alias for `all()` (FR-006 public contract).
        """
        return self.all(ids=ids, field_ids=field_ids, field_types=field_types, filter=filter)

    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,
    ) -> PaginatedResponse[ListEntry]:
        """Get all list entries for a person across all lists."""
        data = await self._client.get(f"/persons/{person_id}/list-entries")

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

    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,
        *,
        field_types: Sequence[FieldType] | None = None,
    ) -> builtins.list[FieldMetadata]:
        """
        Get metadata about person fields.

        Cached for performance.
        """
        params: dict[str, Any] = {}
        if field_types:
            params["fieldTypes"] = [field_type.value for field_type in field_types]

        data = await self._client.get(
            "/persons/fields",
            params=params or None,
            cache_key=f"person_fields:{','.join(field_types or [])}",
            cache_ttl=300,
        )

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

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

    async def create(self, data: PersonCreate) -> Person:
        """
        Create a new person.

        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

        Raises:
            ValidationError: If email conflicts with existing person
        """
        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 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)

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

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

    # =========================================================================
    # 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.
        """
        if not self._client.enable_beta_endpoints:
            raise BetaEndpointDisabledError(
                "Person merge is a beta endpoint; set enable_beta_endpoints=True to use it."
            )
        result = await self._client.post(
            "/person-merges",
            json={
                "primaryPersonId": int(primary_id),
                "duplicatePersonId": int(duplicate_id),
            },
        )
        return str(result.get("taskUrl", ""))

    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)

all(*, ids: Sequence[PersonId] | None = None, field_ids: Sequence[AnyFieldId] | None = None, field_types: Sequence[FieldType] | None = None, filter: str | FilterExpression | None = None) -> 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
filter	str | FilterExpression | None	V2 filter expression	None

Yields:

Type	Description
AsyncIterator[Person]	Person objects

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,
    filter: str | FilterExpression | None = None,
) -> 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
        filter: V2 filter expression

    Yields:
        Person objects
    """

    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, filter=filter
        )

    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.

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.

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

create(data: PersonCreate) -> Person async

Create a new person.

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

Raises:

Type	Description
ValidationError	If email conflicts with existing person

Source code in affinity/services/persons.py

async def create(self, data: PersonCreate) -> Person:
    """
    Create a new person.

    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

    Raises:
        ValidationError: If email conflicts with existing person
    """
    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)

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

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

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_fields(*, field_types: Sequence[FieldType] | None = None) -> builtins.list[FieldMetadata] async

Get metadata about person fields.

Cached for performance.

Source code in affinity/services/persons.py

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

    Cached for performance.
    """
    params: dict[str, Any] = {}
    if field_types:
        params["fieldTypes"] = [field_type.value for field_type in field_types]

    data = await self._client.get(
        "/persons/fields",
        params=params or None,
        cache_key=f"person_fields:{','.join(field_types or [])}",
        cache_ttl=300,
    )

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

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

Get the first person matching the filter, or None if no match.

See PersonService.get_first() for details.

Source code in affinity/services/persons.py

async def get_first(
    self,
    *,
    filter: str | FilterExpression | None = None,
    field_ids: Sequence[AnyFieldId] | None = None,
    field_types: Sequence[FieldType] | None = None,
) -> Person | None:
    """
    Get the first person matching the filter, or None if no match.

    See PersonService.get_first() for details.
    """
    page = await self.list(
        filter=filter,
        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) -> PaginatedResponse[ListEntry] async

Get all list entries for a person across all lists.

Source code in affinity/services/persons.py

async def get_list_entries(
    self,
    person_id: PersonId,
) -> PaginatedResponse[ListEntry]:
    """Get all list entries for a person across all lists."""
    data = await self._client.get(f"/persons/{person_id}/list-entries")

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

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_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, filter: str | FilterExpression | None = None) -> AsyncIterator[Person]

Auto-paginate all persons.

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

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,
    filter: str | FilterExpression | None = None,
) -> AsyncIterator[Person]:
    """
    Auto-paginate all persons.

    Alias for `all()` (FR-006 public contract).
    """
    return self.all(ids=ids, field_ids=field_ids, field_types=field_types, filter=filter)

list(*, ids: Sequence[PersonId] | None = None, field_ids: Sequence[AnyFieldId] | None = None, field_types: Sequence[FieldType] | None = None, filter: str | FilterExpression | None = None, limit: int | None = None, cursor: str | None = None) -> 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
filter	str | FilterExpression | None	V2 filter expression string, or a FilterExpression built via affinity.F	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

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,
    filter: str | FilterExpression | None = None,
    limit: int | None = None,
    cursor: str | None = None,
) -> 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
        filter: V2 filter expression string, or a FilterExpression built via `affinity.F`
        limit: Maximum number of results
        cursor: Cursor to resume pagination (opaque; obtained from prior responses)

    Returns:
        Paginated response with persons
    """
    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, filter, 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:
        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 filter is not None:
            filter_text = str(filter).strip()
            if filter_text:
                params["filter"] = filter_text
        if limit:
            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", {})),
    )

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.
    """
    if not self._client.enable_beta_endpoints:
        raise BetaEndpointDisabledError(
            "Person merge is a beta endpoint; set enable_beta_endpoints=True to use it."
        )
    result = await self._client.post(
        "/person-merges",
        json={
            "primaryPersonId": int(primary_id),
            "duplicatePersonId": int(duplicate_id),
        },
    )
    return str(result.get("taskUrl", ""))

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

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
filter	str | FilterExpression | None	V2 filter expression string or FilterExpression	None
limit	int | None	Maximum results per page	None
cursor	str | None	Cursor to resume pagination	None

Yields:

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

Source code in affinity/services/persons.py

async def pages(
    self,
    *,
    ids: Sequence[PersonId] | None = None,
    field_ids: Sequence[AnyFieldId] | None = None,
    field_types: Sequence[FieldType] | None = None,
    filter: str | FilterExpression | None = None,
    limit: int | None = None,
    cursor: str | None = None,
) -> 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
        filter: V2 filter expression string or FilterExpression
        limit: Maximum results per page
        cursor: Cursor to resume pagination

    Yields:
        PaginatedResponse[Person] for each page
    """
    other_params = (ids, field_ids, field_types, filter, 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
    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,
            filter=filter,
            limit=limit,
        )
    while True:
        yield page
        if not page.has_next:
            return
        next_cursor = page.next_cursor
        if next_cursor is None or next_cursor == requested_cursor:
            return
        requested_cursor = next_cursor
        page = await self.list(cursor=next_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)