Lists

Finding list entries with Affinity's search

client.lists.entries(list_id).find(...), iter_find(...) and find_body(...) filter and sort a list's entries on Affinity's side (V2 search, beta) and return the same entries as entries(list_id).list(). They work like the companies methods and need the "Export data from Lists" permission.

from affinity.criteria import Condition, Sort
from affinity.types import ListId

page = client.lists.entries(ListId(123)).find(
    Condition("field-456", "is-within-the-last", (30, "day")),
    sort=Sort("field-789", "desc"),
    limit=100,
)

Service for managing lists.

Lists are spreadsheet-like collections of people, companies, or opportunities.

Source code in affinity/services/lists.py

class ListService:
    """
    Service for managing lists.

    Lists are spreadsheet-like collections of people, companies, or opportunities.
    """

    def __init__(self, client: HTTPClient):
        self._client = client
        self._resolve_cache: dict[tuple[str, ListType | None], AffinityList | None] = {}
        # (timestamp, size), shared by every list service on this client
        self._size_cache: dict[int, tuple[float, int]] = _size_cache_for(client)
        # Fields, dropdown options and refused searches for filtered reads, per list
        self._search_cache = SearchCache()

    def entries(self, list_id: ListId) -> ListEntryService:
        """
        Get a ListEntryService for a specific list.

        This is the explicit path for retrieving "full row" data via list entries.
        """
        return ListEntryService(self._client, list_id, self._search_cache)

    # =========================================================================
    # List Operations (V2 for read, V1 for write)
    # =========================================================================

    def list(
        self,
        *,
        limit: int | None = None,
        cursor: str | None = None,
        term: str | None = None,
    ) -> PaginatedResponse[AffinityList]:
        """
        Get all lists accessible to you.

        Args:
            limit: Maximum results per page.
            cursor: Cursor to resume pagination (opaque; obtained from prior responses).
            term: Only lists whose name contains this text (case-insensitive substring,
                matched by Affinity). Kept across pages by the cursor.

        Returns:
            Paginated list of lists (without field metadata)
        """
        term = term or None
        if cursor is not None:
            if limit is not None or term 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:
            if limit is not None and limit <= 0:
                raise ValueError("'limit' must be > 0")
            params: dict[str, Any] = {}
            if limit is not None:
                params["limit"] = limit
            if term is not None:
                params["term"] = term
            data = self._client.get("/lists", params=params or None)

        return PaginatedResponse[AffinityList](
            data=[
                _safe_model_validate(AffinityList, list_item) for list_item in data.get("data", [])
            ],
            pagination=_safe_model_validate(PaginationInfo, data.get("pagination", {})),
        )

    def get_first(self) -> AffinityList | None:
        """Get the first list, or None if no lists exist."""
        page = self.list(limit=1)
        return page.data[0] if page.data else None

    def pages(
        self,
        *,
        limit: int | None = None,
        cursor: str | None = None,
        term: str | None = None,
    ) -> Iterator[PaginatedResponse[AffinityList]]:
        """
        Iterate list pages (not items), yielding `PaginatedResponse[AffinityList]`.

        This is useful for ETL scripts that want checkpoint/resume via `page.next_cursor`.
        ``term`` keeps only lists whose name contains it (case-insensitive, matched by Affinity).
        """
        term = term or None
        if cursor is not None and (limit is not None or term 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."
            )
        requested_cursor = cursor
        guard = _CursorGuard(cursor)
        page = self.list(limit=limit, term=term) if cursor is None else self.list(cursor=cursor)
        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) -> Iterator[AffinityList]:
        """Iterate through all accessible lists."""

        def fetch_page(next_url: str | None) -> PaginatedResponse[AffinityList]:
            if next_url:
                data = self._client.get_url(next_url)
                return PaginatedResponse[AffinityList](
                    data=[
                        _safe_model_validate(AffinityList, list_item)
                        for list_item in data.get("data", [])
                    ],
                    pagination=_safe_model_validate(PaginationInfo, data.get("pagination", {})),
                )
            return self.list()

        return PageIterator(fetch_page)

    def iter(self) -> Iterator[AffinityList]:
        """
        Auto-paginate all lists.

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

    def get(self, list_id: ListId) -> AffinityList:
        """
        Get a single list by ID (V2).

        ``fields`` is not filled: use :meth:`get_fields`. For the number of entries use
        :meth:`get_size`.
        """
        data = self._client.get(f"/lists/{int(list_id)}")
        return _safe_model_validate(AffinityList, data)

    def get_size(self, list_id: ListId, *, force: bool = False) -> int:
        """
        Get accurate list size. Uses V1 API, cached for 5 minutes.

        Adding or removing a row through this client (``entries(list_id).add_*`` / ``delete``,
        ``opportunities.create``) drops the cached size; changes made elsewhere show after the
        cache expires, or with ``force=True``.

        Args:
            list_id: The list ID.
            force: If True, bypass cache and fetch fresh value from API.

        Note: The V2 API's listSize field is unreliable (often returns 0 for
        non-empty lists). This method uses the V1 API which returns accurate values.
        """
        if not force and list_id in self._size_cache:
            cached_at, size = self._size_cache[list_id]
            if time.monotonic() - cached_at < _SIZE_CACHE_TTL:
                return size

        # V2 has no list size: read it from V1
        data = self._client.get(f"/lists/{int(list_id)}", v1=True)
        size = _safe_model_validate(AffinityList, data)._list_size_hint
        self._size_cache[list_id] = (time.monotonic(), size)
        return size

    def resolve(
        self,
        *,
        name: str,
        list_type: ListType | None = None,
    ) -> AffinityList | None:
        """
        Find a single list by name (case-insensitive exact match; optionally filtered by type).

        Notes:
        - Affinity narrows the lists by name first (``term``); the exact comparison is made
          here. If that finds nothing, every list is checked (see `resolve_all()`).
        - Matches are cached in-memory on this service instance (a miss is not). If you call
          this frequently, reuse the client, or persist the resolved `ListId` in your own
          configuration.

        If multiple matches exist, returns the first match in server-provided order.
        """
        key = (name.lower(), list_type)
        if key in self._resolve_cache:
            return self._resolve_cache[key]

        matches = self.resolve_all(name=name, list_type=list_type)
        found = matches[0] if matches else None
        if found is not None:  # a miss isn't cached: the list may be created later
            self._resolve_cache[key] = found
        return found

    def resolve_all(
        self,
        *,
        name: str,
        list_type: ListType | None = None,
    ) -> builtins.list[AffinityList]:
        """
        Find all lists matching a name (case-insensitive exact match; optionally by type).

        Notes:
        - For an ASCII name, Affinity first narrows the lists by name (``term``, a
          case-insensitive substring match) and the exact comparison is made here. If that
          yields no match, every list is checked, so a difference in how Affinity folds case
          or whitespace can't hide a list. Non-ASCII names always check every list.
        - Unlike `resolve()`, this does not cache results.
        """
        term = name.strip()
        if term and term.isascii():
            candidates = (item for page in self.pages(term=term) for item in page.data)
            matches = _exact_name_matches(candidates, name, list_type)
            if matches:
                return matches
        return _exact_name_matches(self.all(), name, list_type)

    def create(self, data: ListCreate) -> AffinityList:
        """
        Create a list (V2); the caller becomes its creator and owner.

        A public list needs the "Share accessible Lists globally" permission. V2 can't set
        ``owner_id`` or ``additional_permissions``: with either, the list is created through V1.
        """
        if _uses_v1_create(data):
            result = self._client.post("/lists", json=_v1_list_create_payload(data), v1=True)
        else:
            result = self._client.post("/lists", json=_list_create_body(data))

        # Invalidate cache
        if self._client.cache:
            self._client.cache.invalidate_prefix("list")
        self._resolve_cache.clear()

        return _safe_model_validate(AffinityList, result)

    # =========================================================================
    # Field Operations
    # =========================================================================

    def get_fields(
        self,
        list_id: ListId,
        *,
        filter: str | None = None,
        includes: Sequence[str] | None = None,
    ) -> builtins.list[FieldMetadata]:
        """
        Get fields (columns) for a list.

        Includes list-specific, global, enriched, and relationship intelligence 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(
            f"list_{list_id}_fields:",
            filter=filter,
            includes=includes,
        )

        data = self._client.get_all_pages(
            f"/lists/{list_id}/fields",
            params=params or None,
            cache_key=cache_key,
            cache_ttl=300,
        )

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

    # =========================================================================
    # Dropdown options of list fields
    # =========================================================================

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

        Older API versions report status options as ``ranked-dropdown`` without
        ``status_category``. ``with_status_types=True`` reads with API version
        {DROPDOWN_OPTION_WRITES_MIN_API_VERSION} (as the option writes do) to get the real
        ``status-dropdown`` type; a client pinned to an older version then raises
        ``ApiVersionTooOldError``.
        """
        data = self._client.get_all_pages(
            options_path(field_id, list_id=list_id, entity=None),
            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_dropdown_option(
        self, list_id: ListId, field_id: AnyFieldId, option_id: int
    ) -> DropdownOption:
        """One option of a list field."""
        return DropdownOption.model_validate(
            self._client.get(option_path(list_id, field_id, option_id))
        )

    def create_field_dropdown_option(
        self,
        list_id: ListId,
        field_id: AnyFieldId,
        *,
        option_type: DropdownOptionType,
        text: str,
        rank: int | None = None,
        color: DropdownOptionColorName | None = None,
        status_category: DropdownStatusCategory | None = None,
        win_rate: int | None = None,
    ) -> DropdownOption:
        """Add an option to a list field.

        ``option_type`` must match the field: ``"dropdown"`` (text only), ``"ranked-dropdown"``
        (``rank`` and ``color`` required) or ``"status-dropdown"`` (``rank``, ``color`` and
        ``status_category`` required; ``win_rate`` 0-100 only for ``"open"``). Needs Affinity API
        version {DROPDOWN_OPTION_WRITES_MIN_API_VERSION} (sent for this call).
        """
        body = create_body(
            option_type=option_type,
            text=text,
            rank=rank,
            color=color,
            status_category=status_category,
            win_rate=win_rate,
        )
        data = self._client.post(
            options_path(field_id, list_id=list_id, entity=None),
            json=body,
            min_api_version=DROPDOWN_OPTION_WRITES_MIN_API_VERSION,
        )
        return DropdownOption.model_validate(data)

    def update_field_dropdown_option(
        self,
        list_id: ListId,
        field_id: AnyFieldId,
        option_id: int,
        *,
        text: str | None = None,
        rank: int | None = None,
        color: DropdownOptionColorName | None = None,
        status_category: DropdownStatusCategory | None = None,
        win_rate: int | None = None,
    ) -> DropdownOption:
        """Change an option of a list field (only what you pass).

        A ``dropdown`` option can change its text; ``ranked-dropdown`` also rank and color;
        ``status-dropdown`` also status category and win rate. Renaming changes the value shown
        on every entry that has this option. Returns the option as read back afterwards.
        """
        body = update_body(
            text=text,
            rank=rank,
            color=color,
            status_category=status_category,
            win_rate=win_rate,
        )
        self._client.post(
            option_path(list_id, field_id, option_id),
            json=body,
            min_api_version=DROPDOWN_OPTION_WRITES_MIN_API_VERSION,
        )
        # The API answers 204 with no body: read the option back
        return DropdownOption.model_validate(
            self._client.get(
                option_path(list_id, field_id, option_id),
                min_api_version=DROPDOWN_OPTION_WRITES_MIN_API_VERSION,
            )
        )

    def delete_field_dropdown_option(
        self, list_id: ListId, field_id: AnyFieldId, option_id: int
    ) -> bool:
        """Delete an option of a list field. **Cannot be undone.**

        Affinity also clears this field on every list entry currently set to the option; those
        values cannot be recovered.
        """
        self._client.delete(
            option_path(list_id, field_id, option_id),
            min_api_version=DROPDOWN_OPTION_WRITES_MIN_API_VERSION,
        )
        return True

    # =========================================================================
    # Saved View Operations
    # =========================================================================

    def get_saved_views(
        self,
        list_id: ListId,
        *,
        limit: int | None = None,
        cursor: str | None = None,
    ) -> PaginatedResponse[SavedView]:
        """
        Get saved views for a list.

        Args:
            list_id: List id for the initial request.
            limit: Maximum results per page.
            cursor: Cursor to resume pagination (opaque; obtained from prior responses).
        """
        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."
                )
            cursor_list_id = _saved_views_list_id_from_cursor(cursor)
            if cursor_list_id is not None and int(list_id) != cursor_list_id:
                raise ValueError(
                    f"Cursor does not match list_id: cursor is for list {cursor_list_id}, "
                    f"requested list_id is {int(list_id)}"
                )
            data = self._client.get_url(cursor)
        else:
            if limit is not None and limit <= 0:
                raise ValueError("'limit' must be > 0")
            params: dict[str, Any] = {}
            if limit is not None:
                params["limit"] = limit
            data = self._client.get(f"/lists/{list_id}/saved-views", params=params or None)

        return PaginatedResponse[SavedView](
            data=[_safe_model_validate(SavedView, v) for v in data.get("data", [])],
            pagination=_safe_model_validate(PaginationInfo, data.get("pagination", {})),
        )

    def saved_views_pages(
        self,
        list_id: ListId,
        *,
        limit: int | None = None,
        cursor: str | None = None,
    ) -> Iterator[PaginatedResponse[SavedView]]:
        """Iterate saved view pages, yielding `PaginatedResponse[SavedView]`."""
        if cursor is not None and 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."
            )
        requested_cursor = cursor
        guard = _CursorGuard(cursor)
        page = (
            self.get_saved_views(list_id, limit=limit)
            if cursor is None
            else self.get_saved_views(list_id, cursor=cursor)
        )
        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.get_saved_views(list_id, cursor=requested_cursor)

    def saved_views_all(self, list_id: ListId) -> Iterator[SavedView]:
        """Iterate all saved views for a list."""
        for page in self.saved_views_pages(list_id):
            yield from page.data

    def get_saved_view(self, list_id: ListId, view_id: SavedViewId) -> SavedView:
        """Get a single saved view."""
        data = self._client.get(f"/lists/{list_id}/saved-views/{view_id}")
        return _safe_model_validate(SavedView, data)

all() -> Iterator[AffinityList]

Iterate through all accessible lists.

Source code in affinity/services/lists.py

def all(self) -> Iterator[AffinityList]:
    """Iterate through all accessible lists."""

    def fetch_page(next_url: str | None) -> PaginatedResponse[AffinityList]:
        if next_url:
            data = self._client.get_url(next_url)
            return PaginatedResponse[AffinityList](
                data=[
                    _safe_model_validate(AffinityList, list_item)
                    for list_item in data.get("data", [])
                ],
                pagination=_safe_model_validate(PaginationInfo, data.get("pagination", {})),
            )
        return self.list()

    return PageIterator(fetch_page)

create(data: ListCreate) -> AffinityList

Create a list (V2); the caller becomes its creator and owner.

A public list needs the "Share accessible Lists globally" permission. V2 can't set owner_id or additional_permissions: with either, the list is created through V1.

Source code in affinity/services/lists.py

def create(self, data: ListCreate) -> AffinityList:
    """
    Create a list (V2); the caller becomes its creator and owner.

    A public list needs the "Share accessible Lists globally" permission. V2 can't set
    ``owner_id`` or ``additional_permissions``: with either, the list is created through V1.
    """
    if _uses_v1_create(data):
        result = self._client.post("/lists", json=_v1_list_create_payload(data), v1=True)
    else:
        result = self._client.post("/lists", json=_list_create_body(data))

    # Invalidate cache
    if self._client.cache:
        self._client.cache.invalidate_prefix("list")
    self._resolve_cache.clear()

    return _safe_model_validate(AffinityList, result)

create_field_dropdown_option(list_id: ListId, field_id: AnyFieldId, *, option_type: DropdownOptionType, text: str, rank: int | None = None, color: DropdownOptionColorName | None = None, status_category: DropdownStatusCategory | None = None, win_rate: int | None = None) -> DropdownOption

Add an option to a list field.

option_type must match the field: "dropdown" (text only), "ranked-dropdown" (rank and color required) or "status-dropdown" (rank, color and status_category required; win_rate 0-100 only for "open"). Needs Affinity API version {DROPDOWN_OPTION_WRITES_MIN_API_VERSION} (sent for this call).

Source code in affinity/services/lists.py

def create_field_dropdown_option(
    self,
    list_id: ListId,
    field_id: AnyFieldId,
    *,
    option_type: DropdownOptionType,
    text: str,
    rank: int | None = None,
    color: DropdownOptionColorName | None = None,
    status_category: DropdownStatusCategory | None = None,
    win_rate: int | None = None,
) -> DropdownOption:
    """Add an option to a list field.

    ``option_type`` must match the field: ``"dropdown"`` (text only), ``"ranked-dropdown"``
    (``rank`` and ``color`` required) or ``"status-dropdown"`` (``rank``, ``color`` and
    ``status_category`` required; ``win_rate`` 0-100 only for ``"open"``). Needs Affinity API
    version {DROPDOWN_OPTION_WRITES_MIN_API_VERSION} (sent for this call).
    """
    body = create_body(
        option_type=option_type,
        text=text,
        rank=rank,
        color=color,
        status_category=status_category,
        win_rate=win_rate,
    )
    data = self._client.post(
        options_path(field_id, list_id=list_id, entity=None),
        json=body,
        min_api_version=DROPDOWN_OPTION_WRITES_MIN_API_VERSION,
    )
    return DropdownOption.model_validate(data)

delete_field_dropdown_option(list_id: ListId, field_id: AnyFieldId, option_id: int) -> bool

Delete an option of a list field. Cannot be undone.

Affinity also clears this field on every list entry currently set to the option; those values cannot be recovered.

Source code in affinity/services/lists.py

def delete_field_dropdown_option(
    self, list_id: ListId, field_id: AnyFieldId, option_id: int
) -> bool:
    """Delete an option of a list field. **Cannot be undone.**

    Affinity also clears this field on every list entry currently set to the option; those
    values cannot be recovered.
    """
    self._client.delete(
        option_path(list_id, field_id, option_id),
        min_api_version=DROPDOWN_OPTION_WRITES_MIN_API_VERSION,
    )
    return True

entries(list_id: ListId) -> ListEntryService

Get a ListEntryService for a specific list.

This is the explicit path for retrieving "full row" data via list entries.

Source code in affinity/services/lists.py

def entries(self, list_id: ListId) -> ListEntryService:
    """
    Get a ListEntryService for a specific list.

    This is the explicit path for retrieving "full row" data via list entries.
    """
    return ListEntryService(self._client, list_id, self._search_cache)

get(list_id: ListId) -> AffinityList

Get a single list by ID (V2).

fields is not filled: use :meth:get_fields. For the number of entries use :meth:get_size.

Source code in affinity/services/lists.py

def get(self, list_id: ListId) -> AffinityList:
    """
    Get a single list by ID (V2).

    ``fields`` is not filled: use :meth:`get_fields`. For the number of entries use
    :meth:`get_size`.
    """
    data = self._client.get(f"/lists/{int(list_id)}")
    return _safe_model_validate(AffinityList, data)

get_field_dropdown_option(list_id: ListId, field_id: AnyFieldId, option_id: int) -> DropdownOption

One option of a list field.

Source code in affinity/services/lists.py

def get_field_dropdown_option(
    self, list_id: ListId, field_id: AnyFieldId, option_id: int
) -> DropdownOption:
    """One option of a list field."""
    return DropdownOption.model_validate(
        self._client.get(option_path(list_id, field_id, option_id))
    )

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

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

Older API versions report status options as ranked-dropdown without status_category. with_status_types=True reads with API version {DROPDOWN_OPTION_WRITES_MIN_API_VERSION} (as the option writes do) to get the real status-dropdown type; a client pinned to an older version then raises ApiVersionTooOldError.

Source code in affinity/services/lists.py

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

    Older API versions report status options as ``ranked-dropdown`` without
    ``status_category``. ``with_status_types=True`` reads with API version
    {DROPDOWN_OPTION_WRITES_MIN_API_VERSION} (as the option writes do) to get the real
    ``status-dropdown`` type; a client pinned to an older version then raises
    ``ApiVersionTooOldError``.
    """
    data = self._client.get_all_pages(
        options_path(field_id, list_id=list_id, entity=None),
        params={"limit": 100},
        min_api_version=DROPDOWN_OPTION_WRITES_MIN_API_VERSION if with_status_types else None,
    )
    return parse_options(data)

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

Get fields (columns) for a list.

Includes list-specific, global, enriched, and relationship intelligence 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/lists.py

def get_fields(
    self,
    list_id: ListId,
    *,
    filter: str | None = None,
    includes: Sequence[str] | None = None,
) -> builtins.list[FieldMetadata]:
    """
    Get fields (columns) for a list.

    Includes list-specific, global, enriched, and relationship intelligence 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(
        f"list_{list_id}_fields:",
        filter=filter,
        includes=includes,
    )

    data = self._client.get_all_pages(
        f"/lists/{list_id}/fields",
        params=params or None,
        cache_key=cache_key,
        cache_ttl=300,
    )

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

get_first() -> AffinityList | None

Get the first list, or None if no lists exist.

Source code in affinity/services/lists.py

def get_first(self) -> AffinityList | None:
    """Get the first list, or None if no lists exist."""
    page = self.list(limit=1)
    return page.data[0] if page.data else None

get_saved_view(list_id: ListId, view_id: SavedViewId) -> SavedView

Get a single saved view.

Source code in affinity/services/lists.py

def get_saved_view(self, list_id: ListId, view_id: SavedViewId) -> SavedView:
    """Get a single saved view."""
    data = self._client.get(f"/lists/{list_id}/saved-views/{view_id}")
    return _safe_model_validate(SavedView, data)

get_saved_views(list_id: ListId, *, limit: int | None = None, cursor: str | None = None) -> PaginatedResponse[SavedView]

Get saved views for a list.

Parameters:

Name	Type	Description	Default
list_id	ListId	List id for the initial request.	required
limit	int | None	Maximum results per page.	None
cursor	str | None	Cursor to resume pagination (opaque; obtained from prior responses).	None

Source code in affinity/services/lists.py

def get_saved_views(
    self,
    list_id: ListId,
    *,
    limit: int | None = None,
    cursor: str | None = None,
) -> PaginatedResponse[SavedView]:
    """
    Get saved views for a list.

    Args:
        list_id: List id for the initial request.
        limit: Maximum results per page.
        cursor: Cursor to resume pagination (opaque; obtained from prior responses).
    """
    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."
            )
        cursor_list_id = _saved_views_list_id_from_cursor(cursor)
        if cursor_list_id is not None and int(list_id) != cursor_list_id:
            raise ValueError(
                f"Cursor does not match list_id: cursor is for list {cursor_list_id}, "
                f"requested list_id is {int(list_id)}"
            )
        data = self._client.get_url(cursor)
    else:
        if limit is not None and limit <= 0:
            raise ValueError("'limit' must be > 0")
        params: dict[str, Any] = {}
        if limit is not None:
            params["limit"] = limit
        data = self._client.get(f"/lists/{list_id}/saved-views", params=params or None)

    return PaginatedResponse[SavedView](
        data=[_safe_model_validate(SavedView, v) for v in data.get("data", [])],
        pagination=_safe_model_validate(PaginationInfo, data.get("pagination", {})),
    )

get_size(list_id: ListId, *, force: bool = False) -> int

Get accurate list size. Uses V1 API, cached for 5 minutes.

Adding or removing a row through this client (entries(list_id).add_* / delete, opportunities.create) drops the cached size; changes made elsewhere show after the cache expires, or with force=True.

Parameters:

Name	Type	Description	Default
list_id	ListId	The list ID.	required
force	bool	If True, bypass cache and fetch fresh value from API.	False

Note: The V2 API's listSize field is unreliable (often returns 0 for non-empty lists). This method uses the V1 API which returns accurate values.

Source code in affinity/services/lists.py

def get_size(self, list_id: ListId, *, force: bool = False) -> int:
    """
    Get accurate list size. Uses V1 API, cached for 5 minutes.

    Adding or removing a row through this client (``entries(list_id).add_*`` / ``delete``,
    ``opportunities.create``) drops the cached size; changes made elsewhere show after the
    cache expires, or with ``force=True``.

    Args:
        list_id: The list ID.
        force: If True, bypass cache and fetch fresh value from API.

    Note: The V2 API's listSize field is unreliable (often returns 0 for
    non-empty lists). This method uses the V1 API which returns accurate values.
    """
    if not force and list_id in self._size_cache:
        cached_at, size = self._size_cache[list_id]
        if time.monotonic() - cached_at < _SIZE_CACHE_TTL:
            return size

    # V2 has no list size: read it from V1
    data = self._client.get(f"/lists/{int(list_id)}", v1=True)
    size = _safe_model_validate(AffinityList, data)._list_size_hint
    self._size_cache[list_id] = (time.monotonic(), size)
    return size

iter() -> Iterator[AffinityList]

Auto-paginate all lists.

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

Source code in affinity/services/lists.py

def iter(self) -> Iterator[AffinityList]:
    """
    Auto-paginate all lists.

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

list(*, limit: int | None = None, cursor: str | None = None, term: str | None = None) -> PaginatedResponse[AffinityList]

Get all lists accessible to you.

Parameters:

Name	Type	Description	Default
limit	int | None	Maximum results per page.	None
cursor	str | None	Cursor to resume pagination (opaque; obtained from prior responses).	None
term	str | None	Only lists whose name contains this text (case-insensitive substring, matched by Affinity). Kept across pages by the cursor.	None

Returns:

Type	Description
PaginatedResponse[AffinityList]	Paginated list of lists (without field metadata)

Source code in affinity/services/lists.py

def list(
    self,
    *,
    limit: int | None = None,
    cursor: str | None = None,
    term: str | None = None,
) -> PaginatedResponse[AffinityList]:
    """
    Get all lists accessible to you.

    Args:
        limit: Maximum results per page.
        cursor: Cursor to resume pagination (opaque; obtained from prior responses).
        term: Only lists whose name contains this text (case-insensitive substring,
            matched by Affinity). Kept across pages by the cursor.

    Returns:
        Paginated list of lists (without field metadata)
    """
    term = term or None
    if cursor is not None:
        if limit is not None or term 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:
        if limit is not None and limit <= 0:
            raise ValueError("'limit' must be > 0")
        params: dict[str, Any] = {}
        if limit is not None:
            params["limit"] = limit
        if term is not None:
            params["term"] = term
        data = self._client.get("/lists", params=params or None)

    return PaginatedResponse[AffinityList](
        data=[
            _safe_model_validate(AffinityList, list_item) for list_item in data.get("data", [])
        ],
        pagination=_safe_model_validate(PaginationInfo, data.get("pagination", {})),
    )

pages(*, limit: int | None = None, cursor: str | None = None, term: str | None = None) -> Iterator[PaginatedResponse[AffinityList]]

Iterate list pages (not items), yielding PaginatedResponse[AffinityList].

This is useful for ETL scripts that want checkpoint/resume via page.next_cursor. term keeps only lists whose name contains it (case-insensitive, matched by Affinity).

Source code in affinity/services/lists.py

def pages(
    self,
    *,
    limit: int | None = None,
    cursor: str | None = None,
    term: str | None = None,
) -> Iterator[PaginatedResponse[AffinityList]]:
    """
    Iterate list pages (not items), yielding `PaginatedResponse[AffinityList]`.

    This is useful for ETL scripts that want checkpoint/resume via `page.next_cursor`.
    ``term`` keeps only lists whose name contains it (case-insensitive, matched by Affinity).
    """
    term = term or None
    if cursor is not None and (limit is not None or term 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."
        )
    requested_cursor = cursor
    guard = _CursorGuard(cursor)
    page = self.list(limit=limit, term=term) if cursor is None else self.list(cursor=cursor)
    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(*, name: str, list_type: ListType | None = None) -> AffinityList | None

Find a single list by name (case-insensitive exact match; optionally filtered by type).

Notes: - Affinity narrows the lists by name first (term); the exact comparison is made here. If that finds nothing, every list is checked (see resolve_all()). - Matches are cached in-memory on this service instance (a miss is not). If you call this frequently, reuse the client, or persist the resolved ListId in your own configuration.

If multiple matches exist, returns the first match in server-provided order.

Source code in affinity/services/lists.py

def resolve(
    self,
    *,
    name: str,
    list_type: ListType | None = None,
) -> AffinityList | None:
    """
    Find a single list by name (case-insensitive exact match; optionally filtered by type).

    Notes:
    - Affinity narrows the lists by name first (``term``); the exact comparison is made
      here. If that finds nothing, every list is checked (see `resolve_all()`).
    - Matches are cached in-memory on this service instance (a miss is not). If you call
      this frequently, reuse the client, or persist the resolved `ListId` in your own
      configuration.

    If multiple matches exist, returns the first match in server-provided order.
    """
    key = (name.lower(), list_type)
    if key in self._resolve_cache:
        return self._resolve_cache[key]

    matches = self.resolve_all(name=name, list_type=list_type)
    found = matches[0] if matches else None
    if found is not None:  # a miss isn't cached: the list may be created later
        self._resolve_cache[key] = found
    return found

resolve_all(*, name: str, list_type: ListType | None = None) -> builtins.list[AffinityList]

Find all lists matching a name (case-insensitive exact match; optionally by type).

Notes: - For an ASCII name, Affinity first narrows the lists by name (term, a case-insensitive substring match) and the exact comparison is made here. If that yields no match, every list is checked, so a difference in how Affinity folds case or whitespace can't hide a list. Non-ASCII names always check every list. - Unlike resolve(), this does not cache results.

Source code in affinity/services/lists.py

def resolve_all(
    self,
    *,
    name: str,
    list_type: ListType | None = None,
) -> builtins.list[AffinityList]:
    """
    Find all lists matching a name (case-insensitive exact match; optionally by type).

    Notes:
    - For an ASCII name, Affinity first narrows the lists by name (``term``, a
      case-insensitive substring match) and the exact comparison is made here. If that
      yields no match, every list is checked, so a difference in how Affinity folds case
      or whitespace can't hide a list. Non-ASCII names always check every list.
    - Unlike `resolve()`, this does not cache results.
    """
    term = name.strip()
    if term and term.isascii():
        candidates = (item for page in self.pages(term=term) for item in page.data)
        matches = _exact_name_matches(candidates, name, list_type)
        if matches:
            return matches
    return _exact_name_matches(self.all(), name, list_type)

saved_views_all(list_id: ListId) -> Iterator[SavedView]

Iterate all saved views for a list.

Source code in affinity/services/lists.py

def saved_views_all(self, list_id: ListId) -> Iterator[SavedView]:
    """Iterate all saved views for a list."""
    for page in self.saved_views_pages(list_id):
        yield from page.data

saved_views_pages(list_id: ListId, *, limit: int | None = None, cursor: str | None = None) -> Iterator[PaginatedResponse[SavedView]]

Iterate saved view pages, yielding PaginatedResponse[SavedView].

Source code in affinity/services/lists.py

def saved_views_pages(
    self,
    list_id: ListId,
    *,
    limit: int | None = None,
    cursor: str | None = None,
) -> Iterator[PaginatedResponse[SavedView]]:
    """Iterate saved view pages, yielding `PaginatedResponse[SavedView]`."""
    if cursor is not None and 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."
        )
    requested_cursor = cursor
    guard = _CursorGuard(cursor)
    page = (
        self.get_saved_views(list_id, limit=limit)
        if cursor is None
        else self.get_saved_views(list_id, cursor=cursor)
    )
    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.get_saved_views(list_id, cursor=requested_cursor)

update_field_dropdown_option(list_id: ListId, field_id: AnyFieldId, option_id: int, *, text: str | None = None, rank: int | None = None, color: DropdownOptionColorName | None = None, status_category: DropdownStatusCategory | None = None, win_rate: int | None = None) -> DropdownOption

Change an option of a list field (only what you pass).

A dropdown option can change its text; ranked-dropdown also rank and color; status-dropdown also status category and win rate. Renaming changes the value shown on every entry that has this option. Returns the option as read back afterwards.

Source code in affinity/services/lists.py

def update_field_dropdown_option(
    self,
    list_id: ListId,
    field_id: AnyFieldId,
    option_id: int,
    *,
    text: str | None = None,
    rank: int | None = None,
    color: DropdownOptionColorName | None = None,
    status_category: DropdownStatusCategory | None = None,
    win_rate: int | None = None,
) -> DropdownOption:
    """Change an option of a list field (only what you pass).

    A ``dropdown`` option can change its text; ``ranked-dropdown`` also rank and color;
    ``status-dropdown`` also status category and win rate. Renaming changes the value shown
    on every entry that has this option. Returns the option as read back afterwards.
    """
    body = update_body(
        text=text,
        rank=rank,
        color=color,
        status_category=status_category,
        win_rate=win_rate,
    )
    self._client.post(
        option_path(list_id, field_id, option_id),
        json=body,
        min_api_version=DROPDOWN_OPTION_WRITES_MIN_API_VERSION,
    )
    # The API answers 204 with no body: read the option back
    return DropdownOption.model_validate(
        self._client.get(
            option_path(list_id, field_id, option_id),
            min_api_version=DROPDOWN_OPTION_WRITES_MIN_API_VERSION,
        )
    )

Async list operations (TR-009).

Source code in affinity/services/lists.py

class AsyncListService:
    """Async list operations (TR-009)."""

    def __init__(self, client: AsyncHTTPClient):
        self._client = client
        self._resolve_cache: dict[tuple[str, ListType | None], AffinityList | None] = {}
        # (timestamp, size), shared by every list service on this client
        self._size_cache: dict[int, tuple[float, int]] = _size_cache_for(client)
        # Fields, dropdown options and refused searches for filtered reads, per list
        self._search_cache = SearchCache()

    def entries(self, list_id: ListId) -> AsyncListEntryService:
        """
        Get an AsyncListEntryService for a specific list.

        This is the explicit path for retrieving "full row" data via list entries.
        """
        return AsyncListEntryService(self._client, list_id, self._search_cache)

    async def list(
        self,
        *,
        limit: int | None = None,
        cursor: str | None = None,
        term: str | None = None,
    ) -> PaginatedResponse[AffinityList]:
        """
        Get all lists accessible to you.

        Args:
            limit: Maximum results per page.
            cursor: Cursor to resume pagination (opaque; obtained from prior responses).
            term: Only lists whose name contains this text (case-insensitive substring,
                matched by Affinity). Kept across pages by the cursor.

        Returns:
            Paginated list of lists (without field metadata)
        """
        term = term or None
        if cursor is not None:
            if limit is not None or term 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:
            if limit is not None and limit <= 0:
                raise ValueError("'limit' must be > 0")
            params: dict[str, Any] = {}
            if limit is not None:
                params["limit"] = limit
            if term is not None:
                params["term"] = term
            data = await self._client.get("/lists", params=params or None)
        return PaginatedResponse[AffinityList](
            data=[
                _safe_model_validate(AffinityList, list_item) for list_item in data.get("data", [])
            ],
            pagination=_safe_model_validate(PaginationInfo, data.get("pagination", {})),
        )

    async def get_first(self) -> AffinityList | None:
        """Get the first list, or None if no lists exist."""
        page = await self.list(limit=1)
        return page.data[0] if page.data else None

    async def pages(
        self,
        *,
        limit: int | None = None,
        cursor: str | None = None,
        term: str | None = None,
    ) -> AsyncIterator[PaginatedResponse[AffinityList]]:
        """
        Iterate list pages (not items), yielding `PaginatedResponse[AffinityList]`.

        This is useful for ETL scripts that want checkpoint/resume via `page.next_cursor`.
        ``term`` keeps only lists whose name contains it (case-insensitive, matched by Affinity).
        """
        term = term or None
        if cursor is not None and (limit is not None or term 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."
            )
        requested_cursor = cursor
        guard = _CursorGuard(cursor)
        page = (
            await self.list(limit=limit, term=term)
            if cursor is None
            else await self.list(cursor=cursor)
        )
        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) -> AsyncIterator[AffinityList]:
        """Iterate through all accessible lists."""

        async def fetch_page(next_url: str | None) -> PaginatedResponse[AffinityList]:
            if next_url:
                data = await self._client.get_url(next_url)
                return PaginatedResponse[AffinityList](
                    data=[
                        _safe_model_validate(AffinityList, list_item)
                        for list_item in data.get("data", [])
                    ],
                    pagination=_safe_model_validate(PaginationInfo, data.get("pagination", {})),
                )
            return await self.list()

        return AsyncPageIterator(fetch_page)

    def iter(self) -> AsyncIterator[AffinityList]:
        """
        Auto-paginate all lists.

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

    # =========================================================================
    # Saved View Operations
    # =========================================================================

    async def get_saved_views(
        self,
        list_id: ListId,
        *,
        limit: int | None = None,
        cursor: str | None = None,
    ) -> PaginatedResponse[SavedView]:
        """
        Get saved views for a list.

        Args:
            list_id: List id for the initial request.
            limit: Maximum results per page.
            cursor: Cursor to resume pagination (opaque; obtained from prior responses).
        """
        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."
                )
            cursor_list_id = _saved_views_list_id_from_cursor(cursor)
            if cursor_list_id is not None and int(list_id) != cursor_list_id:
                raise ValueError(
                    f"Cursor does not match list_id: cursor is for list {cursor_list_id}, "
                    f"requested list_id is {int(list_id)}"
                )
            data = await self._client.get_url(cursor)
        else:
            if limit is not None and limit <= 0:
                raise ValueError("'limit' must be > 0")
            params: dict[str, Any] = {}
            if limit is not None:
                params["limit"] = limit
            data = await self._client.get(f"/lists/{list_id}/saved-views", params=params or None)

        return PaginatedResponse[SavedView](
            data=[_safe_model_validate(SavedView, v) for v in data.get("data", [])],
            pagination=_safe_model_validate(PaginationInfo, data.get("pagination", {})),
        )

    async def saved_views_pages(
        self,
        list_id: ListId,
        *,
        limit: int | None = None,
        cursor: str | None = None,
    ) -> AsyncIterator[PaginatedResponse[SavedView]]:
        """Iterate saved view pages, yielding `PaginatedResponse[SavedView]`."""
        if cursor is not None and 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."
            )
        requested_cursor = cursor
        guard = _CursorGuard(cursor)
        page = (
            await self.get_saved_views(list_id, limit=limit)
            if cursor is None
            else await self.get_saved_views(list_id, cursor=cursor)
        )
        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.get_saved_views(list_id, cursor=requested_cursor)

    async def saved_views_all(self, list_id: ListId) -> AsyncIterator[SavedView]:
        """Iterate all saved views for a list."""
        async for page in self.saved_views_pages(list_id):
            for item in page.data:
                yield item

    async def get_saved_view(self, list_id: ListId, view_id: SavedViewId) -> SavedView:
        """Get a single saved view."""
        data = await self._client.get(f"/lists/{list_id}/saved-views/{view_id}")
        return _safe_model_validate(SavedView, data)

    async def get(self, list_id: ListId) -> AffinityList:
        """
        Get a single list by ID (V2).

        ``fields`` is not filled: use :meth:`get_fields`. For the number of entries use
        :meth:`get_size`.
        """
        data = await self._client.get(f"/lists/{int(list_id)}")
        return _safe_model_validate(AffinityList, data)

    async def get_size(self, list_id: ListId, *, force: bool = False) -> int:
        """
        Get accurate list size. Uses V1 API, cached for 5 minutes.

        Adding or removing a row through this client (``entries(list_id).add_*`` / ``delete``,
        ``opportunities.create``) drops the cached size; changes made elsewhere show after the
        cache expires, or with ``force=True``.

        Args:
            list_id: The list ID.
            force: If True, bypass cache and fetch fresh value from API.

        Note: The V2 API's listSize field is unreliable (often returns 0 for
        non-empty lists). This method uses the V1 API which returns accurate values.
        """
        if not force and list_id in self._size_cache:
            cached_at, size = self._size_cache[list_id]
            if time.monotonic() - cached_at < _SIZE_CACHE_TTL:
                return size

        # V2 has no list size: read it from V1
        data = await self._client.get(f"/lists/{int(list_id)}", v1=True)
        size = _safe_model_validate(AffinityList, data)._list_size_hint
        self._size_cache[list_id] = (time.monotonic(), size)
        return size

    async def resolve(
        self,
        *,
        name: str,
        list_type: ListType | None = None,
    ) -> AffinityList | None:
        """
        Find a single list by name (case-insensitive exact match; optionally filtered by type).

        Notes:
        - Affinity narrows the lists by name first (``term``); the exact comparison is made
          here. If that finds nothing, every list is checked (see `resolve_all()`).
        - Matches are cached in-memory on this service instance (a miss is not). If you call
          this frequently, reuse the client, or persist the resolved `ListId` in your own
          configuration.

        If multiple matches exist, returns the first match in server-provided order.
        """
        key = (name.lower(), list_type)
        if key in self._resolve_cache:
            return self._resolve_cache[key]

        matches = await self.resolve_all(name=name, list_type=list_type)
        found = matches[0] if matches else None
        if found is not None:  # a miss isn't cached: the list may be created later
            self._resolve_cache[key] = found
        return found

    async def resolve_all(
        self,
        *,
        name: str,
        list_type: ListType | None = None,
    ) -> builtins.list[AffinityList]:
        """
        Find all lists matching a name (case-insensitive exact match; optionally by type).

        Notes:
        - For an ASCII name, Affinity first narrows the lists by name (``term``, a
          case-insensitive substring match) and the exact comparison is made here. If that
          yields no match, every list is checked, so a difference in how Affinity folds case
          or whitespace can't hide a list. Non-ASCII names always check every list.
        - Unlike `resolve()`, this does not cache results.
        """
        term = name.strip()
        if term and term.isascii():
            candidates = [item async for page in self.pages(term=term) for item in page.data]
            matches = _exact_name_matches(candidates, name, list_type)
            if matches:
                return matches
        return _exact_name_matches([item async for item in self.all()], name, list_type)

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

    async def create(self, data: ListCreate) -> AffinityList:
        """
        Create a list (V2); the caller becomes its creator and owner.

        A public list needs the "Share accessible Lists globally" permission. V2 can't set
        ``owner_id`` or ``additional_permissions``: with either, the list is created through V1.
        """
        if _uses_v1_create(data):
            result = await self._client.post("/lists", json=_v1_list_create_payload(data), v1=True)
        else:
            result = await self._client.post("/lists", json=_list_create_body(data))

        if self._client.cache:
            self._client.cache.invalidate_prefix("list")
        self._resolve_cache.clear()

        return _safe_model_validate(AffinityList, result)

    # =========================================================================
    # Field Operations
    # =========================================================================

    async def get_fields(
        self,
        list_id: ListId,
        *,
        filter: str | None = None,
        includes: Sequence[str] | None = None,
    ) -> builtins.list[FieldMetadata]:
        """
        Get fields (columns) for a list.

        Includes list-specific, global, enriched, and relationship intelligence 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(
            f"list_{list_id}_fields:",
            filter=filter,
            includes=includes,
        )

        data = await self._client.get_all_pages(
            f"/lists/{list_id}/fields",
            params=params or None,
            cache_key=cache_key,
            cache_ttl=300,
        )

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

    # =========================================================================
    # Dropdown options of list fields
    # =========================================================================

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

        Older API versions report status options as ``ranked-dropdown`` without
        ``status_category``. ``with_status_types=True`` reads with API version
        {DROPDOWN_OPTION_WRITES_MIN_API_VERSION} (as the option writes do) to get the real
        ``status-dropdown`` type; a client pinned to an older version then raises
        ``ApiVersionTooOldError``.
        """
        data = await self._client.get_all_pages(
            options_path(field_id, list_id=list_id, entity=None),
            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_dropdown_option(
        self, list_id: ListId, field_id: AnyFieldId, option_id: int
    ) -> DropdownOption:
        """One option of a list field."""
        return DropdownOption.model_validate(
            await self._client.get(option_path(list_id, field_id, option_id))
        )

    async def create_field_dropdown_option(
        self,
        list_id: ListId,
        field_id: AnyFieldId,
        *,
        option_type: DropdownOptionType,
        text: str,
        rank: int | None = None,
        color: DropdownOptionColorName | None = None,
        status_category: DropdownStatusCategory | None = None,
        win_rate: int | None = None,
    ) -> DropdownOption:
        """Add an option to a list field.

        ``option_type`` must match the field: ``"dropdown"`` (text only), ``"ranked-dropdown"``
        (``rank`` and ``color`` required) or ``"status-dropdown"`` (``rank``, ``color`` and
        ``status_category`` required; ``win_rate`` 0-100 only for ``"open"``). Needs Affinity API
        version {DROPDOWN_OPTION_WRITES_MIN_API_VERSION} (sent for this call).
        """
        body = create_body(
            option_type=option_type,
            text=text,
            rank=rank,
            color=color,
            status_category=status_category,
            win_rate=win_rate,
        )
        data = await self._client.post(
            options_path(field_id, list_id=list_id, entity=None),
            json=body,
            min_api_version=DROPDOWN_OPTION_WRITES_MIN_API_VERSION,
        )
        return DropdownOption.model_validate(data)

    async def update_field_dropdown_option(
        self,
        list_id: ListId,
        field_id: AnyFieldId,
        option_id: int,
        *,
        text: str | None = None,
        rank: int | None = None,
        color: DropdownOptionColorName | None = None,
        status_category: DropdownStatusCategory | None = None,
        win_rate: int | None = None,
    ) -> DropdownOption:
        """Change an option of a list field (only what you pass).

        A ``dropdown`` option can change its text; ``ranked-dropdown`` also rank and color;
        ``status-dropdown`` also status category and win rate. Renaming changes the value shown
        on every entry that has this option. Returns the option as read back afterwards.
        """
        body = update_body(
            text=text,
            rank=rank,
            color=color,
            status_category=status_category,
            win_rate=win_rate,
        )
        await self._client.post(
            option_path(list_id, field_id, option_id),
            json=body,
            min_api_version=DROPDOWN_OPTION_WRITES_MIN_API_VERSION,
        )
        # The API answers 204 with no body: read the option back
        return DropdownOption.model_validate(
            await self._client.get(
                option_path(list_id, field_id, option_id),
                min_api_version=DROPDOWN_OPTION_WRITES_MIN_API_VERSION,
            )
        )

    async def delete_field_dropdown_option(
        self, list_id: ListId, field_id: AnyFieldId, option_id: int
    ) -> bool:
        """Delete an option of a list field. **Cannot be undone.**

        Affinity also clears this field on every list entry currently set to the option; those
        values cannot be recovered.
        """
        await self._client.delete(
            option_path(list_id, field_id, option_id),
            min_api_version=DROPDOWN_OPTION_WRITES_MIN_API_VERSION,
        )
        return True

all() -> AsyncIterator[AffinityList]

Iterate through all accessible lists.

Source code in affinity/services/lists.py

def all(self) -> AsyncIterator[AffinityList]:
    """Iterate through all accessible lists."""

    async def fetch_page(next_url: str | None) -> PaginatedResponse[AffinityList]:
        if next_url:
            data = await self._client.get_url(next_url)
            return PaginatedResponse[AffinityList](
                data=[
                    _safe_model_validate(AffinityList, list_item)
                    for list_item in data.get("data", [])
                ],
                pagination=_safe_model_validate(PaginationInfo, data.get("pagination", {})),
            )
        return await self.list()

    return AsyncPageIterator(fetch_page)

create(data: ListCreate) -> AffinityList async

Create a list (V2); the caller becomes its creator and owner.

A public list needs the "Share accessible Lists globally" permission. V2 can't set owner_id or additional_permissions: with either, the list is created through V1.

Source code in affinity/services/lists.py

async def create(self, data: ListCreate) -> AffinityList:
    """
    Create a list (V2); the caller becomes its creator and owner.

    A public list needs the "Share accessible Lists globally" permission. V2 can't set
    ``owner_id`` or ``additional_permissions``: with either, the list is created through V1.
    """
    if _uses_v1_create(data):
        result = await self._client.post("/lists", json=_v1_list_create_payload(data), v1=True)
    else:
        result = await self._client.post("/lists", json=_list_create_body(data))

    if self._client.cache:
        self._client.cache.invalidate_prefix("list")
    self._resolve_cache.clear()

    return _safe_model_validate(AffinityList, result)

create_field_dropdown_option(list_id: ListId, field_id: AnyFieldId, *, option_type: DropdownOptionType, text: str, rank: int | None = None, color: DropdownOptionColorName | None = None, status_category: DropdownStatusCategory | None = None, win_rate: int | None = None) -> DropdownOption async

Add an option to a list field.

option_type must match the field: "dropdown" (text only), "ranked-dropdown" (rank and color required) or "status-dropdown" (rank, color and status_category required; win_rate 0-100 only for "open"). Needs Affinity API version {DROPDOWN_OPTION_WRITES_MIN_API_VERSION} (sent for this call).

Source code in affinity/services/lists.py

async def create_field_dropdown_option(
    self,
    list_id: ListId,
    field_id: AnyFieldId,
    *,
    option_type: DropdownOptionType,
    text: str,
    rank: int | None = None,
    color: DropdownOptionColorName | None = None,
    status_category: DropdownStatusCategory | None = None,
    win_rate: int | None = None,
) -> DropdownOption:
    """Add an option to a list field.

    ``option_type`` must match the field: ``"dropdown"`` (text only), ``"ranked-dropdown"``
    (``rank`` and ``color`` required) or ``"status-dropdown"`` (``rank``, ``color`` and
    ``status_category`` required; ``win_rate`` 0-100 only for ``"open"``). Needs Affinity API
    version {DROPDOWN_OPTION_WRITES_MIN_API_VERSION} (sent for this call).
    """
    body = create_body(
        option_type=option_type,
        text=text,
        rank=rank,
        color=color,
        status_category=status_category,
        win_rate=win_rate,
    )
    data = await self._client.post(
        options_path(field_id, list_id=list_id, entity=None),
        json=body,
        min_api_version=DROPDOWN_OPTION_WRITES_MIN_API_VERSION,
    )
    return DropdownOption.model_validate(data)

delete_field_dropdown_option(list_id: ListId, field_id: AnyFieldId, option_id: int) -> bool async

Delete an option of a list field. Cannot be undone.

Affinity also clears this field on every list entry currently set to the option; those values cannot be recovered.

Source code in affinity/services/lists.py

async def delete_field_dropdown_option(
    self, list_id: ListId, field_id: AnyFieldId, option_id: int
) -> bool:
    """Delete an option of a list field. **Cannot be undone.**

    Affinity also clears this field on every list entry currently set to the option; those
    values cannot be recovered.
    """
    await self._client.delete(
        option_path(list_id, field_id, option_id),
        min_api_version=DROPDOWN_OPTION_WRITES_MIN_API_VERSION,
    )
    return True

entries(list_id: ListId) -> AsyncListEntryService

Get an AsyncListEntryService for a specific list.

This is the explicit path for retrieving "full row" data via list entries.

Source code in affinity/services/lists.py

def entries(self, list_id: ListId) -> AsyncListEntryService:
    """
    Get an AsyncListEntryService for a specific list.

    This is the explicit path for retrieving "full row" data via list entries.
    """
    return AsyncListEntryService(self._client, list_id, self._search_cache)

get(list_id: ListId) -> AffinityList async

Get a single list by ID (V2).

fields is not filled: use :meth:get_fields. For the number of entries use :meth:get_size.

Source code in affinity/services/lists.py

async def get(self, list_id: ListId) -> AffinityList:
    """
    Get a single list by ID (V2).

    ``fields`` is not filled: use :meth:`get_fields`. For the number of entries use
    :meth:`get_size`.
    """
    data = await self._client.get(f"/lists/{int(list_id)}")
    return _safe_model_validate(AffinityList, data)

get_field_dropdown_option(list_id: ListId, field_id: AnyFieldId, option_id: int) -> DropdownOption async

One option of a list field.

Source code in affinity/services/lists.py

async def get_field_dropdown_option(
    self, list_id: ListId, field_id: AnyFieldId, option_id: int
) -> DropdownOption:
    """One option of a list field."""
    return DropdownOption.model_validate(
        await self._client.get(option_path(list_id, field_id, option_id))
    )

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

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

Older API versions report status options as ranked-dropdown without status_category. with_status_types=True reads with API version {DROPDOWN_OPTION_WRITES_MIN_API_VERSION} (as the option writes do) to get the real status-dropdown type; a client pinned to an older version then raises ApiVersionTooOldError.

Source code in affinity/services/lists.py

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

    Older API versions report status options as ``ranked-dropdown`` without
    ``status_category``. ``with_status_types=True`` reads with API version
    {DROPDOWN_OPTION_WRITES_MIN_API_VERSION} (as the option writes do) to get the real
    ``status-dropdown`` type; a client pinned to an older version then raises
    ``ApiVersionTooOldError``.
    """
    data = await self._client.get_all_pages(
        options_path(field_id, list_id=list_id, entity=None),
        params={"limit": 100},
        min_api_version=DROPDOWN_OPTION_WRITES_MIN_API_VERSION if with_status_types else None,
    )
    return parse_options(data)

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

Get fields (columns) for a list.

Includes list-specific, global, enriched, and relationship intelligence 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/lists.py

async def get_fields(
    self,
    list_id: ListId,
    *,
    filter: str | None = None,
    includes: Sequence[str] | None = None,
) -> builtins.list[FieldMetadata]:
    """
    Get fields (columns) for a list.

    Includes list-specific, global, enriched, and relationship intelligence 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(
        f"list_{list_id}_fields:",
        filter=filter,
        includes=includes,
    )

    data = await self._client.get_all_pages(
        f"/lists/{list_id}/fields",
        params=params or None,
        cache_key=cache_key,
        cache_ttl=300,
    )

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

get_first() -> AffinityList | None async

Get the first list, or None if no lists exist.

Source code in affinity/services/lists.py

async def get_first(self) -> AffinityList | None:
    """Get the first list, or None if no lists exist."""
    page = await self.list(limit=1)
    return page.data[0] if page.data else None

get_saved_view(list_id: ListId, view_id: SavedViewId) -> SavedView async

Get a single saved view.

Source code in affinity/services/lists.py

async def get_saved_view(self, list_id: ListId, view_id: SavedViewId) -> SavedView:
    """Get a single saved view."""
    data = await self._client.get(f"/lists/{list_id}/saved-views/{view_id}")
    return _safe_model_validate(SavedView, data)

get_saved_views(list_id: ListId, *, limit: int | None = None, cursor: str | None = None) -> PaginatedResponse[SavedView] async

Get saved views for a list.

Parameters:

Name	Type	Description	Default
list_id	ListId	List id for the initial request.	required
limit	int | None	Maximum results per page.	None
cursor	str | None	Cursor to resume pagination (opaque; obtained from prior responses).	None

Source code in affinity/services/lists.py

async def get_saved_views(
    self,
    list_id: ListId,
    *,
    limit: int | None = None,
    cursor: str | None = None,
) -> PaginatedResponse[SavedView]:
    """
    Get saved views for a list.

    Args:
        list_id: List id for the initial request.
        limit: Maximum results per page.
        cursor: Cursor to resume pagination (opaque; obtained from prior responses).
    """
    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."
            )
        cursor_list_id = _saved_views_list_id_from_cursor(cursor)
        if cursor_list_id is not None and int(list_id) != cursor_list_id:
            raise ValueError(
                f"Cursor does not match list_id: cursor is for list {cursor_list_id}, "
                f"requested list_id is {int(list_id)}"
            )
        data = await self._client.get_url(cursor)
    else:
        if limit is not None and limit <= 0:
            raise ValueError("'limit' must be > 0")
        params: dict[str, Any] = {}
        if limit is not None:
            params["limit"] = limit
        data = await self._client.get(f"/lists/{list_id}/saved-views", params=params or None)

    return PaginatedResponse[SavedView](
        data=[_safe_model_validate(SavedView, v) for v in data.get("data", [])],
        pagination=_safe_model_validate(PaginationInfo, data.get("pagination", {})),
    )

get_size(list_id: ListId, *, force: bool = False) -> int async

Get accurate list size. Uses V1 API, cached for 5 minutes.

Adding or removing a row through this client (entries(list_id).add_* / delete, opportunities.create) drops the cached size; changes made elsewhere show after the cache expires, or with force=True.

Parameters:

Name	Type	Description	Default
list_id	ListId	The list ID.	required
force	bool	If True, bypass cache and fetch fresh value from API.	False

Note: The V2 API's listSize field is unreliable (often returns 0 for non-empty lists). This method uses the V1 API which returns accurate values.

Source code in affinity/services/lists.py

async def get_size(self, list_id: ListId, *, force: bool = False) -> int:
    """
    Get accurate list size. Uses V1 API, cached for 5 minutes.

    Adding or removing a row through this client (``entries(list_id).add_*`` / ``delete``,
    ``opportunities.create``) drops the cached size; changes made elsewhere show after the
    cache expires, or with ``force=True``.

    Args:
        list_id: The list ID.
        force: If True, bypass cache and fetch fresh value from API.

    Note: The V2 API's listSize field is unreliable (often returns 0 for
    non-empty lists). This method uses the V1 API which returns accurate values.
    """
    if not force and list_id in self._size_cache:
        cached_at, size = self._size_cache[list_id]
        if time.monotonic() - cached_at < _SIZE_CACHE_TTL:
            return size

    # V2 has no list size: read it from V1
    data = await self._client.get(f"/lists/{int(list_id)}", v1=True)
    size = _safe_model_validate(AffinityList, data)._list_size_hint
    self._size_cache[list_id] = (time.monotonic(), size)
    return size

iter() -> AsyncIterator[AffinityList]

Auto-paginate all lists.

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

Source code in affinity/services/lists.py

def iter(self) -> AsyncIterator[AffinityList]:
    """
    Auto-paginate all lists.

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

list(*, limit: int | None = None, cursor: str | None = None, term: str | None = None) -> PaginatedResponse[AffinityList] async

Get all lists accessible to you.

Parameters:

Name	Type	Description	Default
limit	int | None	Maximum results per page.	None
cursor	str | None	Cursor to resume pagination (opaque; obtained from prior responses).	None
term	str | None	Only lists whose name contains this text (case-insensitive substring, matched by Affinity). Kept across pages by the cursor.	None

Returns:

Type	Description
PaginatedResponse[AffinityList]	Paginated list of lists (without field metadata)

Source code in affinity/services/lists.py

async def list(
    self,
    *,
    limit: int | None = None,
    cursor: str | None = None,
    term: str | None = None,
) -> PaginatedResponse[AffinityList]:
    """
    Get all lists accessible to you.

    Args:
        limit: Maximum results per page.
        cursor: Cursor to resume pagination (opaque; obtained from prior responses).
        term: Only lists whose name contains this text (case-insensitive substring,
            matched by Affinity). Kept across pages by the cursor.

    Returns:
        Paginated list of lists (without field metadata)
    """
    term = term or None
    if cursor is not None:
        if limit is not None or term 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:
        if limit is not None and limit <= 0:
            raise ValueError("'limit' must be > 0")
        params: dict[str, Any] = {}
        if limit is not None:
            params["limit"] = limit
        if term is not None:
            params["term"] = term
        data = await self._client.get("/lists", params=params or None)
    return PaginatedResponse[AffinityList](
        data=[
            _safe_model_validate(AffinityList, list_item) for list_item in data.get("data", [])
        ],
        pagination=_safe_model_validate(PaginationInfo, data.get("pagination", {})),
    )

pages(*, limit: int | None = None, cursor: str | None = None, term: str | None = None) -> AsyncIterator[PaginatedResponse[AffinityList]] async

Iterate list pages (not items), yielding PaginatedResponse[AffinityList].

This is useful for ETL scripts that want checkpoint/resume via page.next_cursor. term keeps only lists whose name contains it (case-insensitive, matched by Affinity).

Source code in affinity/services/lists.py

async def pages(
    self,
    *,
    limit: int | None = None,
    cursor: str | None = None,
    term: str | None = None,
) -> AsyncIterator[PaginatedResponse[AffinityList]]:
    """
    Iterate list pages (not items), yielding `PaginatedResponse[AffinityList]`.

    This is useful for ETL scripts that want checkpoint/resume via `page.next_cursor`.
    ``term`` keeps only lists whose name contains it (case-insensitive, matched by Affinity).
    """
    term = term or None
    if cursor is not None and (limit is not None or term 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."
        )
    requested_cursor = cursor
    guard = _CursorGuard(cursor)
    page = (
        await self.list(limit=limit, term=term)
        if cursor is None
        else await self.list(cursor=cursor)
    )
    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)

resolve(*, name: str, list_type: ListType | None = None) -> AffinityList | None async

Find a single list by name (case-insensitive exact match; optionally filtered by type).

Notes: - Affinity narrows the lists by name first (term); the exact comparison is made here. If that finds nothing, every list is checked (see resolve_all()). - Matches are cached in-memory on this service instance (a miss is not). If you call this frequently, reuse the client, or persist the resolved ListId in your own configuration.

If multiple matches exist, returns the first match in server-provided order.

Source code in affinity/services/lists.py

async def resolve(
    self,
    *,
    name: str,
    list_type: ListType | None = None,
) -> AffinityList | None:
    """
    Find a single list by name (case-insensitive exact match; optionally filtered by type).

    Notes:
    - Affinity narrows the lists by name first (``term``); the exact comparison is made
      here. If that finds nothing, every list is checked (see `resolve_all()`).
    - Matches are cached in-memory on this service instance (a miss is not). If you call
      this frequently, reuse the client, or persist the resolved `ListId` in your own
      configuration.

    If multiple matches exist, returns the first match in server-provided order.
    """
    key = (name.lower(), list_type)
    if key in self._resolve_cache:
        return self._resolve_cache[key]

    matches = await self.resolve_all(name=name, list_type=list_type)
    found = matches[0] if matches else None
    if found is not None:  # a miss isn't cached: the list may be created later
        self._resolve_cache[key] = found
    return found

resolve_all(*, name: str, list_type: ListType | None = None) -> builtins.list[AffinityList] async

Find all lists matching a name (case-insensitive exact match; optionally by type).

Notes: - For an ASCII name, Affinity first narrows the lists by name (term, a case-insensitive substring match) and the exact comparison is made here. If that yields no match, every list is checked, so a difference in how Affinity folds case or whitespace can't hide a list. Non-ASCII names always check every list. - Unlike resolve(), this does not cache results.

Source code in affinity/services/lists.py

async def resolve_all(
    self,
    *,
    name: str,
    list_type: ListType | None = None,
) -> builtins.list[AffinityList]:
    """
    Find all lists matching a name (case-insensitive exact match; optionally by type).

    Notes:
    - For an ASCII name, Affinity first narrows the lists by name (``term``, a
      case-insensitive substring match) and the exact comparison is made here. If that
      yields no match, every list is checked, so a difference in how Affinity folds case
      or whitespace can't hide a list. Non-ASCII names always check every list.
    - Unlike `resolve()`, this does not cache results.
    """
    term = name.strip()
    if term and term.isascii():
        candidates = [item async for page in self.pages(term=term) for item in page.data]
        matches = _exact_name_matches(candidates, name, list_type)
        if matches:
            return matches
    return _exact_name_matches([item async for item in self.all()], name, list_type)

saved_views_all(list_id: ListId) -> AsyncIterator[SavedView] async

Iterate all saved views for a list.

Source code in affinity/services/lists.py

async def saved_views_all(self, list_id: ListId) -> AsyncIterator[SavedView]:
    """Iterate all saved views for a list."""
    async for page in self.saved_views_pages(list_id):
        for item in page.data:
            yield item

saved_views_pages(list_id: ListId, *, limit: int | None = None, cursor: str | None = None) -> AsyncIterator[PaginatedResponse[SavedView]] async

Iterate saved view pages, yielding PaginatedResponse[SavedView].

Source code in affinity/services/lists.py

async def saved_views_pages(
    self,
    list_id: ListId,
    *,
    limit: int | None = None,
    cursor: str | None = None,
) -> AsyncIterator[PaginatedResponse[SavedView]]:
    """Iterate saved view pages, yielding `PaginatedResponse[SavedView]`."""
    if cursor is not None and 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."
        )
    requested_cursor = cursor
    guard = _CursorGuard(cursor)
    page = (
        await self.get_saved_views(list_id, limit=limit)
        if cursor is None
        else await self.get_saved_views(list_id, cursor=cursor)
    )
    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.get_saved_views(list_id, cursor=requested_cursor)

update_field_dropdown_option(list_id: ListId, field_id: AnyFieldId, option_id: int, *, text: str | None = None, rank: int | None = None, color: DropdownOptionColorName | None = None, status_category: DropdownStatusCategory | None = None, win_rate: int | None = None) -> DropdownOption async

Change an option of a list field (only what you pass).

A dropdown option can change its text; ranked-dropdown also rank and color; status-dropdown also status category and win rate. Renaming changes the value shown on every entry that has this option. Returns the option as read back afterwards.

Source code in affinity/services/lists.py

async def update_field_dropdown_option(
    self,
    list_id: ListId,
    field_id: AnyFieldId,
    option_id: int,
    *,
    text: str | None = None,
    rank: int | None = None,
    color: DropdownOptionColorName | None = None,
    status_category: DropdownStatusCategory | None = None,
    win_rate: int | None = None,
) -> DropdownOption:
    """Change an option of a list field (only what you pass).

    A ``dropdown`` option can change its text; ``ranked-dropdown`` also rank and color;
    ``status-dropdown`` also status category and win rate. Renaming changes the value shown
    on every entry that has this option. Returns the option as read back afterwards.
    """
    body = update_body(
        text=text,
        rank=rank,
        color=color,
        status_category=status_category,
        win_rate=win_rate,
    )
    await self._client.post(
        option_path(list_id, field_id, option_id),
        json=body,
        min_api_version=DROPDOWN_OPTION_WRITES_MIN_API_VERSION,
    )
    # The API answers 204 with no body: read the option back
    return DropdownOption.model_validate(
        await self._client.get(
            option_path(list_id, field_id, option_id),
            min_api_version=DROPDOWN_OPTION_WRITES_MIN_API_VERSION,
        )
    )