Field values (V1)

Service for managing field values.

For list entry field values, prefer ListEntryService.update_field_value(). Use this for global field values not tied to list entries.

Source code in affinity/services/v1_only.py

class FieldValueService:
    """
    Service for managing field values.

    For list entry field values, prefer ListEntryService.update_field_value().
    Use this for global field values not tied to list entries.
    """

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

    def list(
        self,
        *,
        person_id: PersonId | None = None,
        company_id: CompanyId | None = None,
        opportunity_id: OpportunityId | None = None,
        list_entry_id: ListEntryId | None = None,
    ) -> list[FieldValue]:
        """
        Get field values for an entity.

        Exactly one of person_id, company_id, opportunity_id, or list_entry_id
        must be provided.

        Raises:
            ValueError: If zero or multiple IDs are provided.
        """
        provided = {
            name: value
            for name, value in (
                ("person_id", person_id),
                ("company_id", company_id),
                ("opportunity_id", opportunity_id),
                ("list_entry_id", list_entry_id),
            )
            if value is not None
        }
        if len(provided) == 0:
            raise ValueError(
                "field_values.list() requires exactly one entity ID. "
                "Example: client.field_values.list(person_id=PersonId(123))"
            )
        if len(provided) > 1:
            raise ValueError(
                f"field_values.list() accepts only one entity ID, "
                f"but received {len(provided)}: {', '.join(provided.keys())}. "
                "Call list() separately for each entity."
            )

        params: dict[str, Any] = {}
        if person_id is not None:
            params["person_id"] = int(person_id)
        if company_id is not None:
            params["organization_id"] = int(company_id)
        if opportunity_id is not None:
            params["opportunity_id"] = int(opportunity_id)
        if list_entry_id is not None:
            params["list_entry_id"] = int(list_entry_id)

        data = self._client.get("/field-values", params=params or None, v1=True)
        items = data.get("data", [])
        if not isinstance(items, list):
            items = []
        return [FieldValue.model_validate(v) for v in items]

    def create(self, data: FieldValueCreate) -> FieldValue:
        """
        Create a field value (V1 API).

        Note: V1 writes require numeric field IDs. The SDK accepts V2-style
        `field-<digits>` IDs and converts them; enriched/relationship-intelligence
        IDs are not supported.
        """
        payload = data.model_dump(by_alias=True, mode="json", exclude_unset=True, exclude_none=True)
        payload["field_id"] = field_id_to_v1_numeric(data.field_id)

        result = self._client.post("/field-values", json=payload, v1=True)
        return FieldValue.model_validate(result)

    def update(self, field_value_id: FieldValueId, value: Any) -> FieldValue:
        """Update a field value."""
        result = self._client.put(
            f"/field-values/{field_value_id}",
            json={"value": value},
            v1=True,
        )
        return FieldValue.model_validate(result)

    def delete(self, field_value_id: FieldValueId) -> bool:
        """Delete a field value."""
        result = self._client.delete(f"/field-values/{field_value_id}", v1=True)
        return bool(result.get("success", False))

    def get_for_entity(
        self,
        field_id: str | FieldId,
        *,
        person_id: PersonId | None = None,
        company_id: CompanyId | None = None,
        opportunity_id: OpportunityId | None = None,
        list_entry_id: ListEntryId | None = None,
        default: T = _UNSET,
    ) -> FieldValue | T | None:
        """
        Get a specific field value for an entity.

        Convenience method that fetches all field values and returns the one
        matching field_id. Like dict.get(), returns None (or default) if not found.

        Note: This still makes one API call to fetch all field values for the entity.
        For entities with hundreds of field values, prefer using ``list()`` directly
        if you need to inspect multiple fields.

        Args:
            field_id: The field to look up (accepts str or FieldId for convenience)
            person_id: Person entity (exactly one entity ID required)
            company_id: Company entity
            opportunity_id: Opportunity entity
            list_entry_id: List entry entity
            default: Value to return if field not found (default: None)

        Returns:
            FieldValue if the field has a value, default otherwise.
            Note: A FieldValue with ``.value is None`` still counts as "present" (explicit empty).

        Example:
            # Check if a person has a specific field value
            status = client.field_values.get_for_entity(
                "field-123",  # or FieldId("field-123")
                person_id=PersonId(456),
            )
            if status is None:
                print("Field is empty")
            else:
                print(f"Value: {status.value}")

            # With default value
            status = client.field_values.get_for_entity(
                "field-123",
                person_id=PersonId(456),
                default="N/A",
            )
        """
        all_values = self.list(
            person_id=person_id,
            company_id=company_id,
            opportunity_id=opportunity_id,
            list_entry_id=list_entry_id,
        )
        # Normalize field_id for comparison (handles both str and FieldId)
        target_id = FieldId(field_id) if not isinstance(field_id, FieldId) else field_id
        for fv in all_values:
            if fv.field_id == target_id:
                return fv
        return None if default is _UNSET else default

    def list_batch(
        self,
        person_ids: Sequence[PersonId] | None = None,
        company_ids: Sequence[CompanyId] | None = None,
        opportunity_ids: Sequence[OpportunityId] | None = None,
        *,
        on_error: Literal["raise", "skip"] = "raise",
    ) -> dict[PersonId | CompanyId | OpportunityId, builtins.list[FieldValue]]:
        """
        Get field values for multiple entities.

        **Performance note:** This makes one API call per entity (O(n) calls).
        There is no server-side batch endpoint. Use this for convenience and
        consistent error handling, not for performance optimization.
        For parallelism, use the async client.

        Args:
            person_ids: Sequence of person IDs (mutually exclusive with others)
            company_ids: Sequence of company IDs
            opportunity_ids: Sequence of opportunity IDs
            on_error: How to handle errors - "raise" (default) or "skip" failed IDs

        Returns:
            Dict mapping entity_id -> list of field values.
            Note: Dict ordering is not guaranteed; do not rely on insertion order.

        Example:
            # Check which persons have a specific field set
            fv_map = client.field_values.list_batch(person_ids=person_ids)
            for person_id, field_values in fv_map.items():
                has_status = any(fv.field_id == target_field for fv in field_values)
        """
        # Validate exactly one sequence provided
        provided = [
            ("person_ids", person_ids),
            ("company_ids", company_ids),
            ("opportunity_ids", opportunity_ids),
        ]
        non_none = [(name, seq) for name, seq in provided if seq is not None]
        if len(non_none) != 1:
            raise ValueError("Exactly one of person_ids, company_ids, or opportunity_ids required")

        name, ids = non_none[0]
        result: dict[PersonId | CompanyId | OpportunityId, list[FieldValue]] = {}

        for entity_id in ids:
            try:
                if name == "person_ids":
                    result[entity_id] = self.list(person_id=cast(PersonId, entity_id))
                elif name == "company_ids":
                    result[entity_id] = self.list(company_id=cast(CompanyId, entity_id))
                else:
                    result[entity_id] = self.list(opportunity_id=cast(OpportunityId, entity_id))
            except AffinityError:
                if on_error == "raise":
                    raise
                # skip: continue without this entity
            except Exception as e:
                if on_error == "raise":
                    # Preserve status_code if available
                    status_code = getattr(e, "status_code", None)
                    raise AffinityError(
                        f"Failed to get field values for {name[:-1]} {entity_id}: {e}",
                        status_code=status_code,
                    ) from e

        return result

create(data: FieldValueCreate) -> FieldValue

Create a field value (V1 API).

Note: V1 writes require numeric field IDs. The SDK accepts V2-style field-<digits> IDs and converts them; enriched/relationship-intelligence IDs are not supported.

Source code in affinity/services/v1_only.py

def create(self, data: FieldValueCreate) -> FieldValue:
    """
    Create a field value (V1 API).

    Note: V1 writes require numeric field IDs. The SDK accepts V2-style
    `field-<digits>` IDs and converts them; enriched/relationship-intelligence
    IDs are not supported.
    """
    payload = data.model_dump(by_alias=True, mode="json", exclude_unset=True, exclude_none=True)
    payload["field_id"] = field_id_to_v1_numeric(data.field_id)

    result = self._client.post("/field-values", json=payload, v1=True)
    return FieldValue.model_validate(result)

delete(field_value_id: FieldValueId) -> bool

Delete a field value.

Source code in affinity/services/v1_only.py

def delete(self, field_value_id: FieldValueId) -> bool:
    """Delete a field value."""
    result = self._client.delete(f"/field-values/{field_value_id}", v1=True)
    return bool(result.get("success", False))

get_for_entity(field_id: str | FieldId, *, person_id: PersonId | None = None, company_id: CompanyId | None = None, opportunity_id: OpportunityId | None = None, list_entry_id: ListEntryId | None = None, default: T = _UNSET) -> FieldValue | T | None

Get a specific field value for an entity.

Convenience method that fetches all field values and returns the one matching field_id. Like dict.get(), returns None (or default) if not found.

Note: This still makes one API call to fetch all field values for the entity. For entities with hundreds of field values, prefer using list() directly if you need to inspect multiple fields.

Parameters:

Name	Type	Description	Default
field_id	str | FieldId	The field to look up (accepts str or FieldId for convenience)	required
person_id	PersonId | None	Person entity (exactly one entity ID required)	None
company_id	CompanyId | None	Company entity	None
opportunity_id	OpportunityId | None	Opportunity entity	None
list_entry_id	ListEntryId | None	List entry entity	None
default	T	Value to return if field not found (default: None)	_UNSET

Returns:

Name	Type	Description
	FieldValue | T | None	FieldValue if the field has a value, default otherwise.
Note	FieldValue | T | None	A FieldValue with .value is None still counts as "present" (explicit empty).

Example

Check if a person has a specific field value

status = client.field_values.get_for_entity( "field-123", # or FieldId("field-123") person_id=PersonId(456), ) if status is None: print("Field is empty") else: print(f"Value: {status.value}")

With default value

status = client.field_values.get_for_entity( "field-123", person_id=PersonId(456), default="N/A", )

Source code in affinity/services/v1_only.py

def get_for_entity(
    self,
    field_id: str | FieldId,
    *,
    person_id: PersonId | None = None,
    company_id: CompanyId | None = None,
    opportunity_id: OpportunityId | None = None,
    list_entry_id: ListEntryId | None = None,
    default: T = _UNSET,
) -> FieldValue | T | None:
    """
    Get a specific field value for an entity.

    Convenience method that fetches all field values and returns the one
    matching field_id. Like dict.get(), returns None (or default) if not found.

    Note: This still makes one API call to fetch all field values for the entity.
    For entities with hundreds of field values, prefer using ``list()`` directly
    if you need to inspect multiple fields.

    Args:
        field_id: The field to look up (accepts str or FieldId for convenience)
        person_id: Person entity (exactly one entity ID required)
        company_id: Company entity
        opportunity_id: Opportunity entity
        list_entry_id: List entry entity
        default: Value to return if field not found (default: None)

    Returns:
        FieldValue if the field has a value, default otherwise.
        Note: A FieldValue with ``.value is None`` still counts as "present" (explicit empty).

    Example:
        # Check if a person has a specific field value
        status = client.field_values.get_for_entity(
            "field-123",  # or FieldId("field-123")
            person_id=PersonId(456),
        )
        if status is None:
            print("Field is empty")
        else:
            print(f"Value: {status.value}")

        # With default value
        status = client.field_values.get_for_entity(
            "field-123",
            person_id=PersonId(456),
            default="N/A",
        )
    """
    all_values = self.list(
        person_id=person_id,
        company_id=company_id,
        opportunity_id=opportunity_id,
        list_entry_id=list_entry_id,
    )
    # Normalize field_id for comparison (handles both str and FieldId)
    target_id = FieldId(field_id) if not isinstance(field_id, FieldId) else field_id
    for fv in all_values:
        if fv.field_id == target_id:
            return fv
    return None if default is _UNSET else default

list(*, person_id: PersonId | None = None, company_id: CompanyId | None = None, opportunity_id: OpportunityId | None = None, list_entry_id: ListEntryId | None = None) -> list[FieldValue]

Get field values for an entity.

Exactly one of person_id, company_id, opportunity_id, or list_entry_id must be provided.

Raises:

Type	Description
ValueError	If zero or multiple IDs are provided.

Source code in affinity/services/v1_only.py

def list(
    self,
    *,
    person_id: PersonId | None = None,
    company_id: CompanyId | None = None,
    opportunity_id: OpportunityId | None = None,
    list_entry_id: ListEntryId | None = None,
) -> list[FieldValue]:
    """
    Get field values for an entity.

    Exactly one of person_id, company_id, opportunity_id, or list_entry_id
    must be provided.

    Raises:
        ValueError: If zero or multiple IDs are provided.
    """
    provided = {
        name: value
        for name, value in (
            ("person_id", person_id),
            ("company_id", company_id),
            ("opportunity_id", opportunity_id),
            ("list_entry_id", list_entry_id),
        )
        if value is not None
    }
    if len(provided) == 0:
        raise ValueError(
            "field_values.list() requires exactly one entity ID. "
            "Example: client.field_values.list(person_id=PersonId(123))"
        )
    if len(provided) > 1:
        raise ValueError(
            f"field_values.list() accepts only one entity ID, "
            f"but received {len(provided)}: {', '.join(provided.keys())}. "
            "Call list() separately for each entity."
        )

    params: dict[str, Any] = {}
    if person_id is not None:
        params["person_id"] = int(person_id)
    if company_id is not None:
        params["organization_id"] = int(company_id)
    if opportunity_id is not None:
        params["opportunity_id"] = int(opportunity_id)
    if list_entry_id is not None:
        params["list_entry_id"] = int(list_entry_id)

    data = self._client.get("/field-values", params=params or None, v1=True)
    items = data.get("data", [])
    if not isinstance(items, list):
        items = []
    return [FieldValue.model_validate(v) for v in items]

list_batch(person_ids: Sequence[PersonId] | None = None, company_ids: Sequence[CompanyId] | None = None, opportunity_ids: Sequence[OpportunityId] | None = None, *, on_error: Literal['raise', 'skip'] = 'raise') -> dict[PersonId | CompanyId | OpportunityId, builtins.list[FieldValue]]

Get field values for multiple entities.

Performance note: This makes one API call per entity (O(n) calls). There is no server-side batch endpoint. Use this for convenience and consistent error handling, not for performance optimization. For parallelism, use the async client.

Parameters:

Name	Type	Description	Default
person_ids	Sequence[PersonId] | None	Sequence of person IDs (mutually exclusive with others)	None
company_ids	Sequence[CompanyId] | None	Sequence of company IDs	None
opportunity_ids	Sequence[OpportunityId] | None	Sequence of opportunity IDs	None
on_error	Literal['raise', 'skip']	How to handle errors - "raise" (default) or "skip" failed IDs	'raise'

Returns:

Name	Type	Description
	dict[PersonId | CompanyId | OpportunityId, list[FieldValue]]	Dict mapping entity_id -> list of field values.
Note	dict[PersonId | CompanyId | OpportunityId, list[FieldValue]]	Dict ordering is not guaranteed; do not rely on insertion order.

Example

Check which persons have a specific field set

fv_map = client.field_values.list_batch(person_ids=person_ids) for person_id, field_values in fv_map.items(): has_status = any(fv.field_id == target_field for fv in field_values)

Source code in affinity/services/v1_only.py

def list_batch(
    self,
    person_ids: Sequence[PersonId] | None = None,
    company_ids: Sequence[CompanyId] | None = None,
    opportunity_ids: Sequence[OpportunityId] | None = None,
    *,
    on_error: Literal["raise", "skip"] = "raise",
) -> dict[PersonId | CompanyId | OpportunityId, builtins.list[FieldValue]]:
    """
    Get field values for multiple entities.

    **Performance note:** This makes one API call per entity (O(n) calls).
    There is no server-side batch endpoint. Use this for convenience and
    consistent error handling, not for performance optimization.
    For parallelism, use the async client.

    Args:
        person_ids: Sequence of person IDs (mutually exclusive with others)
        company_ids: Sequence of company IDs
        opportunity_ids: Sequence of opportunity IDs
        on_error: How to handle errors - "raise" (default) or "skip" failed IDs

    Returns:
        Dict mapping entity_id -> list of field values.
        Note: Dict ordering is not guaranteed; do not rely on insertion order.

    Example:
        # Check which persons have a specific field set
        fv_map = client.field_values.list_batch(person_ids=person_ids)
        for person_id, field_values in fv_map.items():
            has_status = any(fv.field_id == target_field for fv in field_values)
    """
    # Validate exactly one sequence provided
    provided = [
        ("person_ids", person_ids),
        ("company_ids", company_ids),
        ("opportunity_ids", opportunity_ids),
    ]
    non_none = [(name, seq) for name, seq in provided if seq is not None]
    if len(non_none) != 1:
        raise ValueError("Exactly one of person_ids, company_ids, or opportunity_ids required")

    name, ids = non_none[0]
    result: dict[PersonId | CompanyId | OpportunityId, list[FieldValue]] = {}

    for entity_id in ids:
        try:
            if name == "person_ids":
                result[entity_id] = self.list(person_id=cast(PersonId, entity_id))
            elif name == "company_ids":
                result[entity_id] = self.list(company_id=cast(CompanyId, entity_id))
            else:
                result[entity_id] = self.list(opportunity_id=cast(OpportunityId, entity_id))
        except AffinityError:
            if on_error == "raise":
                raise
            # skip: continue without this entity
        except Exception as e:
            if on_error == "raise":
                # Preserve status_code if available
                status_code = getattr(e, "status_code", None)
                raise AffinityError(
                    f"Failed to get field values for {name[:-1]} {entity_id}: {e}",
                    status_code=status_code,
                ) from e

    return result

update(field_value_id: FieldValueId, value: Any) -> FieldValue

Update a field value.

Source code in affinity/services/v1_only.py

def update(self, field_value_id: FieldValueId, value: Any) -> FieldValue:
    """Update a field value."""
    result = self._client.put(
        f"/field-values/{field_value_id}",
        json={"value": value},
        v1=True,
    )
    return FieldValue.model_validate(result)