Persons

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,
        *,
        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:
            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
        """
        if cursor is not None:
            if any(p is not None for p in (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 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 pages(
        self,
        *,
        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:
            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 = (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(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,
        *,
        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:
            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(
                field_ids=field_ids,
                field_types=field_types,
                filter=filter,
            )

        return PageIterator(fetch_page)

    def iter(
        self,
        *,
        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(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,
    ) -> Person:
        """
        Get a single person by ID.

        Args:
            person_id: The person ID
            field_ids: Specific field IDs to include (V2 API)
            field_types: Field types to include (V2 API)
            include_field_values: If True, use V1 API to fetch embedded field values.
                This saves one API call when you need both person info and field values.
                Note: V1 response is ~2-3x larger than V2; consider this when fetching
                many persons. V1 may include field values for deleted fields.

        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.
        """
        if include_field_values:
            # Use V1 API which returns field values embedded
            data = self._client.get(f"/persons/{person_id}", v1=True)

            # Extract field_values before normalization
            field_values_data = data.get("field_values", [])

            # Normalize V1 response to V2 schema
            normalized = _normalize_v1_person_response(data)
            person = Person.model_validate(normalized)

            # Attach field_values as an attribute
            # Using object.__setattr__ to bypass Pydantic's frozen/immutable if needed
            object.__setattr__(person, "field_values", field_values_data)

            return person

        # Standard V2 path
        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", [])]

    # =========================================================================
    # 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,
    ) -> V1PaginatedResponse[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:
            Dict with 'persons' and 'next_page_token'
        """
        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 V1PaginatedResponse[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[V1PaginatedResponse[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:
            V1PaginatedResponse[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.

        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(*, 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
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,
    *,
    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:
        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(
            field_ids=field_ids,
            field_types=field_types,
            filter=filter,
        )

    return PageIterator(fetch_page)

create(data: PersonCreate) -> Person

Create a new person.

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.

    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) -> 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 (V2 API)	None
field_types	Sequence[FieldType] | None	Field types to include (V2 API)	None
include_field_values	bool	If True, use V1 API to fetch embedded field values. This saves one API call when you need both person info and field values. Note: V1 response is ~2-3x larger than V2; consider this when fetching many persons. V1 may include field values for deleted fields.	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.

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,
) -> Person:
    """
    Get a single person by ID.

    Args:
        person_id: The person ID
        field_ids: Specific field IDs to include (V2 API)
        field_types: Field types to include (V2 API)
        include_field_values: If True, use V1 API to fetch embedded field values.
            This saves one API call when you need both person info and field values.
            Note: V1 response is ~2-3x larger than V2; consider this when fetching
            many persons. V1 may include field values for deleted fields.

    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.
    """
    if include_field_values:
        # Use V1 API which returns field values embedded
        data = self._client.get(f"/persons/{person_id}", v1=True)

        # Extract field_values before normalization
        field_values_data = data.get("field_values", [])

        # Normalize V1 response to V2 schema
        normalized = _normalize_v1_person_response(data)
        person = Person.model_validate(normalized)

        # Attach field_values as an attribute
        # Using object.__setattr__ to bypass Pydantic's frozen/immutable if needed
        object.__setattr__(person, "field_values", field_values_data)

        return person

    # Standard V2 path
    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)

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_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(*, 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,
    *,
    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(field_ids=field_ids, field_types=field_types, filter=filter)

list(*, 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
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,
    *,
    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:
        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
    """
    if cursor is not None:
        if any(p is not None for p in (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 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(*, 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
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,
    *,
    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:
        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 = (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(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) -> V1PaginatedResponse[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
V1PaginatedResponse[Person]	Dict with 'persons' and 'next_page_token'

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,
) -> V1PaginatedResponse[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:
        Dict with 'persons' and 'next_page_token'
    """
    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 V1PaginatedResponse[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[V1PaginatedResponse[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
V1PaginatedResponse[Person]	V1PaginatedResponse[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[V1PaginatedResponse[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:
        V1PaginatedResponse[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,
        *,
        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:
            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
        """
        if cursor is not None:
            if any(p is not None for p in (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 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 pages(
        self,
        *,
        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:
            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 = (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(
                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,
        *,
        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:
            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(field_ids=field_ids, field_types=field_types, filter=filter)

        return AsyncPageIterator(fetch_page)

    def iter(
        self,
        *,
        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(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,
    ) -> Person:
        """
        Get a single person by ID.

        Args:
            person_id: The person ID
            field_ids: Specific field IDs to include (V2 API)
            field_types: Field types to include (V2 API)
            include_field_values: If True, use V1 API to fetch embedded field values.
                This saves one API call when you need both person info and field values.
                Note: V1 response is ~2-3x larger than V2; consider this when fetching
                many persons. V1 may include field values for deleted fields.

        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.
        """
        if include_field_values:
            # Use V1 API which returns field values embedded
            data = await self._client.get(f"/persons/{person_id}", v1=True)

            # Extract field_values before normalization
            field_values_data = data.get("field_values", [])

            # Normalize V1 response to V2 schema
            normalized = _normalize_v1_person_response(data)
            person = Person.model_validate(normalized)

            # Attach field_values as an attribute
            object.__setattr__(person, "field_values", field_values_data)

            return person

        # Standard V2 path
        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", [])]

    # =========================================================================
    # 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,
    ) -> V1PaginatedResponse[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 V1PaginatedResponse[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[V1PaginatedResponse[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:
            V1PaginatedResponse[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.

        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(*, 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
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,
    *,
    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:
        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(field_ids=field_ids, field_types=field_types, filter=filter)

    return AsyncPageIterator(fetch_page)

create(data: PersonCreate) -> Person async

Create a new person.

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.

    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) -> 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 (V2 API)	None
field_types	Sequence[FieldType] | None	Field types to include (V2 API)	None
include_field_values	bool	If True, use V1 API to fetch embedded field values. This saves one API call when you need both person info and field values. Note: V1 response is ~2-3x larger than V2; consider this when fetching many persons. V1 may include field values for deleted fields.	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.

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,
) -> Person:
    """
    Get a single person by ID.

    Args:
        person_id: The person ID
        field_ids: Specific field IDs to include (V2 API)
        field_types: Field types to include (V2 API)
        include_field_values: If True, use V1 API to fetch embedded field values.
            This saves one API call when you need both person info and field values.
            Note: V1 response is ~2-3x larger than V2; consider this when fetching
            many persons. V1 may include field values for deleted fields.

    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.
    """
    if include_field_values:
        # Use V1 API which returns field values embedded
        data = await self._client.get(f"/persons/{person_id}", v1=True)

        # Extract field_values before normalization
        field_values_data = data.get("field_values", [])

        # Normalize V1 response to V2 schema
        normalized = _normalize_v1_person_response(data)
        person = Person.model_validate(normalized)

        # Attach field_values as an attribute
        object.__setattr__(person, "field_values", field_values_data)

        return person

    # Standard V2 path
    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)

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_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(*, 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,
    *,
    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(field_ids=field_ids, field_types=field_types, filter=filter)

list(*, 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
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,
    *,
    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:
        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
    """
    if cursor is not None:
        if any(p is not None for p in (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 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(*, 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
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,
    *,
    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:
        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 = (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(
            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) -> V1PaginatedResponse[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,
) -> V1PaginatedResponse[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 V1PaginatedResponse[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[V1PaginatedResponse[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[V1PaginatedResponse[Person]]	V1PaginatedResponse[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[V1PaginatedResponse[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:
        V1PaginatedResponse[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)