Companies

Note: CompanyService includes two v1-only exceptions for company -> people associations: get_associated_person_ids(...) and get_associated_people(...). V2 does not expose a direct company -> people relationship endpoint, so these methods use the v1 organizations API under the hood. They are documented as exceptions and may be superseded when v2 adds parity.

Interaction Dates

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

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

from affinity import Affinity
from affinity.types import CompanyId

with Affinity(api_key="YOUR_API_KEY") as client:
    # Fetch company with interaction dates
    company = client.companies.get(
        CompanyId(123),
        with_interaction_dates=True,
        with_interaction_persons=True,  # Include person IDs for each interaction
    )

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

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

Many companies

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

For just the last and next meeting dates, the relationship-intelligence fields are cheaper: they come with the company records, one call per 100, with no per-company call:

from affinity.types import FieldType

for company in client.companies.iter(ids=ids[:100], field_types=[FieldType.RELATIONSHIP_INTELLIGENCE]):
    ...  # the last/next event fields are among company.fields

Pass at most 100 ids per call (the ids go in the URL). The other interaction dates and the interaction persons have no confirmed counterpart there, so use with_interaction_dates for those.

Finding records with Affinity's search

find() filters and sorts on Affinity's side (V2 search, beta), so only matching records are downloaded. The same methods exist on client.persons and on client.lists.entries(list_id).

find(
    where=None, *, sort=None, term=None, term_fields=None, field_ids=None,
    field_types=None, limit=None, total_count=False, cursor=None,
) -> PaginatedResponse[Company]
iter_find(where=None, *, sort=None, term=None, term_fields=None,
          field_ids=None, field_types=None) -> Iterator[Company]
find_body(body, *, field_ids=None, field_types=None, limit=None,
          total_count=False, cursor=None) -> PaginatedResponse[Company]

• where: a Condition or a group from all_of(...) / any_of(...) (affinity.criteria).
• sort: one Sort or a list of up to 5.
• term: a keyword (3+ characters) matched against names, domains and emails, and against the fields in term_fields.
• field_ids / field_types: the field values to return with each record.
• limit: 1-100 records per page.
• iter_find yields every matching record across pages. find_body takes a prebuilt search body (the API's SearchCriteria JSON).

from datetime import date
from affinity.criteria import Condition, Sort, all_of

page = client.companies.find(
    all_of(
        Condition("field-498220", "is-any-of", [3, 4]),
        Condition("affinity-data-last-funding-date", "is-after", date(2025, 1, 1)),
    ),
    sort=[Sort("affinity-data-total-funding-amount", "desc")],
    limit=50,
)

Next page: call find again with the same criteria and cursor=page.next_cursor. A cursor alone isn't enough: the search needs its body again.

Values: each condition's value type is read from the fields metadata when omitted (one cached request), and the operator is checked against the field. Company, person, list and dropdown fields take ids (dropdown values are option ids); dates are datetime.date or "YYYY-MM-DD" (not datetimes); relative windows are (6, "month").

Permissions: search needs the "Export All Organizations directory" permission (companies, persons) or "Export data from Lists" (list entries). Without it, find raises AuthorizationError naming the permission. Search is a read, so it's allowed in read-only mode.

search(term) is the separate name/domain search and needs no extra permission.

Bases: CompanySemanticSearchMixin, CompanyDiscoveryMixin, SingleFieldMixin, SearchMixin[Company]

Source code in affinity/services/companies.py

class CompanyService(
    CompanySemanticSearchMixin, CompanyDiscoveryMixin, SingleFieldMixin, SearchMixin[Company]
):
    _entity_path = "/companies"

    _search_model = Company

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

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

    """
    Service for managing companies (organizations).

    Note: Companies are called Organizations in the V1 API. This service
    uses V2 terminology throughout but routes to V1 for create/update/delete.
    """

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

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

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

        Args:
            ids: Specific company IDs to fetch (batch lookup)
            field_ids: Specific field IDs to include in response
            field_types: Field types to include (e.g., ["enriched", "global"])
            limit: Maximum number of results (API default: 100)
            cursor: Cursor to resume pagination (opaque; obtained from prior responses)

        Returns:
            Paginated response with companies

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

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

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

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

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

        Returns:
            First Company, or None if no results.

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

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

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

        Args:
            ids: Specific company IDs to fetch (batch lookup)
            field_ids: Specific field IDs to include in response
            field_types: Field types to include (e.g., ["enriched", "global"])
            limit: Maximum results per page
            cursor: Cursor to resume pagination

        Yields:
            PaginatedResponse[Company] for each page
        """
        if "filter" in _unsupported:
            raise ValueError(
                "V2 /companies does not support server-side filter. "
                "To search by name/domain use client.companies.search_pages(term). "
                "To filter by list-specific fields use "
                "client.lists.entries(list_id).list(filter=...)."
            )
        if _unsupported:
            raise TypeError(f"Unexpected keyword arguments: {list(_unsupported)}")
        other_params = (ids, field_ids, field_types, limit)
        if cursor is not None and any(p is not None for p in other_params):
            raise ValueError(
                "Cannot combine 'cursor' with other parameters; cursor encodes all query context. "
                "Start a new pagination sequence without a cursor to change parameters."
            )
        requested_cursor = cursor
        guard = _CursorGuard(cursor)
        page = (
            self.list(cursor=cursor)
            if cursor is not None
            else self.list(ids=ids, field_ids=field_ids, field_types=field_types, limit=limit)
        )
        while True:
            yield page
            # Stop at the last page, or when paging can't progress (see _CursorGuard).
            if guard.last_page(requested_cursor, page):
                return
            requested_cursor = page.next_cursor
            page = self.list(cursor=requested_cursor)

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

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

        Yields:
            Company objects

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

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

        return PageIterator(fetch_page)

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

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

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

    def get(
        self,
        company_id: CompanyId,
        *,
        field_ids: Sequence[AnyFieldId] | None = None,
        field_types: Sequence[FieldType] | None = None,
        retries: int = 0,
        with_interaction_dates: bool = False,
        with_interaction_persons: bool = False,
    ) -> Company:
        """
        Get a single company by ID.

        Args:
            company_id: The company ID
            field_ids: Specific field IDs to include in response
            field_types: Field types to include (e.g., ["enriched", "global"])
            retries: Number of retries on 404 NotFoundError. Default is 0 (fail fast).
                Set to 2-3 if calling immediately after create() to handle eventual
                consistency lag.
            with_interaction_dates: Include interaction date summaries (last/next
                meeting dates, email dates).
            with_interaction_persons: Include person IDs for each interaction.
                Only applies when with_interaction_dates=True.

        Returns:
            Company object with requested field data. When with_interaction_dates=True,
            the Company will have interaction_dates and interactions populated.

        Raises:
            NotFoundError: If company does not exist after all retries.

        Note:
            When combining with_interaction_dates with field_ids/field_types,
            two API calls are made internally and the results are merged.
        """
        last_error: NotFoundError | None = None
        attempts = retries + 1  # retries=0 means 1 attempt
        has_field_filters = field_ids is not None or field_types is not None

        for attempt in range(attempts):
            try:
                if with_interaction_dates:
                    # Fetch interaction data
                    v1_params: dict[str, Any] = {"with_interaction_dates": True}
                    if with_interaction_persons:
                        v1_params["with_interaction_persons"] = True
                    interaction_data = self._client.get(
                        f"/organizations/{company_id}",
                        params=v1_params,
                        v1=True,
                    )

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

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

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

                    # No field filtering, return interaction data directly
                    return Company.model_validate(interaction_data)

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

                data = self._client.get(
                    f"/companies/{company_id}",
                    params=params or None,
                )
                return Company.model_validate(data)
            except NotFoundError as e:
                last_error = e
                if attempt < attempts - 1:  # Don't sleep after last attempt
                    time.sleep(0.5 * (attempt + 1))  # 0.5s, 1s, 1.5s backoff

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

        raise last_error  # type: ignore[misc]

    def get_many(
        self,
        company_ids: Sequence[CompanyId],
        *,
        field_ids: Sequence[AnyFieldId] | None = None,
        field_types: Sequence[FieldType] | None = None,
    ) -> PaginatedResponse[Company]:
        """
        Fetch multiple companies by ID in a single API call.

        This is a convenience alias for ``list(ids=[...])``.

        Args:
            company_ids: Company IDs to fetch
            field_ids: Specific field IDs to include in response
            field_types: Field types to include (e.g., ["enriched", "global"])

        Returns:
            Paginated response with companies

        Example:
            >>> companies = client.companies.get_many([CompanyId(1), CompanyId(2)])
            >>> for company in companies.data:
            ...     print(company.name)
        """
        return self.list(ids=company_ids, field_ids=field_ids, field_types=field_types)

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

        V1-only exception: V2 does not expose company -> people associations.
        Uses GET `/organizations/{id}` and returns `person_ids` if present.
        """
        data = self._client.get(f"/organizations/{company_id}", v1=True)
        organization = data.get("organization") if isinstance(data, dict) else None
        source = organization if isinstance(organization, dict) else data
        person_ids = None
        if isinstance(source, dict):
            person_ids = source.get("person_ids") or source.get("personIds")

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

        ids = [PersonId(int(value)) for value in person_ids if value is not None]
        if max_results is not None and max_results >= 0:
            return ids[:max_results]
        return ids

    def get_associated_person_ids_batch(
        self,
        company_ids: Sequence[CompanyId],
        *,
        on_error: Literal["raise", "skip"] = "raise",
    ) -> dict[CompanyId, builtins.list[PersonId]]:
        """
        Get person associations for multiple companies.

        Makes one V1 API call per company.

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

        Returns:
            Dict mapping company_id -> list of person_ids

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

    def get_associated_people(
        self,
        company_id: CompanyId,
        *,
        max_results: int | None = None,
    ) -> builtins.list[Person]:
        """
        Get Person objects associated with a company.

        Uses V2 batch lookup for efficiency (1 API call per 100 persons
        instead of 1 per person).
        """
        person_ids = self.get_associated_person_ids(company_id, max_results=max_results)
        if not person_ids:
            return []

        # Use V2 batch lookup: GET /persons?ids=1&ids=2&ids=3
        # Note: person_ids is already truncated by get_associated_person_ids if max_results set
        params: dict[str, Any] = {"ids": [int(pid) for pid in person_ids]}

        people: builtins.list[Person] = []
        data = self._client.get("/persons", params=params)  # V2 batch
        for item in data.get("data", []):
            people.append(Person.model_validate(item))

        # Handle pagination if needed (>100 persons)
        # Note: max_results check is defensive - person_ids was already truncated above
        pagination = data.get("pagination") or {}
        next_url = pagination.get("nextUrl")
        while next_url and (max_results is None or len(people) < max_results):
            data = self._client.get_url(next_url)
            for item in data.get("data", []):
                people.append(Person.model_validate(item))
            next_url = (data.get("pagination") or {}).get("nextUrl")

        if max_results:
            return people[:max_results]
        return people

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

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

        Args:
            company_id: The company ID
            max_results: Maximum number of opportunity IDs to return

        Returns:
            List of OpportunityId values associated with this company
        """
        data = self._client.get(f"/organizations/{company_id}", v1=True)
        # Defensive: handle potential {"organization": {...}} wrapper
        organization = data.get("organization") if isinstance(data, dict) else None
        source = organization if isinstance(organization, dict) else data
        opp_ids = None
        if isinstance(source, dict):
            opp_ids = source.get("opportunity_ids") or source.get("opportunityIds")

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

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

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

        Makes one V1 API call per company.

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

        Returns:
            Dict mapping company_id -> list of opportunity_ids

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

    def get_list_entries(
        self,
        company_id: CompanyId,
        *,
        limit: int | None = None,
        cursor: str | None = None,
    ) -> PaginatedResponse[ListEntry]:
        """
        Get all list entries for a company across all lists.

        Returns comprehensive field data for each list entry. Each entry's ``entity_id`` is
        ``company_id`` and its ``entity_type`` is ``EntityType.ORGANIZATION``; ``entity`` is
        ``None`` (this endpoint doesn't embed the company). A ``cursor`` must come from this
        company's entries.
        """
        path = f"/companies/{company_id}/list-entries"
        if cursor is not None:
            if limit is not None:
                raise ValueError(
                    "Cannot combine 'cursor' with other parameters; cursor encodes all query "
                    "context. Start a new pagination sequence without a cursor to change "
                    "parameters."
                )
            check_list_entries_cursor(cursor, path)
            data = self._client.get_url(cursor)
        else:
            params: dict[str, Any] = {}
            check_positive_limit(limit)
            if limit is not None:
                params["limit"] = limit
            data = self._client.get(path, params=params or None)

        return entity_list_entries_page(
            data, entity_id=company_id, entity_type=EntityType.ORGANIZATION
        )

    def get_lists(
        self,
        company_id: CompanyId,
        *,
        limit: int | None = None,
        cursor: str | None = None,
    ) -> PaginatedResponse[ListSummary]:
        """Get all lists that contain this company."""
        if cursor is not None:
            if limit is not None:
                raise ValueError(
                    "Cannot combine 'cursor' with other parameters; cursor encodes all query "
                    "context. Start a new pagination sequence without a cursor to change "
                    "parameters."
                )
            data = self._client.get_url(cursor)
        else:
            params: dict[str, Any] = {}
            check_positive_limit(limit)
            if limit is not None:
                params["limit"] = limit
            data = self._client.get(
                f"/companies/{company_id}/lists",
                params=params or None,
            )

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

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

        Cached for performance.

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

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

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

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

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

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

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

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

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

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

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

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

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

        Uses V1 API for search functionality not available in V2.

        Args:
            term: Search term (name or domain)
            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 'organizations' 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("/organizations", params=params, v1=True)
        items = [Company.model_validate(o) for o in data.get("organizations", [])]
        return PaginatedResponse[Company](
            data=items,
            next_page_token=data.get("next_page_token"),
        )

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

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

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

        Yields:
            PaginatedResponse[Company] 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[Company]:
        """
        Iterate all V1 company-search results with automatic pagination.

        Args:
            term: Search term (name or domain)
            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:
            Company 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,
        *,
        domain: str | None = None,
        name: str | None = None,
    ) -> Company | None:
        """
        Find a single company by domain 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:
            domain: Domain to search for (e.g., "acme.com")
            name: Company name to search for

        Returns:
            The matching Company, or None if not found

        Raises:
            ValueError: If neither domain nor name is provided

        Note:
            If multiple matches are found, returns the first one.
            For disambiguation, use search() directly.
        """
        if not domain and not name:
            raise ValueError("Must provide either domain or name")

        term = domain or name or ""
        result = self.search(term, page_size=10)

        for company in result.data:
            if domain and company.domain and company.domain.lower() == domain.lower():
                return company
            if name and company.name and company.name.lower() == name.lower():
                return company

        return None

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

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

    def create(self, data: CompanyCreate, *, if_not_exists: bool = True) -> Company:
        """
        Create a new company.

        Args:
            data: Company creation data
            if_not_exists: If True (default), check for an existing company with
                the same name (case-insensitive exact match) OR the same domain
                (matched against the company's primary domain AND all entries
                in its `domains` list). Globals in Affinity's shared directory
                are treated as duplicates — per the API docs, POST /organizations
                always creates a tenant-scoped copy, so creating against an
                existing global would produce a genuine duplicate. Raises
                DuplicateEntityError carrying the existing company's ID and
                `existing_is_global` flag, so callers can use the global ID
                directly (e.g., via List Entries) instead of POSTing a copy.
                Uses V1 search_pages() with page_size=500, capped at 3 pages.
                Set to False to skip the check and create unconditionally.

        Returns:
            Created company

        Raises:
            DuplicateEntityError: When if_not_exists=True and an exact-name or
                exact-domain match already exists (tenant-scoped OR global).

        Note:
            Creates use V1 API, while reads use V2 API. Due to eventual consistency
            between V1 and V2, a `get()` call immediately after `create()` may return
            404 NotFoundError. If you need to read immediately after creation, either:
            - Use the Company object returned by this method (it contains the created data)
            - Add a short delay (100-500ms) before calling get()
            - Implement retry logic in your application
        """
        if if_not_exists:
            existing = self._find_exact_duplicate(name=data.name, domain=data.domain)
            if existing is not None:
                if existing.is_global:
                    message = (
                        f"A global Affinity directory record matches name={data.name!r} "
                        f"or domain={data.domain!r} (id={existing.id}). Global records "
                        f"are shared across tenants — use this ID directly (e.g., via "
                        f"List Entries) instead of creating a tenant-scoped duplicate."
                    )
                else:
                    message = (
                        f"Company with name={data.name!r} or domain={data.domain!r} "
                        f"already exists (id={existing.id})"
                    )
                raise DuplicateEntityError(
                    message,
                    entity_type="company",
                    existing_id=int(existing.id),
                    existing_name=existing.name,
                    existing_domain=existing.domain,
                    existing_is_global=bool(existing.is_global),
                )

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

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

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

        return Company.model_validate(result)

    @staticmethod
    def _company_matches(company: Company, name_lower: str, domain_lower: str | None) -> bool:
        """True iff company's name or any domain matches exactly (case-insensitive)."""
        if company.name and company.name.lower() == name_lower:
            return True
        if domain_lower:
            if company.domain and company.domain.lower() == domain_lower:
                return True
            for d in company.domains or []:
                if d and d.lower() == domain_lower:
                    return True
        return False

    def _find_exact_duplicate(self, *, name: str, domain: str | None) -> Company | None:
        """Search V1 for an exact-name or exact-domain match. Returns None if no match.

        Domain-first search order: domains are globally more unique than names,
        so if a domain is supplied, we search by domain first. Falls back to a
        name search only when the domain search came up empty and the name
        differs from the domain string.

        Iterates at most `_DEDUP_PAGE_CAP` pages per term (1500 fuzzy results).
        """
        name_lower = name.strip().lower()
        domain_lower = domain.strip().lower() if domain else None

        def _scan(term: str) -> Company | None:
            for page_num, page in enumerate(self.search_pages(term, page_size=500)):
                for company in page.data:
                    if self._company_matches(company, name_lower, domain_lower):
                        return company
                if page_num + 1 >= self._DEDUP_PAGE_CAP:
                    break
                if not page.next_page_token:
                    break
            return None

        if domain_lower:
            # domain is guaranteed non-None when domain_lower is set
            hit = _scan(domain or "")
            if hit is not None:
                return hit
            if name_lower and name_lower != domain_lower:
                return _scan(name)
            return None

        return _scan(name)

    def update(
        self,
        company_id: CompanyId,
        data: CompanyUpdate,
    ) -> Company:
        """
        Update an existing company.

        Note: Cannot update name/domain of global companies.
        """
        payload = data.model_dump(
            by_alias=True,
            mode="json",
            exclude_unset=True,
            exclude_none=True,
        )

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

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

        return Company.model_validate(result)

    def delete(self, company_id: CompanyId) -> bool:
        """
        Delete a company.

        Note: Cannot delete global companies.
        """
        result = self._client.delete(f"/organizations/{company_id}", v1=True)
        invalidate_all_list_sizes(self._client)

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

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

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

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

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

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

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

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

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

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

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

        return PageIterator(fetch_page)

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

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

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

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

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

        return PageIterator(fetch_page)

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

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

        Filters: ``creator_id``, ``created_after`` (inclusive) / ``created_before`` (exclusive),
        ``updated_after``, rounded outward to whole seconds. ``limit`` is 1-100.
        Needs Affinity API version 2026-07-15 or newer (sent for this call and every page).
        """
        check_cursor_alone(
            cursor, creator_id, created_after, created_before, updated_after, limit, total_count
        )
        if cursor is not None:
            return page_of(
                NoteV2, self._client.get_url(cursor, min_api_version=COMPANY_NOTES_MIN_API_VERSION)
            )
        params = note_params(
            creator_id=creator_id,
            created_after=created_after,
            created_before=created_before,
            updated_after=updated_after,
            limit=limit,
            total_count=total_count,
        )
        return page_of(
            NoteV2,
            self._client.get(
                f"/companies/{int(company_id)}/notes",
                params=params or None,
                min_api_version=COMPANY_NOTES_MIN_API_VERSION,
            ),
        )

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

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

        return PageIterator(fetch_page)

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

Iterate through all companies with automatic pagination.

Parameters:

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

Yields:

Type	Description
Company	Company objects

Note

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

Source code in affinity/services/companies.py

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

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

    Yields:
        Company objects

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

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

    return PageIterator(fetch_page)

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

Update several field values of one company in one request.

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

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

Parameters:

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

Source code in affinity/services/companies.py

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

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

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

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

create(data: CompanyCreate, *, if_not_exists: bool = True) -> Company

Create a new company.

Parameters:

Name	Type	Description	Default
data	CompanyCreate	Company creation data	required
if_not_exists	bool	If True (default), check for an existing company with the same name (case-insensitive exact match) OR the same domain (matched against the company's primary domain AND all entries in its domains list). Globals in Affinity's shared directory are treated as duplicates — per the API docs, POST /organizations always creates a tenant-scoped copy, so creating against an existing global would produce a genuine duplicate. Raises DuplicateEntityError carrying the existing company's ID and existing_is_global flag, so callers can use the global ID directly (e.g., via List Entries) instead of POSTing a copy. Uses V1 search_pages() with page_size=500, capped at 3 pages. Set to False to skip the check and create unconditionally.	True

Returns:

Type	Description
Company	Created company

Raises:

Type	Description
DuplicateEntityError	When if_not_exists=True and an exact-name or exact-domain match already exists (tenant-scoped OR global).

Note

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

Source code in affinity/services/companies.py

def create(self, data: CompanyCreate, *, if_not_exists: bool = True) -> Company:
    """
    Create a new company.

    Args:
        data: Company creation data
        if_not_exists: If True (default), check for an existing company with
            the same name (case-insensitive exact match) OR the same domain
            (matched against the company's primary domain AND all entries
            in its `domains` list). Globals in Affinity's shared directory
            are treated as duplicates — per the API docs, POST /organizations
            always creates a tenant-scoped copy, so creating against an
            existing global would produce a genuine duplicate. Raises
            DuplicateEntityError carrying the existing company's ID and
            `existing_is_global` flag, so callers can use the global ID
            directly (e.g., via List Entries) instead of POSTing a copy.
            Uses V1 search_pages() with page_size=500, capped at 3 pages.
            Set to False to skip the check and create unconditionally.

    Returns:
        Created company

    Raises:
        DuplicateEntityError: When if_not_exists=True and an exact-name or
            exact-domain match already exists (tenant-scoped OR global).

    Note:
        Creates use V1 API, while reads use V2 API. Due to eventual consistency
        between V1 and V2, a `get()` call immediately after `create()` may return
        404 NotFoundError. If you need to read immediately after creation, either:
        - Use the Company object returned by this method (it contains the created data)
        - Add a short delay (100-500ms) before calling get()
        - Implement retry logic in your application
    """
    if if_not_exists:
        existing = self._find_exact_duplicate(name=data.name, domain=data.domain)
        if existing is not None:
            if existing.is_global:
                message = (
                    f"A global Affinity directory record matches name={data.name!r} "
                    f"or domain={data.domain!r} (id={existing.id}). Global records "
                    f"are shared across tenants — use this ID directly (e.g., via "
                    f"List Entries) instead of creating a tenant-scoped duplicate."
                )
            else:
                message = (
                    f"Company with name={data.name!r} or domain={data.domain!r} "
                    f"already exists (id={existing.id})"
                )
            raise DuplicateEntityError(
                message,
                entity_type="company",
                existing_id=int(existing.id),
                existing_name=existing.name,
                existing_domain=existing.domain,
                existing_is_global=bool(existing.is_global),
            )

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

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

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

    return Company.model_validate(result)

delete(company_id: CompanyId) -> bool

Delete a company.

Note: Cannot delete global companies.

Source code in affinity/services/companies.py

def delete(self, company_id: CompanyId) -> bool:
    """
    Delete a company.

    Note: Cannot delete global companies.
    """
    result = self._client.delete(f"/organizations/{company_id}", v1=True)
    invalidate_all_list_sizes(self._client)

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

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

get(company_id: CompanyId, *, field_ids: Sequence[AnyFieldId] | None = None, field_types: Sequence[FieldType] | None = None, retries: int = 0, with_interaction_dates: bool = False, with_interaction_persons: bool = False) -> Company

Get a single company by ID.

Parameters:

Name	Type	Description	Default
company_id	CompanyId	The company ID	required
field_ids	Sequence[AnyFieldId] | None	Specific field IDs to include in response	None
field_types	Sequence[FieldType] | None	Field types to include (e.g., ["enriched", "global"])	None
retries	int	Number of retries on 404 NotFoundError. Default is 0 (fail fast). Set to 2-3 if calling immediately after create() to handle eventual consistency lag.	0
with_interaction_dates	bool	Include interaction date summaries (last/next meeting dates, email dates).	False
with_interaction_persons	bool	Include person IDs for each interaction. Only applies when with_interaction_dates=True.	False

Returns:

Type	Description
Company	Company object with requested field data. When with_interaction_dates=True,
Company	the Company will have interaction_dates and interactions populated.

Raises:

Type	Description
NotFoundError	If company does not exist after all retries.

Note

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

Source code in affinity/services/companies.py

def get(
    self,
    company_id: CompanyId,
    *,
    field_ids: Sequence[AnyFieldId] | None = None,
    field_types: Sequence[FieldType] | None = None,
    retries: int = 0,
    with_interaction_dates: bool = False,
    with_interaction_persons: bool = False,
) -> Company:
    """
    Get a single company by ID.

    Args:
        company_id: The company ID
        field_ids: Specific field IDs to include in response
        field_types: Field types to include (e.g., ["enriched", "global"])
        retries: Number of retries on 404 NotFoundError. Default is 0 (fail fast).
            Set to 2-3 if calling immediately after create() to handle eventual
            consistency lag.
        with_interaction_dates: Include interaction date summaries (last/next
            meeting dates, email dates).
        with_interaction_persons: Include person IDs for each interaction.
            Only applies when with_interaction_dates=True.

    Returns:
        Company object with requested field data. When with_interaction_dates=True,
        the Company will have interaction_dates and interactions populated.

    Raises:
        NotFoundError: If company does not exist after all retries.

    Note:
        When combining with_interaction_dates with field_ids/field_types,
        two API calls are made internally and the results are merged.
    """
    last_error: NotFoundError | None = None
    attempts = retries + 1  # retries=0 means 1 attempt
    has_field_filters = field_ids is not None or field_types is not None

    for attempt in range(attempts):
        try:
            if with_interaction_dates:
                # Fetch interaction data
                v1_params: dict[str, Any] = {"with_interaction_dates": True}
                if with_interaction_persons:
                    v1_params["with_interaction_persons"] = True
                interaction_data = self._client.get(
                    f"/organizations/{company_id}",
                    params=v1_params,
                    v1=True,
                )

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

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

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

                # No field filtering, return interaction data directly
                return Company.model_validate(interaction_data)

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

            data = self._client.get(
                f"/companies/{company_id}",
                params=params or None,
            )
            return Company.model_validate(data)
        except NotFoundError as e:
            last_error = e
            if attempt < attempts - 1:  # Don't sleep after last attempt
                time.sleep(0.5 * (attempt + 1))  # 0.5s, 1s, 1.5s backoff

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

    raise last_error  # type: ignore[misc]

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

Get associated opportunity IDs for a company.

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

Parameters:

Name	Type	Description	Default
company_id	CompanyId	The company ID	required
max_results	int | None	Maximum number of opportunity IDs to return	None

Returns:

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

Source code in affinity/services/companies.py

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

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

    Args:
        company_id: The company ID
        max_results: Maximum number of opportunity IDs to return

    Returns:
        List of OpportunityId values associated with this company
    """
    data = self._client.get(f"/organizations/{company_id}", v1=True)
    # Defensive: handle potential {"organization": {...}} wrapper
    organization = data.get("organization") if isinstance(data, dict) else None
    source = organization if isinstance(organization, dict) else data
    opp_ids = None
    if isinstance(source, dict):
        opp_ids = source.get("opportunity_ids") or source.get("opportunityIds")

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

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

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

Get opportunity associations for multiple companies.

Makes one V1 API call per company.

Parameters:

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

Returns:

Type	Description
dict[CompanyId, list[OpportunityId]]	Dict mapping company_id -> list of opportunity_ids

Raises:

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

Source code in affinity/services/companies.py

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

    Makes one V1 API call per company.

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

    Returns:
        Dict mapping company_id -> list of opportunity_ids

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

get_associated_people(company_id: CompanyId, *, max_results: int | None = None) -> builtins.list[Person]

Get Person objects associated with a company.

Uses V2 batch lookup for efficiency (1 API call per 100 persons instead of 1 per person).

Source code in affinity/services/companies.py

def get_associated_people(
    self,
    company_id: CompanyId,
    *,
    max_results: int | None = None,
) -> builtins.list[Person]:
    """
    Get Person objects associated with a company.

    Uses V2 batch lookup for efficiency (1 API call per 100 persons
    instead of 1 per person).
    """
    person_ids = self.get_associated_person_ids(company_id, max_results=max_results)
    if not person_ids:
        return []

    # Use V2 batch lookup: GET /persons?ids=1&ids=2&ids=3
    # Note: person_ids is already truncated by get_associated_person_ids if max_results set
    params: dict[str, Any] = {"ids": [int(pid) for pid in person_ids]}

    people: builtins.list[Person] = []
    data = self._client.get("/persons", params=params)  # V2 batch
    for item in data.get("data", []):
        people.append(Person.model_validate(item))

    # Handle pagination if needed (>100 persons)
    # Note: max_results check is defensive - person_ids was already truncated above
    pagination = data.get("pagination") or {}
    next_url = pagination.get("nextUrl")
    while next_url and (max_results is None or len(people) < max_results):
        data = self._client.get_url(next_url)
        for item in data.get("data", []):
            people.append(Person.model_validate(item))
        next_url = (data.get("pagination") or {}).get("nextUrl")

    if max_results:
        return people[:max_results]
    return people

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

Get associated person IDs for a company.

V1-only exception: V2 does not expose company -> people associations. Uses GET /organizations/{id} and returns person_ids if present.

Source code in affinity/services/companies.py

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

    V1-only exception: V2 does not expose company -> people associations.
    Uses GET `/organizations/{id}` and returns `person_ids` if present.
    """
    data = self._client.get(f"/organizations/{company_id}", v1=True)
    organization = data.get("organization") if isinstance(data, dict) else None
    source = organization if isinstance(organization, dict) else data
    person_ids = None
    if isinstance(source, dict):
        person_ids = source.get("person_ids") or source.get("personIds")

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

    ids = [PersonId(int(value)) for value in person_ids if value is not None]
    if max_results is not None and max_results >= 0:
        return ids[:max_results]
    return ids

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

Get person associations for multiple companies.

Makes one V1 API call per company.

Parameters:

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

Returns:

Type	Description
dict[CompanyId, list[PersonId]]	Dict mapping company_id -> list of person_ids

Raises:

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

Source code in affinity/services/companies.py

def get_associated_person_ids_batch(
    self,
    company_ids: Sequence[CompanyId],
    *,
    on_error: Literal["raise", "skip"] = "raise",
) -> dict[CompanyId, builtins.list[PersonId]]:
    """
    Get person associations for multiple companies.

    Makes one V1 API call per company.

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

    Returns:
        Dict mapping company_id -> list of person_ids

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

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

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

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

Source code in affinity/services/companies.py

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

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

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

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

Parameters:

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

Source code in affinity/services/companies.py

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

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

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

Get metadata about company fields.

Cached for performance.

Parameters:

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

Source code in affinity/services/companies.py

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

    Cached for performance.

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

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

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

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

Get the first company, or None if no results.

This is a convenience method equivalent to

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

Parameters:

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

Returns:

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

Note

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

Source code in affinity/services/companies.py

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

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

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

    Returns:
        First Company, or None if no results.

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

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

Get all list entries for a company across all lists.

Returns comprehensive field data for each list entry. Each entry's entity_id is company_id and its entity_type is EntityType.ORGANIZATION; entity is None (this endpoint doesn't embed the company). A cursor must come from this company's entries.

Source code in affinity/services/companies.py

def get_list_entries(
    self,
    company_id: CompanyId,
    *,
    limit: int | None = None,
    cursor: str | None = None,
) -> PaginatedResponse[ListEntry]:
    """
    Get all list entries for a company across all lists.

    Returns comprehensive field data for each list entry. Each entry's ``entity_id`` is
    ``company_id`` and its ``entity_type`` is ``EntityType.ORGANIZATION``; ``entity`` is
    ``None`` (this endpoint doesn't embed the company). A ``cursor`` must come from this
    company's entries.
    """
    path = f"/companies/{company_id}/list-entries"
    if cursor is not None:
        if limit is not None:
            raise ValueError(
                "Cannot combine 'cursor' with other parameters; cursor encodes all query "
                "context. Start a new pagination sequence without a cursor to change "
                "parameters."
            )
        check_list_entries_cursor(cursor, path)
        data = self._client.get_url(cursor)
    else:
        params: dict[str, Any] = {}
        check_positive_limit(limit)
        if limit is not None:
            params["limit"] = limit
        data = self._client.get(path, params=params or None)

    return entity_list_entries_page(
        data, entity_id=company_id, entity_type=EntityType.ORGANIZATION
    )

get_lists(company_id: CompanyId, *, limit: int | None = None, cursor: str | None = None) -> PaginatedResponse[ListSummary]

Get all lists that contain this company.

Source code in affinity/services/companies.py

def get_lists(
    self,
    company_id: CompanyId,
    *,
    limit: int | None = None,
    cursor: str | None = None,
) -> PaginatedResponse[ListSummary]:
    """Get all lists that contain this company."""
    if cursor is not None:
        if limit is not None:
            raise ValueError(
                "Cannot combine 'cursor' with other parameters; cursor encodes all query "
                "context. Start a new pagination sequence without a cursor to change "
                "parameters."
            )
        data = self._client.get_url(cursor)
    else:
        params: dict[str, Any] = {}
        check_positive_limit(limit)
        if limit is not None:
            params["limit"] = limit
        data = self._client.get(
            f"/companies/{company_id}/lists",
            params=params or None,
        )

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

get_many(company_ids: Sequence[CompanyId], *, field_ids: Sequence[AnyFieldId] | None = None, field_types: Sequence[FieldType] | None = None) -> PaginatedResponse[Company]

Fetch multiple companies by ID in a single API call.

This is a convenience alias for list(ids=[...]).

Parameters:

Name	Type	Description	Default
company_ids	Sequence[CompanyId]	Company IDs to fetch	required
field_ids	Sequence[AnyFieldId] | None	Specific field IDs to include in response	None
field_types	Sequence[FieldType] | None	Field types to include (e.g., ["enriched", "global"])	None

Returns:

Type	Description
PaginatedResponse[Company]	Paginated response with companies

Example

companies = client.companies.get_many([CompanyId(1), CompanyId(2)]) for company in companies.data: ... print(company.name)

Source code in affinity/services/companies.py

def get_many(
    self,
    company_ids: Sequence[CompanyId],
    *,
    field_ids: Sequence[AnyFieldId] | None = None,
    field_types: Sequence[FieldType] | None = None,
) -> PaginatedResponse[Company]:
    """
    Fetch multiple companies by ID in a single API call.

    This is a convenience alias for ``list(ids=[...])``.

    Args:
        company_ids: Company IDs to fetch
        field_ids: Specific field IDs to include in response
        field_types: Field types to include (e.g., ["enriched", "global"])

    Returns:
        Paginated response with companies

    Example:
        >>> companies = client.companies.get_many([CompanyId(1), CompanyId(2)])
        >>> for company in companies.data:
        ...     print(company.name)
    """
    return self.list(ids=company_ids, field_ids=field_ids, field_types=field_types)

get_merge_state(merge_id: int) -> CompanyMergeState

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

Source code in affinity/services/companies.py

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

get_merge_status(task_id: str) -> MergeTask

Check the status of a merge operation.

Source code in affinity/services/companies.py

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

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

Auto-paginate all companies.

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

Note

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

Source code in affinity/services/companies.py

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

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

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

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

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

Source code in affinity/services/companies.py

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

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

    return PageIterator(fetch_page)

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

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

Source code in affinity/services/companies.py

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

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

    return PageIterator(fetch_page)

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

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

Source code in affinity/services/companies.py

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

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

    return PageIterator(fetch_page)

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

Get a page of companies.

Parameters:

Name	Type	Description	Default
ids	Sequence[CompanyId] | None	Specific company IDs to fetch (batch lookup)	None
field_ids	Sequence[AnyFieldId] | None	Specific field IDs to include in response	None
field_types	Sequence[FieldType] | None	Field types to include (e.g., ["enriched", "global"])	None
limit	int | None	Maximum number of results (API default: 100)	None
cursor	str | None	Cursor to resume pagination (opaque; obtained from prior responses)	None

Returns:

Type	Description
PaginatedResponse[Company]	Paginated response with companies

Note

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

Source code in affinity/services/companies.py

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

    Args:
        ids: Specific company IDs to fetch (batch lookup)
        field_ids: Specific field IDs to include in response
        field_types: Field types to include (e.g., ["enriched", "global"])
        limit: Maximum number of results (API default: 100)
        cursor: Cursor to resume pagination (opaque; obtained from prior responses)

    Returns:
        Paginated response with companies

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

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

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

One page of past company merges, newest first.

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

Source code in affinity/services/companies.py

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

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

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

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

Filters: creator_id, created_after (inclusive) / created_before (exclusive), updated_after, rounded outward to whole seconds. limit is 1-100. Needs Affinity API version 2026-07-15 or newer (sent for this call and every page).

Source code in affinity/services/companies.py

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

    Filters: ``creator_id``, ``created_after`` (inclusive) / ``created_before`` (exclusive),
    ``updated_after``, rounded outward to whole seconds. ``limit`` is 1-100.
    Needs Affinity API version 2026-07-15 or newer (sent for this call and every page).
    """
    check_cursor_alone(
        cursor, creator_id, created_after, created_before, updated_after, limit, total_count
    )
    if cursor is not None:
        return page_of(
            NoteV2, self._client.get_url(cursor, min_api_version=COMPANY_NOTES_MIN_API_VERSION)
        )
    params = note_params(
        creator_id=creator_id,
        created_after=created_after,
        created_before=created_before,
        updated_after=updated_after,
        limit=limit,
        total_count=total_count,
    )
    return page_of(
        NoteV2,
        self._client.get(
            f"/companies/{int(company_id)}/notes",
            params=params or None,
            min_api_version=COMPANY_NOTES_MIN_API_VERSION,
        ),
    )

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

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

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

Source code in affinity/services/companies.py

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

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

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

Merge a duplicate company into a primary company.

Returns a task URL to check merge status.

Source code in affinity/services/companies.py

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

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

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

Iterate company pages (not items), yielding PaginatedResponse[Company].

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

Parameters:

Name	Type	Description	Default
ids	Sequence[CompanyId] | None	Specific company IDs to fetch (batch lookup)	None
field_ids	Sequence[AnyFieldId] | None	Specific field IDs to include in response	None
field_types	Sequence[FieldType] | None	Field types to include (e.g., ["enriched", "global"])	None
limit	int | None	Maximum results per page	None
cursor	str | None	Cursor to resume pagination	None

Yields:

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

Source code in affinity/services/companies.py

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

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

    Args:
        ids: Specific company IDs to fetch (batch lookup)
        field_ids: Specific field IDs to include in response
        field_types: Field types to include (e.g., ["enriched", "global"])
        limit: Maximum results per page
        cursor: Cursor to resume pagination

    Yields:
        PaginatedResponse[Company] for each page
    """
    if "filter" in _unsupported:
        raise ValueError(
            "V2 /companies does not support server-side filter. "
            "To search by name/domain use client.companies.search_pages(term). "
            "To filter by list-specific fields use "
            "client.lists.entries(list_id).list(filter=...)."
        )
    if _unsupported:
        raise TypeError(f"Unexpected keyword arguments: {list(_unsupported)}")
    other_params = (ids, field_ids, field_types, limit)
    if cursor is not None and any(p is not None for p in other_params):
        raise ValueError(
            "Cannot combine 'cursor' with other parameters; cursor encodes all query context. "
            "Start a new pagination sequence without a cursor to change parameters."
        )
    requested_cursor = cursor
    guard = _CursorGuard(cursor)
    page = (
        self.list(cursor=cursor)
        if cursor is not None
        else self.list(ids=ids, field_ids=field_ids, field_types=field_types, limit=limit)
    )
    while True:
        yield page
        # Stop at the last page, or when paging can't progress (see _CursorGuard).
        if guard.last_page(requested_cursor, page):
            return
        requested_cursor = page.next_cursor
        page = self.list(cursor=requested_cursor)

resolve(*, domain: str | None = None, name: str | None = None) -> Company | None

Find a single company by domain 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
domain	str | None	Domain to search for (e.g., "acme.com")	None
name	str | None	Company name to search for	None

Returns:

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

Raises:

Type	Description
ValueError	If neither domain nor name is provided

Note

If multiple matches are found, returns the first one. For disambiguation, use search() directly.

Source code in affinity/services/companies.py

def resolve(
    self,
    *,
    domain: str | None = None,
    name: str | None = None,
) -> Company | None:
    """
    Find a single company by domain 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:
        domain: Domain to search for (e.g., "acme.com")
        name: Company name to search for

    Returns:
        The matching Company, or None if not found

    Raises:
        ValueError: If neither domain nor name is provided

    Note:
        If multiple matches are found, returns the first one.
        For disambiguation, use search() directly.
    """
    if not domain and not name:
        raise ValueError("Must provide either domain or name")

    term = domain or name or ""
    result = self.search(term, page_size=10)

    for company in result.data:
        if domain and company.domain and company.domain.lower() == domain.lower():
            return company
        if name and company.name and company.name.lower() == name.lower():
            return company

    return None

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

Search for companies by name or domain.

Uses V1 API for search functionality not available in V2.

Parameters:

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

Returns:

Type	Description
PaginatedResponse[Company]	Dict with 'organizations' and 'next_page_token'

Source code in affinity/services/companies.py

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

    Uses V1 API for search functionality not available in V2.

    Args:
        term: Search term (name or domain)
        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 'organizations' 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("/organizations", params=params, v1=True)
    items = [Company.model_validate(o) for o in data.get("organizations", [])]
    return PaginatedResponse[Company](
        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[Company]

Iterate all V1 company-search results with automatic pagination.

Parameters:

Name	Type	Description	Default
term	str	Search term (name or domain)	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
Company	Company objects matching the search term

Source code in affinity/services/companies.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[Company]:
    """
    Iterate all V1 company-search results with automatic pagination.

    Args:
        term: Search term (name or domain)
        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:
        Company objects matching the search term
    """
    for page in self.search_pages(
        term,
        with_interaction_dates=with_interaction_dates,
        with_interaction_persons=with_interaction_persons,
        with_opportunities=with_opportunities,
        page_size=page_size,
        page_token=page_token,
    ):
        yield from page.data

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

Iterate V1 company-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 domain)	required
with_interaction_dates	bool	Include interaction date data	False
with_interaction_persons	bool	Include persons for interactions	False
with_opportunities	bool	Include associated opportunity IDs	False
page_size	int | None	Results per page (max 500)	None
page_token	str | None	Resume from this pagination token	None

Yields:

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

Source code in affinity/services/companies.py

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

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

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

    Yields:
        PaginatedResponse[Company] 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(company_id: CompanyId, data: CompanyUpdate) -> Company

Update an existing company.

Note: Cannot update name/domain of global companies.

Source code in affinity/services/companies.py

def update(
    self,
    company_id: CompanyId,
    data: CompanyUpdate,
) -> Company:
    """
    Update an existing company.

    Note: Cannot update name/domain of global companies.
    """
    payload = data.model_dump(
        by_alias=True,
        mode="json",
        exclude_unset=True,
        exclude_none=True,
    )

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

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

    return Company.model_validate(result)

Bases: AsyncCompanySemanticSearchMixin, AsyncCompanyDiscoveryMixin, AsyncSingleFieldMixin, AsyncSearchMixin[Company]

Source code in affinity/services/companies.py

class AsyncCompanyService(
    AsyncCompanySemanticSearchMixin,
    AsyncCompanyDiscoveryMixin,
    AsyncSingleFieldMixin,
    AsyncSearchMixin[Company],
):
    _entity_path = "/companies"

    _search_model = Company

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

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

    """
    Async version of CompanyService.

    Mirrors sync behavior for V2 reads, V1 writes, and V1 search helpers.
    """

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

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

        Args:
            ids: Specific company IDs to fetch (batch lookup)
            field_ids: Specific field IDs to include in response
            field_types: Field types to include (e.g., ["enriched", "global"])
            limit: Maximum number of results (API default: 100)
            cursor: Cursor to resume pagination (opaque; obtained from prior responses)

        Returns:
            Paginated response with companies

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

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

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

        See CompanyService.get_first() for details.

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

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

        Unlike get_many() which uses the API's batch endpoint, this method
        makes individual get() calls with bounded concurrency. Use this when
        you need parameters not supported by the batch endpoint (e.g.,
        with_interaction_dates=True).

        Calls: one per id. With with_interaction_dates and field_ids/field_types, the fields
        come from one call per 100 ids and the dates from one V1 call per id (about N + N/100
        calls, not 2N). For just the last and next event dates of many companies, the
        relationship-intelligence fields are cheaper: ``iter(ids=<up to 100 ids>,
        field_types=[FieldType.RELATIONSHIP_INTELLIGENCE])`` reads them in one call per 100
        ids, without the V1 calls (the other interaction dates and interaction persons have
        no confirmed V2 counterpart).

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

        Returns:
            Dict mapping company_id -> Company for successfully fetched companies.

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

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

        async def fetch_one(cid: CompanyId) -> tuple[CompanyId, Company | None]:
            try:
                company = await self.get(
                    cid,
                    field_ids=field_ids,
                    field_types=field_types,
                    with_interaction_dates=with_interaction_dates,
                    with_interaction_persons=with_interaction_persons,
                )
                return (cid, company)
            except AffinityError:
                if on_error == "raise":
                    raise
                return (cid, None)

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

        return results

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

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

        Args:
            ids: Specific company IDs to fetch (batch lookup)
            field_ids: Specific field IDs to include in response
            field_types: Field types to include (e.g., ["enriched", "global"])
            limit: Maximum results per page
            cursor: Cursor to resume pagination

        Returns:
            AsyncIterator of PaginatedResponse[Company]

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

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

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

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

        Yields:
            Company objects

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

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

        return AsyncPageIterator(fetch_page)

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

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

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

    async def get(
        self,
        company_id: CompanyId,
        *,
        field_ids: Sequence[AnyFieldId] | None = None,
        field_types: Sequence[FieldType] | None = None,
        retries: int = 0,
        with_interaction_dates: bool = False,
        with_interaction_persons: bool = False,
    ) -> Company:
        """
        Get a single company by ID.

        Args:
            company_id: The company ID
            field_ids: Specific field IDs to include in response
            field_types: Field types to include (e.g., ["enriched", "global"])
            retries: Number of retries on 404 NotFoundError. Default is 0 (fail fast).
                Set to 2-3 if calling immediately after create() to handle eventual
                consistency lag.
            with_interaction_dates: Include interaction date summaries (last/next
                meeting dates, email dates).
            with_interaction_persons: Include person IDs for each interaction.
                Only applies when with_interaction_dates=True.

        Returns:
            Company object with requested field data. When with_interaction_dates=True,
            the Company will have interaction_dates and interactions populated.

        Raises:
            NotFoundError: If company does not exist after all retries.

        Note:
            When combining with_interaction_dates with field_ids/field_types,
            two API calls are made internally and the results are merged.
        """
        last_error: NotFoundError | None = None
        attempts = retries + 1  # retries=0 means 1 attempt
        has_field_filters = field_ids is not None or field_types is not None

        for attempt in range(attempts):
            try:
                if with_interaction_dates:
                    # Fetch interaction data
                    v1_params: dict[str, Any] = {"with_interaction_dates": True}
                    if with_interaction_persons:
                        v1_params["with_interaction_persons"] = True
                    interaction_data = await self._client.get(
                        f"/organizations/{company_id}",
                        params=v1_params,
                        v1=True,
                    )

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

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

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

                    # No field filtering, return interaction data directly
                    return Company.model_validate(interaction_data)

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

                data = await self._client.get(f"/companies/{company_id}", params=params or None)
                return Company.model_validate(data)
            except NotFoundError as e:
                last_error = e
                if attempt < attempts - 1:  # Don't sleep after last attempt
                    await asyncio.sleep(0.5 * (attempt + 1))  # 0.5s, 1s, 1.5s backoff

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

        raise last_error  # type: ignore[misc]

    async def get_many(
        self,
        company_ids: Sequence[CompanyId],
        *,
        field_ids: Sequence[AnyFieldId] | None = None,
        field_types: Sequence[FieldType] | None = None,
    ) -> PaginatedResponse[Company]:
        """
        Fetch multiple companies by ID in a single API call.

        This is a convenience alias for ``list(ids=[...])``.

        Args:
            company_ids: Company IDs to fetch
            field_ids: Specific field IDs to include in response
            field_types: Field types to include (e.g., ["enriched", "global"])

        Returns:
            Paginated response with companies

        Example:
            >>> companies = await client.companies.get_many([CompanyId(1), CompanyId(2)])
            >>> for company in companies.data:
            ...     print(company.name)
        """
        return await self.list(ids=company_ids, field_ids=field_ids, field_types=field_types)

    async def get_list_entries(
        self,
        company_id: CompanyId,
        *,
        limit: int | None = None,
        cursor: str | None = None,
    ) -> PaginatedResponse[ListEntry]:
        """
        Get all list entries for a company across all lists.

        Returns comprehensive field data for each list entry. Each entry's ``entity_id`` is
        ``company_id`` and its ``entity_type`` is ``EntityType.ORGANIZATION``; ``entity`` is
        ``None`` (this endpoint doesn't embed the company). A ``cursor`` must come from this
        company's entries.
        """
        path = f"/companies/{company_id}/list-entries"
        if cursor is not None:
            if limit is not None:
                raise ValueError(
                    "Cannot combine 'cursor' with other parameters; cursor encodes all query "
                    "context. Start a new pagination sequence without a cursor to change "
                    "parameters."
                )
            check_list_entries_cursor(cursor, path)
            data = await self._client.get_url(cursor)
        else:
            params: dict[str, Any] = {}
            check_positive_limit(limit)
            if limit is not None:
                params["limit"] = limit
            data = await self._client.get(path, params=params or None)

        return entity_list_entries_page(
            data, entity_id=company_id, entity_type=EntityType.ORGANIZATION
        )

    async def get_lists(
        self,
        company_id: CompanyId,
        *,
        limit: int | None = None,
        cursor: str | None = None,
    ) -> PaginatedResponse[ListSummary]:
        """Get all lists that contain this company."""
        if cursor is not None:
            if limit is not None:
                raise ValueError(
                    "Cannot combine 'cursor' with other parameters; cursor encodes all query "
                    "context. Start a new pagination sequence without a cursor to change "
                    "parameters."
                )
            data = await self._client.get_url(cursor)
        else:
            params: dict[str, Any] = {}
            check_positive_limit(limit)
            if limit is not None:
                params["limit"] = limit
            data = await self._client.get(
                f"/companies/{company_id}/lists",
                params=params or None,
            )

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

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

        Cached for performance.

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

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

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

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

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

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

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

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

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

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

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

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

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

        Uses V1 API for search functionality not available in V2.
        """
        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("/organizations", params=params, v1=True)
        items = [Company.model_validate(o) for o in data.get("organizations", [])]
        return PaginatedResponse[Company](
            data=items,
            next_page_token=data.get("next_page_token"),
        )

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

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

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

        Yields:
            PaginatedResponse[Company] 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[Company]:
        """
        Iterate all V1 company-search results with automatic pagination.

        Args:
            term: Search term (name or domain)
            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:
            Company 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 company in page.data:
                yield company

    async def resolve(
        self,
        *,
        domain: str | None = None,
        name: str | None = None,
    ) -> Company | None:
        """
        Find a single company by domain 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 domain and not name:
            raise ValueError("Must provide either domain or name")

        term = domain or name or ""
        result = await self.search(term, page_size=10)

        for company in result.data:
            if domain and company.domain and company.domain.lower() == domain.lower():
                return company
            if name and company.name and company.name.lower() == name.lower():
                return company

        return None

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

        V1-only exception: V2 does not expose company -> people associations.
        Uses GET `/organizations/{id}` and returns `person_ids` if present.
        """
        data = await self._client.get(f"/organizations/{company_id}", v1=True)
        organization = data.get("organization") if isinstance(data, dict) else None
        source = organization if isinstance(organization, dict) else data
        person_ids = None
        if isinstance(source, dict):
            person_ids = source.get("person_ids") or source.get("personIds")

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

        ids = [PersonId(int(value)) for value in person_ids if value is not None]
        if max_results is not None and max_results >= 0:
            return ids[:max_results]
        return ids

    async def get_associated_person_ids_batch(
        self,
        company_ids: Sequence[CompanyId],
        *,
        on_error: Literal["raise", "skip"] = "raise",
    ) -> dict[CompanyId, builtins.list[PersonId]]:
        """
        Get person associations for multiple companies.

        Makes one V1 API call per company.

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

        Returns:
            Dict mapping company_id -> list of person_ids

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

    async def get_associated_people(
        self,
        company_id: CompanyId,
        *,
        max_results: int | None = None,
    ) -> builtins.list[Person]:
        """
        Get Person objects associated with a company.

        Uses V2 batch lookup for efficiency (1 API call per 100 persons
        instead of 1 per person).
        """
        person_ids = await self.get_associated_person_ids(company_id, max_results=max_results)
        if not person_ids:
            return []

        # Use V2 batch lookup: GET /persons?ids=1&ids=2&ids=3
        # Note: person_ids is already truncated by get_associated_person_ids if max_results set
        params: dict[str, Any] = {"ids": [int(pid) for pid in person_ids]}

        people: builtins.list[Person] = []
        data = await self._client.get("/persons", params=params)  # V2 batch
        for item in data.get("data", []):
            people.append(Person.model_validate(item))

        # Handle pagination if needed (>100 persons)
        # Note: max_results check is defensive - person_ids was already truncated above
        pagination = data.get("pagination") or {}
        next_url = pagination.get("nextUrl")
        while next_url and (max_results is None or len(people) < max_results):
            data = await self._client.get_url(next_url)
            for item in data.get("data", []):
                people.append(Person.model_validate(item))
            next_url = (data.get("pagination") or {}).get("nextUrl")

        if max_results:
            return people[:max_results]
        return people

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

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

        Args:
            company_id: The company ID
            max_results: Maximum number of opportunity IDs to return

        Returns:
            List of OpportunityId values associated with this company
        """
        data = await self._client.get(f"/organizations/{company_id}", v1=True)
        # Defensive: handle potential {"organization": {...}} wrapper
        organization = data.get("organization") if isinstance(data, dict) else None
        source = organization if isinstance(organization, dict) else data
        opp_ids = None
        if isinstance(source, dict):
            opp_ids = source.get("opportunity_ids") or source.get("opportunityIds")

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

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

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

        Makes one V1 API call per company.

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

        Returns:
            Dict mapping company_id -> list of opportunity_ids

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

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

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

    async def create(self, data: CompanyCreate, *, if_not_exists: bool = True) -> Company:
        """
        Create a new company.

        Args:
            data: Company creation data
            if_not_exists: If True (default), check for an existing company with
                the same name (case-insensitive exact match) OR the same domain
                (matched against the company's primary domain AND all entries
                in its `domains` list). Globals in Affinity's shared directory
                are treated as duplicates — per the API docs, POST /organizations
                always creates a tenant-scoped copy, so creating against an
                existing global would produce a genuine duplicate. Raises
                DuplicateEntityError carrying the existing company's ID and
                `existing_is_global` flag, so callers can use the global ID
                directly (e.g., via List Entries) instead of POSTing a copy.
                Uses V1 search_pages() with page_size=500, capped at 3 pages.
                Set to False to skip the check and create unconditionally.

        Returns:
            Created company

        Raises:
            DuplicateEntityError: When if_not_exists=True and an exact-name or
                exact-domain match already exists (tenant-scoped OR global).

        Note:
            Creates use V1 API, while reads use V2 API. Due to eventual consistency
            between V1 and V2, a `get()` call immediately after `create()` may return
            404 NotFoundError. If you need to read immediately after creation, either:
            - Use the Company object returned by this method (it contains the created data)
            - Add a short delay (100-500ms) before calling get()
            - Implement retry logic in your application
        """
        if if_not_exists:
            existing = await self._find_exact_duplicate(name=data.name, domain=data.domain)
            if existing is not None:
                if existing.is_global:
                    message = (
                        f"A global Affinity directory record matches name={data.name!r} "
                        f"or domain={data.domain!r} (id={existing.id}). Global records "
                        f"are shared across tenants — use this ID directly (e.g., via "
                        f"List Entries) instead of creating a tenant-scoped duplicate."
                    )
                else:
                    message = (
                        f"Company with name={data.name!r} or domain={data.domain!r} "
                        f"already exists (id={existing.id})"
                    )
                raise DuplicateEntityError(
                    message,
                    entity_type="company",
                    existing_id=int(existing.id),
                    existing_name=existing.name,
                    existing_domain=existing.domain,
                    existing_is_global=bool(existing.is_global),
                )

        payload = data.model_dump(by_alias=True, mode="json", exclude_none=True)
        if not data.person_ids:
            payload.pop("person_ids", None)
        result = await self._client.post("/organizations", json=payload, v1=True)

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

        return Company.model_validate(result)

    @staticmethod
    def _company_matches(company: Company, name_lower: str, domain_lower: str | None) -> bool:
        """True iff company's name or any domain matches exactly (case-insensitive)."""
        if company.name and company.name.lower() == name_lower:
            return True
        if domain_lower:
            if company.domain and company.domain.lower() == domain_lower:
                return True
            for d in company.domains or []:
                if d and d.lower() == domain_lower:
                    return True
        return False

    async def _find_exact_duplicate(self, *, name: str, domain: str | None) -> Company | None:
        """Search V1 for an exact-name or exact-domain match. Returns None if no match.

        Domain-first search order: domains are globally more unique than names,
        so if a domain is supplied, we search by domain first. Falls back to a
        name search only when the domain search came up empty and the name
        differs from the domain string.

        Iterates at most `_DEDUP_PAGE_CAP` pages per term (1500 fuzzy results).
        """
        name_lower = name.strip().lower()
        domain_lower = domain.strip().lower() if domain else None

        async def _scan(term: str) -> Company | None:
            page_num = 0
            async for page in self.search_pages(term, page_size=500):
                for company in page.data:
                    if self._company_matches(company, name_lower, domain_lower):
                        return company
                if page_num + 1 >= self._DEDUP_PAGE_CAP:
                    break
                if not page.next_page_token:
                    break
                page_num += 1
            return None

        if domain_lower:
            # domain is guaranteed non-None when domain_lower is set
            hit = await _scan(domain or "")
            if hit is not None:
                return hit
            if name_lower and name_lower != domain_lower:
                return await _scan(name)
            return None

        return await _scan(name)

    async def update(self, company_id: CompanyId, data: CompanyUpdate) -> Company:
        """
        Update an existing company.

        Uses V1 API.
        """
        payload = data.model_dump(
            by_alias=True,
            mode="json",
            exclude_unset=True,
            exclude_none=True,
        )
        result = await self._client.put(
            f"/organizations/{company_id}",
            json=payload,
            v1=True,
        )

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

        return Company.model_validate(result)

    async def delete(self, company_id: CompanyId) -> bool:
        """
        Delete a company.

        Uses V1 API.
        """
        result = await self._client.delete(f"/organizations/{company_id}", v1=True)
        invalidate_all_list_sizes(self._client)

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

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

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

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

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

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

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

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

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

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

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

        return AsyncPageIterator(fetch_page)

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

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

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

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

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

        return AsyncPageIterator(fetch_page)

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

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

        Filters: ``creator_id``, ``created_after`` (inclusive) / ``created_before`` (exclusive),
        ``updated_after``, rounded outward to whole seconds. ``limit`` is 1-100.
        Needs Affinity API version 2026-07-15 or newer (sent for this call and every page).
        """
        check_cursor_alone(
            cursor, creator_id, created_after, created_before, updated_after, limit, total_count
        )
        if cursor is not None:
            return page_of(
                NoteV2,
                await self._client.get_url(cursor, min_api_version=COMPANY_NOTES_MIN_API_VERSION),
            )
        params = note_params(
            creator_id=creator_id,
            created_after=created_after,
            created_before=created_before,
            updated_after=updated_after,
            limit=limit,
            total_count=total_count,
        )
        return page_of(
            NoteV2,
            await self._client.get(
                f"/companies/{int(company_id)}/notes",
                params=params or None,
                min_api_version=COMPANY_NOTES_MIN_API_VERSION,
            ),
        )

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

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

        return AsyncPageIterator(fetch_page)

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

Iterate through all companies with automatic pagination.

Parameters:

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

Yields:

Type	Description
AsyncIterator[Company]	Company objects

Note

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

Source code in affinity/services/companies.py

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

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

    Yields:
        Company objects

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

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

    return AsyncPageIterator(fetch_page)

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

Fetch multiple companies with controlled concurrency.

Unlike get_many() which uses the API's batch endpoint, this method makes individual get() calls with bounded concurrency. Use this when you need parameters not supported by the batch endpoint (e.g., with_interaction_dates=True).

Calls: one per id. With with_interaction_dates and field_ids/field_types, the fields come from one call per 100 ids and the dates from one V1 call per id (about N + N/100 calls, not 2N). For just the last and next event dates of many companies, the relationship-intelligence fields are cheaper: iter(ids=<up to 100 ids>, field_types=[FieldType.RELATIONSHIP_INTELLIGENCE]) reads them in one call per 100 ids, without the V1 calls (the other interaction dates and interaction persons have no confirmed V2 counterpart).

Parameters:

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

Returns:

Type	Description
dict[CompanyId, Company]	Dict mapping company_id -> Company for successfully fetched companies.

Raises:

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

Source code in affinity/services/companies.py

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

    Unlike get_many() which uses the API's batch endpoint, this method
    makes individual get() calls with bounded concurrency. Use this when
    you need parameters not supported by the batch endpoint (e.g.,
    with_interaction_dates=True).

    Calls: one per id. With with_interaction_dates and field_ids/field_types, the fields
    come from one call per 100 ids and the dates from one V1 call per id (about N + N/100
    calls, not 2N). For just the last and next event dates of many companies, the
    relationship-intelligence fields are cheaper: ``iter(ids=<up to 100 ids>,
    field_types=[FieldType.RELATIONSHIP_INTELLIGENCE])`` reads them in one call per 100
    ids, without the V1 calls (the other interaction dates and interaction persons have
    no confirmed V2 counterpart).

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

    Returns:
        Dict mapping company_id -> Company for successfully fetched companies.

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

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

    async def fetch_one(cid: CompanyId) -> tuple[CompanyId, Company | None]:
        try:
            company = await self.get(
                cid,
                field_ids=field_ids,
                field_types=field_types,
                with_interaction_dates=with_interaction_dates,
                with_interaction_persons=with_interaction_persons,
            )
            return (cid, company)
        except AffinityError:
            if on_error == "raise":
                raise
            return (cid, None)

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

    return results

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

Update several field values of one company in one request.

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

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

Parameters:

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

Source code in affinity/services/companies.py

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

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

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

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

create(data: CompanyCreate, *, if_not_exists: bool = True) -> Company async

Create a new company.

Parameters:

Name	Type	Description	Default
data	CompanyCreate	Company creation data	required
if_not_exists	bool	If True (default), check for an existing company with the same name (case-insensitive exact match) OR the same domain (matched against the company's primary domain AND all entries in its domains list). Globals in Affinity's shared directory are treated as duplicates — per the API docs, POST /organizations always creates a tenant-scoped copy, so creating against an existing global would produce a genuine duplicate. Raises DuplicateEntityError carrying the existing company's ID and existing_is_global flag, so callers can use the global ID directly (e.g., via List Entries) instead of POSTing a copy. Uses V1 search_pages() with page_size=500, capped at 3 pages. Set to False to skip the check and create unconditionally.	True

Returns:

Type	Description
Company	Created company

Raises:

Type	Description
DuplicateEntityError	When if_not_exists=True and an exact-name or exact-domain match already exists (tenant-scoped OR global).

Note

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

Source code in affinity/services/companies.py

async def create(self, data: CompanyCreate, *, if_not_exists: bool = True) -> Company:
    """
    Create a new company.

    Args:
        data: Company creation data
        if_not_exists: If True (default), check for an existing company with
            the same name (case-insensitive exact match) OR the same domain
            (matched against the company's primary domain AND all entries
            in its `domains` list). Globals in Affinity's shared directory
            are treated as duplicates — per the API docs, POST /organizations
            always creates a tenant-scoped copy, so creating against an
            existing global would produce a genuine duplicate. Raises
            DuplicateEntityError carrying the existing company's ID and
            `existing_is_global` flag, so callers can use the global ID
            directly (e.g., via List Entries) instead of POSTing a copy.
            Uses V1 search_pages() with page_size=500, capped at 3 pages.
            Set to False to skip the check and create unconditionally.

    Returns:
        Created company

    Raises:
        DuplicateEntityError: When if_not_exists=True and an exact-name or
            exact-domain match already exists (tenant-scoped OR global).

    Note:
        Creates use V1 API, while reads use V2 API. Due to eventual consistency
        between V1 and V2, a `get()` call immediately after `create()` may return
        404 NotFoundError. If you need to read immediately after creation, either:
        - Use the Company object returned by this method (it contains the created data)
        - Add a short delay (100-500ms) before calling get()
        - Implement retry logic in your application
    """
    if if_not_exists:
        existing = await self._find_exact_duplicate(name=data.name, domain=data.domain)
        if existing is not None:
            if existing.is_global:
                message = (
                    f"A global Affinity directory record matches name={data.name!r} "
                    f"or domain={data.domain!r} (id={existing.id}). Global records "
                    f"are shared across tenants — use this ID directly (e.g., via "
                    f"List Entries) instead of creating a tenant-scoped duplicate."
                )
            else:
                message = (
                    f"Company with name={data.name!r} or domain={data.domain!r} "
                    f"already exists (id={existing.id})"
                )
            raise DuplicateEntityError(
                message,
                entity_type="company",
                existing_id=int(existing.id),
                existing_name=existing.name,
                existing_domain=existing.domain,
                existing_is_global=bool(existing.is_global),
            )

    payload = data.model_dump(by_alias=True, mode="json", exclude_none=True)
    if not data.person_ids:
        payload.pop("person_ids", None)
    result = await self._client.post("/organizations", json=payload, v1=True)

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

    return Company.model_validate(result)

delete(company_id: CompanyId) -> bool async

Delete a company.

Uses V1 API.

Source code in affinity/services/companies.py

async def delete(self, company_id: CompanyId) -> bool:
    """
    Delete a company.

    Uses V1 API.
    """
    result = await self._client.delete(f"/organizations/{company_id}", v1=True)
    invalidate_all_list_sizes(self._client)

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

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

get(company_id: CompanyId, *, field_ids: Sequence[AnyFieldId] | None = None, field_types: Sequence[FieldType] | None = None, retries: int = 0, with_interaction_dates: bool = False, with_interaction_persons: bool = False) -> Company async

Get a single company by ID.

Parameters:

Name	Type	Description	Default
company_id	CompanyId	The company ID	required
field_ids	Sequence[AnyFieldId] | None	Specific field IDs to include in response	None
field_types	Sequence[FieldType] | None	Field types to include (e.g., ["enriched", "global"])	None
retries	int	Number of retries on 404 NotFoundError. Default is 0 (fail fast). Set to 2-3 if calling immediately after create() to handle eventual consistency lag.	0
with_interaction_dates	bool	Include interaction date summaries (last/next meeting dates, email dates).	False
with_interaction_persons	bool	Include person IDs for each interaction. Only applies when with_interaction_dates=True.	False

Returns:

Type	Description
Company	Company object with requested field data. When with_interaction_dates=True,
Company	the Company will have interaction_dates and interactions populated.

Raises:

Type	Description
NotFoundError	If company does not exist after all retries.

Note

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

Source code in affinity/services/companies.py

async def get(
    self,
    company_id: CompanyId,
    *,
    field_ids: Sequence[AnyFieldId] | None = None,
    field_types: Sequence[FieldType] | None = None,
    retries: int = 0,
    with_interaction_dates: bool = False,
    with_interaction_persons: bool = False,
) -> Company:
    """
    Get a single company by ID.

    Args:
        company_id: The company ID
        field_ids: Specific field IDs to include in response
        field_types: Field types to include (e.g., ["enriched", "global"])
        retries: Number of retries on 404 NotFoundError. Default is 0 (fail fast).
            Set to 2-3 if calling immediately after create() to handle eventual
            consistency lag.
        with_interaction_dates: Include interaction date summaries (last/next
            meeting dates, email dates).
        with_interaction_persons: Include person IDs for each interaction.
            Only applies when with_interaction_dates=True.

    Returns:
        Company object with requested field data. When with_interaction_dates=True,
        the Company will have interaction_dates and interactions populated.

    Raises:
        NotFoundError: If company does not exist after all retries.

    Note:
        When combining with_interaction_dates with field_ids/field_types,
        two API calls are made internally and the results are merged.
    """
    last_error: NotFoundError | None = None
    attempts = retries + 1  # retries=0 means 1 attempt
    has_field_filters = field_ids is not None or field_types is not None

    for attempt in range(attempts):
        try:
            if with_interaction_dates:
                # Fetch interaction data
                v1_params: dict[str, Any] = {"with_interaction_dates": True}
                if with_interaction_persons:
                    v1_params["with_interaction_persons"] = True
                interaction_data = await self._client.get(
                    f"/organizations/{company_id}",
                    params=v1_params,
                    v1=True,
                )

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

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

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

                # No field filtering, return interaction data directly
                return Company.model_validate(interaction_data)

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

            data = await self._client.get(f"/companies/{company_id}", params=params or None)
            return Company.model_validate(data)
        except NotFoundError as e:
            last_error = e
            if attempt < attempts - 1:  # Don't sleep after last attempt
                await asyncio.sleep(0.5 * (attempt + 1))  # 0.5s, 1s, 1.5s backoff

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

    raise last_error  # type: ignore[misc]

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

Get associated opportunity IDs for a company.

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

Parameters:

Name	Type	Description	Default
company_id	CompanyId	The company ID	required
max_results	int | None	Maximum number of opportunity IDs to return	None

Returns:

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

Source code in affinity/services/companies.py

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

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

    Args:
        company_id: The company ID
        max_results: Maximum number of opportunity IDs to return

    Returns:
        List of OpportunityId values associated with this company
    """
    data = await self._client.get(f"/organizations/{company_id}", v1=True)
    # Defensive: handle potential {"organization": {...}} wrapper
    organization = data.get("organization") if isinstance(data, dict) else None
    source = organization if isinstance(organization, dict) else data
    opp_ids = None
    if isinstance(source, dict):
        opp_ids = source.get("opportunity_ids") or source.get("opportunityIds")

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

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

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

Get opportunity associations for multiple companies.

Makes one V1 API call per company.

Parameters:

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

Returns:

Type	Description
dict[CompanyId, list[OpportunityId]]	Dict mapping company_id -> list of opportunity_ids

Raises:

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

Source code in affinity/services/companies.py

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

    Makes one V1 API call per company.

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

    Returns:
        Dict mapping company_id -> list of opportunity_ids

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

get_associated_people(company_id: CompanyId, *, max_results: int | None = None) -> builtins.list[Person] async

Get Person objects associated with a company.

Uses V2 batch lookup for efficiency (1 API call per 100 persons instead of 1 per person).

Source code in affinity/services/companies.py

async def get_associated_people(
    self,
    company_id: CompanyId,
    *,
    max_results: int | None = None,
) -> builtins.list[Person]:
    """
    Get Person objects associated with a company.

    Uses V2 batch lookup for efficiency (1 API call per 100 persons
    instead of 1 per person).
    """
    person_ids = await self.get_associated_person_ids(company_id, max_results=max_results)
    if not person_ids:
        return []

    # Use V2 batch lookup: GET /persons?ids=1&ids=2&ids=3
    # Note: person_ids is already truncated by get_associated_person_ids if max_results set
    params: dict[str, Any] = {"ids": [int(pid) for pid in person_ids]}

    people: builtins.list[Person] = []
    data = await self._client.get("/persons", params=params)  # V2 batch
    for item in data.get("data", []):
        people.append(Person.model_validate(item))

    # Handle pagination if needed (>100 persons)
    # Note: max_results check is defensive - person_ids was already truncated above
    pagination = data.get("pagination") or {}
    next_url = pagination.get("nextUrl")
    while next_url and (max_results is None or len(people) < max_results):
        data = await self._client.get_url(next_url)
        for item in data.get("data", []):
            people.append(Person.model_validate(item))
        next_url = (data.get("pagination") or {}).get("nextUrl")

    if max_results:
        return people[:max_results]
    return people

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

Get associated person IDs for a company.

V1-only exception: V2 does not expose company -> people associations. Uses GET /organizations/{id} and returns person_ids if present.

Source code in affinity/services/companies.py

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

    V1-only exception: V2 does not expose company -> people associations.
    Uses GET `/organizations/{id}` and returns `person_ids` if present.
    """
    data = await self._client.get(f"/organizations/{company_id}", v1=True)
    organization = data.get("organization") if isinstance(data, dict) else None
    source = organization if isinstance(organization, dict) else data
    person_ids = None
    if isinstance(source, dict):
        person_ids = source.get("person_ids") or source.get("personIds")

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

    ids = [PersonId(int(value)) for value in person_ids if value is not None]
    if max_results is not None and max_results >= 0:
        return ids[:max_results]
    return ids

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

Get person associations for multiple companies.

Makes one V1 API call per company.

Parameters:

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

Returns:

Type	Description
dict[CompanyId, list[PersonId]]	Dict mapping company_id -> list of person_ids

Raises:

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

Source code in affinity/services/companies.py

async def get_associated_person_ids_batch(
    self,
    company_ids: Sequence[CompanyId],
    *,
    on_error: Literal["raise", "skip"] = "raise",
) -> dict[CompanyId, builtins.list[PersonId]]:
    """
    Get person associations for multiple companies.

    Makes one V1 API call per company.

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

    Returns:
        Dict mapping company_id -> list of person_ids

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

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

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

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

Source code in affinity/services/companies.py

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

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

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

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

Parameters:

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

Source code in affinity/services/companies.py

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

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

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

Get metadata about company fields.

Cached for performance.

Parameters:

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

Source code in affinity/services/companies.py

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

    Cached for performance.

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

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

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

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

Get the first company from the list endpoint, or None if no results.

See CompanyService.get_first() for details.

Note

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

Source code in affinity/services/companies.py

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

    See CompanyService.get_first() for details.

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

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

Get all list entries for a company across all lists.

Returns comprehensive field data for each list entry. Each entry's entity_id is company_id and its entity_type is EntityType.ORGANIZATION; entity is None (this endpoint doesn't embed the company). A cursor must come from this company's entries.

Source code in affinity/services/companies.py

async def get_list_entries(
    self,
    company_id: CompanyId,
    *,
    limit: int | None = None,
    cursor: str | None = None,
) -> PaginatedResponse[ListEntry]:
    """
    Get all list entries for a company across all lists.

    Returns comprehensive field data for each list entry. Each entry's ``entity_id`` is
    ``company_id`` and its ``entity_type`` is ``EntityType.ORGANIZATION``; ``entity`` is
    ``None`` (this endpoint doesn't embed the company). A ``cursor`` must come from this
    company's entries.
    """
    path = f"/companies/{company_id}/list-entries"
    if cursor is not None:
        if limit is not None:
            raise ValueError(
                "Cannot combine 'cursor' with other parameters; cursor encodes all query "
                "context. Start a new pagination sequence without a cursor to change "
                "parameters."
            )
        check_list_entries_cursor(cursor, path)
        data = await self._client.get_url(cursor)
    else:
        params: dict[str, Any] = {}
        check_positive_limit(limit)
        if limit is not None:
            params["limit"] = limit
        data = await self._client.get(path, params=params or None)

    return entity_list_entries_page(
        data, entity_id=company_id, entity_type=EntityType.ORGANIZATION
    )

get_lists(company_id: CompanyId, *, limit: int | None = None, cursor: str | None = None) -> PaginatedResponse[ListSummary] async

Get all lists that contain this company.

Source code in affinity/services/companies.py

async def get_lists(
    self,
    company_id: CompanyId,
    *,
    limit: int | None = None,
    cursor: str | None = None,
) -> PaginatedResponse[ListSummary]:
    """Get all lists that contain this company."""
    if cursor is not None:
        if limit is not None:
            raise ValueError(
                "Cannot combine 'cursor' with other parameters; cursor encodes all query "
                "context. Start a new pagination sequence without a cursor to change "
                "parameters."
            )
        data = await self._client.get_url(cursor)
    else:
        params: dict[str, Any] = {}
        check_positive_limit(limit)
        if limit is not None:
            params["limit"] = limit
        data = await self._client.get(
            f"/companies/{company_id}/lists",
            params=params or None,
        )

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

get_many(company_ids: Sequence[CompanyId], *, field_ids: Sequence[AnyFieldId] | None = None, field_types: Sequence[FieldType] | None = None) -> PaginatedResponse[Company] async

Fetch multiple companies by ID in a single API call.

This is a convenience alias for list(ids=[...]).

Parameters:

Name	Type	Description	Default
company_ids	Sequence[CompanyId]	Company IDs to fetch	required
field_ids	Sequence[AnyFieldId] | None	Specific field IDs to include in response	None
field_types	Sequence[FieldType] | None	Field types to include (e.g., ["enriched", "global"])	None

Returns:

Type	Description
PaginatedResponse[Company]	Paginated response with companies

Example

companies = await client.companies.get_many([CompanyId(1), CompanyId(2)]) for company in companies.data: ... print(company.name)

Source code in affinity/services/companies.py

async def get_many(
    self,
    company_ids: Sequence[CompanyId],
    *,
    field_ids: Sequence[AnyFieldId] | None = None,
    field_types: Sequence[FieldType] | None = None,
) -> PaginatedResponse[Company]:
    """
    Fetch multiple companies by ID in a single API call.

    This is a convenience alias for ``list(ids=[...])``.

    Args:
        company_ids: Company IDs to fetch
        field_ids: Specific field IDs to include in response
        field_types: Field types to include (e.g., ["enriched", "global"])

    Returns:
        Paginated response with companies

    Example:
        >>> companies = await client.companies.get_many([CompanyId(1), CompanyId(2)])
        >>> for company in companies.data:
        ...     print(company.name)
    """
    return await self.list(ids=company_ids, field_ids=field_ids, field_types=field_types)

get_merge_state(merge_id: int) -> CompanyMergeState async

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

Source code in affinity/services/companies.py

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

get_merge_status(task_id: str) -> MergeTask async

Check the status of a merge operation.

Source code in affinity/services/companies.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/company-merges/{task_id}")
    return MergeTask.model_validate(data)

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

Auto-paginate all companies.

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

Note

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

Source code in affinity/services/companies.py

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

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

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

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

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

Source code in affinity/services/companies.py

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

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

    return AsyncPageIterator(fetch_page)

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

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

Source code in affinity/services/companies.py

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

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

    return AsyncPageIterator(fetch_page)

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

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

Source code in affinity/services/companies.py

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

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

    return AsyncPageIterator(fetch_page)

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

Get a page of companies.

Parameters:

Name	Type	Description	Default
ids	Sequence[CompanyId] | None	Specific company IDs to fetch (batch lookup)	None
field_ids	Sequence[AnyFieldId] | None	Specific field IDs to include in response	None
field_types	Sequence[FieldType] | None	Field types to include (e.g., ["enriched", "global"])	None
limit	int | None	Maximum number of results (API default: 100)	None
cursor	str | None	Cursor to resume pagination (opaque; obtained from prior responses)	None

Returns:

Type	Description
PaginatedResponse[Company]	Paginated response with companies

Note

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

Source code in affinity/services/companies.py

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

    Args:
        ids: Specific company IDs to fetch (batch lookup)
        field_ids: Specific field IDs to include in response
        field_types: Field types to include (e.g., ["enriched", "global"])
        limit: Maximum number of results (API default: 100)
        cursor: Cursor to resume pagination (opaque; obtained from prior responses)

    Returns:
        Paginated response with companies

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

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

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

One page of past company merges, newest first.

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

Source code in affinity/services/companies.py

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

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

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

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

Filters: creator_id, created_after (inclusive) / created_before (exclusive), updated_after, rounded outward to whole seconds. limit is 1-100. Needs Affinity API version 2026-07-15 or newer (sent for this call and every page).

Source code in affinity/services/companies.py

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

    Filters: ``creator_id``, ``created_after`` (inclusive) / ``created_before`` (exclusive),
    ``updated_after``, rounded outward to whole seconds. ``limit`` is 1-100.
    Needs Affinity API version 2026-07-15 or newer (sent for this call and every page).
    """
    check_cursor_alone(
        cursor, creator_id, created_after, created_before, updated_after, limit, total_count
    )
    if cursor is not None:
        return page_of(
            NoteV2,
            await self._client.get_url(cursor, min_api_version=COMPANY_NOTES_MIN_API_VERSION),
        )
    params = note_params(
        creator_id=creator_id,
        created_after=created_after,
        created_before=created_before,
        updated_after=updated_after,
        limit=limit,
        total_count=total_count,
    )
    return page_of(
        NoteV2,
        await self._client.get(
            f"/companies/{int(company_id)}/notes",
            params=params or None,
            min_api_version=COMPANY_NOTES_MIN_API_VERSION,
        ),
    )

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

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

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

Source code in affinity/services/companies.py

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

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

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

Merge a duplicate company into a primary company.

Returns a task URL to check merge status.

Source code in affinity/services/companies.py

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

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

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

Iterate company pages (not items), yielding PaginatedResponse[Company].

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

Parameters:

Name	Type	Description	Default
ids	Sequence[CompanyId] | None	Specific company IDs to fetch (batch lookup)	None
field_ids	Sequence[AnyFieldId] | None	Specific field IDs to include in response	None
field_types	Sequence[FieldType] | None	Field types to include (e.g., ["enriched", "global"])	None
limit	int | None	Maximum results per page	None
cursor	str | None	Cursor to resume pagination	None

Returns:

Type	Description
AsyncIterator[PaginatedResponse[Company]]	AsyncIterator of PaginatedResponse[Company]

Note

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

Source code in affinity/services/companies.py

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

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

    Args:
        ids: Specific company IDs to fetch (batch lookup)
        field_ids: Specific field IDs to include in response
        field_types: Field types to include (e.g., ["enriched", "global"])
        limit: Maximum results per page
        cursor: Cursor to resume pagination

    Returns:
        AsyncIterator of PaginatedResponse[Company]

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

resolve(*, domain: str | None = None, name: str | None = None) -> Company | None async

Find a single company by domain 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/companies.py

async def resolve(
    self,
    *,
    domain: str | None = None,
    name: str | None = None,
) -> Company | None:
    """
    Find a single company by domain 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 domain and not name:
        raise ValueError("Must provide either domain or name")

    term = domain or name or ""
    result = await self.search(term, page_size=10)

    for company in result.data:
        if domain and company.domain and company.domain.lower() == domain.lower():
            return company
        if name and company.name and company.name.lower() == name.lower():
            return company

    return None

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

Search for companies by name or domain.

Uses V1 API for search functionality not available in V2.

Source code in affinity/services/companies.py

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

    Uses V1 API for search functionality not available in V2.
    """
    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("/organizations", params=params, v1=True)
    items = [Company.model_validate(o) for o in data.get("organizations", [])]
    return PaginatedResponse[Company](
        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[Company] async

Iterate all V1 company-search results with automatic pagination.

Parameters:

Name	Type	Description	Default
term	str	Search term (name or domain)	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[Company]	Company objects matching the search term

Source code in affinity/services/companies.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[Company]:
    """
    Iterate all V1 company-search results with automatic pagination.

    Args:
        term: Search term (name or domain)
        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:
        Company 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 company in page.data:
            yield company

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

Iterate V1 company-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 domain)	required
with_interaction_dates	bool	Include interaction date data	False
with_interaction_persons	bool	Include persons for interactions	False
with_opportunities	bool	Include associated opportunity IDs	False
page_size	int | None	Results per page (max 500)	None
page_token	str | None	Resume from this pagination token	None

Yields:

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

Source code in affinity/services/companies.py

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

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

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

    Yields:
        PaginatedResponse[Company] 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(company_id: CompanyId, data: CompanyUpdate) -> Company async

Update an existing company.

Uses V1 API.

Source code in affinity/services/companies.py

async def update(self, company_id: CompanyId, data: CompanyUpdate) -> Company:
    """
    Update an existing company.

    Uses V1 API.
    """
    payload = data.model_dump(
        by_alias=True,
        mode="json",
        exclude_unset=True,
        exclude_none=True,
    )
    result = await self._client.put(
        f"/organizations/{company_id}",
        json=payload,
        v1=True,
    )

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

    return Company.model_validate(result)