Filters

Filter builder for V2 API filtering support.

Provides a type-safe, Pythonic way to build filter expressions for V2 list endpoints. The builder handles proper escaping and quoting of user inputs.

Example

from affinity.filters import Filter, F

Using the builder (recommended)

filter = ( F.field("name").contains("Acme") & F.field("status").equals("Active") ) companies = client.companies.list(filter=filter)

Or build complex filters

filter = ( (F.field("name").contains("Corp") | F.field("name").contains("Inc")) & ~F.field("archived").equals(True) )

Raw filter string escape hatch (power users)

companies = client.companies.list(filter='name =~ "Acme"')

AndExpression dataclass

Bases: FilterExpression

& combination of two expressions.

Source code in affinity/filters.py

@dataclass
class AndExpression(FilterExpression):
    """`&` combination of two expressions."""

    left: FilterExpression
    right: FilterExpression

    def to_string(self) -> str:
        left_str = self.left.to_string()
        right_str = self.right.to_string()
        # Wrap in parentheses for correct precedence
        return f"({left_str}) & ({right_str})"

    def matches(self, entity: dict[str, Any]) -> bool:
        """Both sides must match."""
        return self.left.matches(entity) and self.right.matches(entity)

matches(entity: dict[str, Any]) -> bool

Both sides must match.

Source code in affinity/filters.py

def matches(self, entity: dict[str, Any]) -> bool:
    """Both sides must match."""
    return self.left.matches(entity) and self.right.matches(entity)

FieldBuilder

Builder for field-based filter expressions.

Source code in affinity/filters.py

class FieldBuilder:
    """Builder for field-based filter expressions."""

    def __init__(self, field_name: str):
        self._field_name = field_name

    def equals(self, value: Any) -> FieldComparison:
        """Field equals value (exact match)."""
        return FieldComparison(self._field_name, "=", value)

    def not_equals(self, value: Any) -> FieldComparison:
        """Field does not equal value."""
        return FieldComparison(self._field_name, "!=", value)

    def contains(self, value: str) -> FieldComparison:
        """Field contains substring (case-insensitive)."""
        return FieldComparison(self._field_name, "=~", value)

    def starts_with(self, value: str) -> FieldComparison:
        """Field starts with prefix."""
        return FieldComparison(self._field_name, "=^", value)

    def ends_with(self, value: str) -> FieldComparison:
        """Field ends with suffix."""
        return FieldComparison(self._field_name, "=$", value)

    def greater_than(self, value: int | float | datetime | date) -> FieldComparison:
        """Field is greater than value."""
        return FieldComparison(self._field_name, ">", value)

    def greater_than_or_equal(self, value: int | float | datetime | date) -> FieldComparison:
        """Field is greater than or equal to value."""
        return FieldComparison(self._field_name, ">=", value)

    def less_than(self, value: int | float | datetime | date) -> FieldComparison:
        """Field is less than value."""
        return FieldComparison(self._field_name, "<", value)

    def less_than_or_equal(self, value: int | float | datetime | date) -> FieldComparison:
        """Field is less than or equal to value."""
        return FieldComparison(self._field_name, "<=", value)

    def is_null(self) -> FieldComparison:
        """Field is null."""
        return FieldComparison(self._field_name, "!=", RawToken("*"))

    def is_not_null(self) -> FieldComparison:
        """Field is not null."""
        return FieldComparison(self._field_name, "=", RawToken("*"))

    def in_list(self, values: list[Any]) -> FilterExpression:
        """Field value is in the given list (OR of equals)."""
        if not values:
            raise ValueError("in_list() requires at least one value")
        expressions: list[FilterExpression] = [self.equals(v) for v in values]
        result: FilterExpression = expressions[0]
        for expr in expressions[1:]:
            result = result | expr
        return result

contains(value: str) -> FieldComparison

Field contains substring (case-insensitive).

Source code in affinity/filters.py

def contains(self, value: str) -> FieldComparison:
    """Field contains substring (case-insensitive)."""
    return FieldComparison(self._field_name, "=~", value)

ends_with(value: str) -> FieldComparison

Field ends with suffix.

Source code in affinity/filters.py

def ends_with(self, value: str) -> FieldComparison:
    """Field ends with suffix."""
    return FieldComparison(self._field_name, "=$", value)

equals(value: Any) -> FieldComparison

Field equals value (exact match).

Source code in affinity/filters.py

def equals(self, value: Any) -> FieldComparison:
    """Field equals value (exact match)."""
    return FieldComparison(self._field_name, "=", value)

greater_than(value: int | float | datetime | date) -> FieldComparison

Field is greater than value.

Source code in affinity/filters.py

def greater_than(self, value: int | float | datetime | date) -> FieldComparison:
    """Field is greater than value."""
    return FieldComparison(self._field_name, ">", value)

greater_than_or_equal(value: int | float | datetime | date) -> FieldComparison

Field is greater than or equal to value.

Source code in affinity/filters.py

def greater_than_or_equal(self, value: int | float | datetime | date) -> FieldComparison:
    """Field is greater than or equal to value."""
    return FieldComparison(self._field_name, ">=", value)

in_list(values: list[Any]) -> FilterExpression

Field value is in the given list (OR of equals).

Source code in affinity/filters.py

def in_list(self, values: list[Any]) -> FilterExpression:
    """Field value is in the given list (OR of equals)."""
    if not values:
        raise ValueError("in_list() requires at least one value")
    expressions: list[FilterExpression] = [self.equals(v) for v in values]
    result: FilterExpression = expressions[0]
    for expr in expressions[1:]:
        result = result | expr
    return result

is_not_null() -> FieldComparison

Field is not null.

Source code in affinity/filters.py

def is_not_null(self) -> FieldComparison:
    """Field is not null."""
    return FieldComparison(self._field_name, "=", RawToken("*"))

is_null() -> FieldComparison

Field is null.

Source code in affinity/filters.py

def is_null(self) -> FieldComparison:
    """Field is null."""
    return FieldComparison(self._field_name, "!=", RawToken("*"))

less_than(value: int | float | datetime | date) -> FieldComparison

Field is less than value.

Source code in affinity/filters.py

def less_than(self, value: int | float | datetime | date) -> FieldComparison:
    """Field is less than value."""
    return FieldComparison(self._field_name, "<", value)

less_than_or_equal(value: int | float | datetime | date) -> FieldComparison

Field is less than or equal to value.

Source code in affinity/filters.py

def less_than_or_equal(self, value: int | float | datetime | date) -> FieldComparison:
    """Field is less than or equal to value."""
    return FieldComparison(self._field_name, "<=", value)

not_equals(value: Any) -> FieldComparison

Field does not equal value.

Source code in affinity/filters.py

def not_equals(self, value: Any) -> FieldComparison:
    """Field does not equal value."""
    return FieldComparison(self._field_name, "!=", value)

starts_with(value: str) -> FieldComparison

Field starts with prefix.

Source code in affinity/filters.py

def starts_with(self, value: str) -> FieldComparison:
    """Field starts with prefix."""
    return FieldComparison(self._field_name, "=^", value)

FieldComparison dataclass

Bases: FilterExpression

A comparison operation on a field.

Source code in affinity/filters.py

@dataclass
class FieldComparison(FilterExpression):
    """A comparison operation on a field."""

    field_name: str
    operator: str
    value: Any

    def to_string(self) -> str:
        formatted_value = _format_value(self.value)
        return f"{self.field_name} {self.operator} {formatted_value}"

    def matches(self, entity: dict[str, Any]) -> bool:
        """Evaluate field comparison against an entity dict.

        For multi-select dropdown fields (arrays), the operators have special semantics:
        - `=` with scalar: checks if value is IN the array (membership)
        - `=` with list: checks set equality (order-insensitive)
        - `!=` with scalar: checks if value is NOT in the array
        - `!=` with list: checks set inequality
        - `=~` (contains): checks if any array element contains the substring
        - `=^` (starts_with): checks if any array element starts with the prefix
        - `=$` (ends_with): checks if any array element ends with the suffix
        - `>`, `>=`, `<`, `<=`: numeric/date comparisons

        Uses the shared compare module for consistent behavior across SDK and Query tool.
        """
        field_value = _get_entity_value(entity, self.field_name)

        # Normalize dropdown dicts and multi-select arrays
        field_value = normalize_value(field_value)

        # Handle NULL checks (Affinity convention: =* means NOT NULL, !=* means IS NULL)
        if isinstance(self.value, RawToken) and self.value.token == "*":
            if self.operator == "=":
                return compare_values(field_value, None, "is_not_null")
            elif self.operator == "!=":
                return compare_values(field_value, None, "is_null")

        # Extract target value
        target = self.value if not isinstance(self.value, RawToken) else self.value.token

        # Map SDK operator symbol to canonical operator name
        try:
            canonical_op = map_operator(self.operator)
        except ValueError:
            raise ValueError(
                f"Unsupported operator '{self.operator}' for client-side matching. "
                f"Supported operators: =, !=, =~, =^, =$, >, >=, <, <=, "
                f"contains, starts_with, ends_with, gt, gte, lt, lte, "
                f"is null, is not null, is empty, "
                f"in, between, has_any, has_all, contains_any, contains_all, starts_word"
            ) from None

        return compare_values(field_value, target, canonical_op)

matches(entity: dict[str, Any]) -> bool

Evaluate field comparison against an entity dict.

For multi-select dropdown fields (arrays), the operators have special semantics: - = with scalar: checks if value is IN the array (membership) - = with list: checks set equality (order-insensitive) - != with scalar: checks if value is NOT in the array - != with list: checks set inequality - =~ (contains): checks if any array element contains the substring - =^ (starts_with): checks if any array element starts with the prefix - =$ (ends_with): checks if any array element ends with the suffix - >, >=, <, <=: numeric/date comparisons

Uses the shared compare module for consistent behavior across SDK and Query tool.

Source code in affinity/filters.py

def matches(self, entity: dict[str, Any]) -> bool:
    """Evaluate field comparison against an entity dict.

    For multi-select dropdown fields (arrays), the operators have special semantics:
    - `=` with scalar: checks if value is IN the array (membership)
    - `=` with list: checks set equality (order-insensitive)
    - `!=` with scalar: checks if value is NOT in the array
    - `!=` with list: checks set inequality
    - `=~` (contains): checks if any array element contains the substring
    - `=^` (starts_with): checks if any array element starts with the prefix
    - `=$` (ends_with): checks if any array element ends with the suffix
    - `>`, `>=`, `<`, `<=`: numeric/date comparisons

    Uses the shared compare module for consistent behavior across SDK and Query tool.
    """
    field_value = _get_entity_value(entity, self.field_name)

    # Normalize dropdown dicts and multi-select arrays
    field_value = normalize_value(field_value)

    # Handle NULL checks (Affinity convention: =* means NOT NULL, !=* means IS NULL)
    if isinstance(self.value, RawToken) and self.value.token == "*":
        if self.operator == "=":
            return compare_values(field_value, None, "is_not_null")
        elif self.operator == "!=":
            return compare_values(field_value, None, "is_null")

    # Extract target value
    target = self.value if not isinstance(self.value, RawToken) else self.value.token

    # Map SDK operator symbol to canonical operator name
    try:
        canonical_op = map_operator(self.operator)
    except ValueError:
        raise ValueError(
            f"Unsupported operator '{self.operator}' for client-side matching. "
            f"Supported operators: =, !=, =~, =^, =$, >, >=, <, <=, "
            f"contains, starts_with, ends_with, gt, gte, lt, lte, "
            f"is null, is not null, is empty, "
            f"in, between, has_any, has_all, contains_any, contains_all, starts_word"
        ) from None

    return compare_values(field_value, target, canonical_op)

Filter

Factory for building filter expressions.

Example

Simple comparison

Filter.field("name").contains("Acme")

Complex boolean logic

(Filter.field("status").equals("Active") & Filter.field("type").in_list(["customer", "prospect"]))

Negation

~Filter.field("archived").equals(True)

Source code in affinity/filters.py

class Filter:
    """
    Factory for building filter expressions.

    Example:
        # Simple comparison
        Filter.field("name").contains("Acme")

        # Complex boolean logic
        (Filter.field("status").equals("Active") &
         Filter.field("type").in_list(["customer", "prospect"]))

        # Negation
        ~Filter.field("archived").equals(True)
    """

    @staticmethod
    def field(name: str) -> FieldBuilder:
        """Start building a filter on a field."""
        return FieldBuilder(name)

    @staticmethod
    def raw(expression: str) -> RawFilter:
        """
        Create a raw filter expression (escape hatch).

        Use this when you need filter syntax not supported by the builder.
        The expression is passed directly to the API without modification.

        Args:
            expression: Raw filter string (e.g., 'name =~ "Acme"')
        """
        return RawFilter(expression)

    @staticmethod
    def and_(*expressions: FilterExpression) -> FilterExpression:
        """Combine multiple expressions with `&`."""
        if not expressions:
            raise ValueError("and_() requires at least one expression")
        result = expressions[0]
        for expr in expressions[1:]:
            result = result & expr
        return result

    @staticmethod
    def or_(*expressions: FilterExpression) -> FilterExpression:
        """Combine multiple expressions with `|`."""
        if not expressions:
            raise ValueError("or_() requires at least one expression")
        result = expressions[0]
        for expr in expressions[1:]:
            result = result | expr
        return result

and_(*expressions: FilterExpression) -> FilterExpression staticmethod

Combine multiple expressions with &.

Source code in affinity/filters.py

@staticmethod
def and_(*expressions: FilterExpression) -> FilterExpression:
    """Combine multiple expressions with `&`."""
    if not expressions:
        raise ValueError("and_() requires at least one expression")
    result = expressions[0]
    for expr in expressions[1:]:
        result = result & expr
    return result

field(name: str) -> FieldBuilder staticmethod

Start building a filter on a field.

Source code in affinity/filters.py

@staticmethod
def field(name: str) -> FieldBuilder:
    """Start building a filter on a field."""
    return FieldBuilder(name)

or_(*expressions: FilterExpression) -> FilterExpression staticmethod

Combine multiple expressions with |.

Source code in affinity/filters.py

@staticmethod
def or_(*expressions: FilterExpression) -> FilterExpression:
    """Combine multiple expressions with `|`."""
    if not expressions:
        raise ValueError("or_() requires at least one expression")
    result = expressions[0]
    for expr in expressions[1:]:
        result = result | expr
    return result

raw(expression: str) -> RawFilter staticmethod

Create a raw filter expression (escape hatch).

Use this when you need filter syntax not supported by the builder. The expression is passed directly to the API without modification.

Parameters:

Name	Type	Description	Default
expression	str	Raw filter string (e.g., 'name =~ "Acme"')	required

Source code in affinity/filters.py

@staticmethod
def raw(expression: str) -> RawFilter:
    """
    Create a raw filter expression (escape hatch).

    Use this when you need filter syntax not supported by the builder.
    The expression is passed directly to the API without modification.

    Args:
        expression: Raw filter string (e.g., 'name =~ "Acme"')
    """
    return RawFilter(expression)

FilterExpression

Bases: ABC

Base class for filter expressions.

Source code in affinity/filters.py

class FilterExpression(ABC):
    """Base class for filter expressions."""

    @abstractmethod
    def to_string(self) -> str:
        """Convert the expression to a filter string."""
        ...

    @abstractmethod
    def matches(self, entity: dict[str, Any]) -> bool:
        """
        Evaluate filter against an entity dict (client-side).

        Used for --expand-filter in list export where filtering happens
        after fetching data from the API.
        """
        ...

    def __and__(self, other: FilterExpression) -> FilterExpression:
        """Combine two expressions with `&`."""
        return AndExpression(self, other)

    def __or__(self, other: FilterExpression) -> FilterExpression:
        """Combine two expressions with `|`."""
        return OrExpression(self, other)

    def __invert__(self) -> FilterExpression:
        """Negate the expression with `!`."""
        return NotExpression(self)

    def __str__(self) -> str:
        return self.to_string()

    def __repr__(self) -> str:
        return f"Filter({self.to_string()!r})"

__and__(other: FilterExpression) -> FilterExpression

Combine two expressions with &.

Source code in affinity/filters.py

def __and__(self, other: FilterExpression) -> FilterExpression:
    """Combine two expressions with `&`."""
    return AndExpression(self, other)

__invert__() -> FilterExpression

Negate the expression with !.

Source code in affinity/filters.py

def __invert__(self) -> FilterExpression:
    """Negate the expression with `!`."""
    return NotExpression(self)

__or__(other: FilterExpression) -> FilterExpression

Combine two expressions with |.

Source code in affinity/filters.py

def __or__(self, other: FilterExpression) -> FilterExpression:
    """Combine two expressions with `|`."""
    return OrExpression(self, other)

matches(entity: dict[str, Any]) -> bool abstractmethod

Evaluate filter against an entity dict (client-side).

Used for --expand-filter in list export where filtering happens after fetching data from the API.

Source code in affinity/filters.py

@abstractmethod
def matches(self, entity: dict[str, Any]) -> bool:
    """
    Evaluate filter against an entity dict (client-side).

    Used for --expand-filter in list export where filtering happens
    after fetching data from the API.
    """
    ...

to_string() -> str abstractmethod

Convert the expression to a filter string.

Source code in affinity/filters.py

@abstractmethod
def to_string(self) -> str:
    """Convert the expression to a filter string."""
    ...

NotExpression dataclass

Bases: FilterExpression

! negation of an expression.

Source code in affinity/filters.py

@dataclass
class NotExpression(FilterExpression):
    """`!` negation of an expression."""

    expr: FilterExpression

    def to_string(self) -> str:
        return f"!({self.expr.to_string()})"

    def matches(self, entity: dict[str, Any]) -> bool:
        """Invert the inner expression."""
        return not self.expr.matches(entity)

matches(entity: dict[str, Any]) -> bool

Invert the inner expression.

Source code in affinity/filters.py

def matches(self, entity: dict[str, Any]) -> bool:
    """Invert the inner expression."""
    return not self.expr.matches(entity)

OrExpression dataclass

Bases: FilterExpression

| combination of two expressions.

Source code in affinity/filters.py

@dataclass
class OrExpression(FilterExpression):
    """`|` combination of two expressions."""

    left: FilterExpression
    right: FilterExpression

    def to_string(self) -> str:
        left_str = self.left.to_string()
        right_str = self.right.to_string()
        return f"({left_str}) | ({right_str})"

    def matches(self, entity: dict[str, Any]) -> bool:
        """Either side must match."""
        return self.left.matches(entity) or self.right.matches(entity)

matches(entity: dict[str, Any]) -> bool

Either side must match.

Source code in affinity/filters.py

def matches(self, entity: dict[str, Any]) -> bool:
    """Either side must match."""
    return self.left.matches(entity) or self.right.matches(entity)

RawFilter dataclass

Bases: FilterExpression

A raw filter string (escape hatch for power users).

Source code in affinity/filters.py

@dataclass
class RawFilter(FilterExpression):
    """A raw filter string (escape hatch for power users)."""

    expression: str

    def to_string(self) -> str:
        return self.expression

    def matches(self, entity: dict[str, Any]) -> bool:
        """RawFilter cannot be evaluated client-side."""
        raise NotImplementedError(
            "RawFilter cannot be evaluated client-side. "
            "Use structured filter expressions for --expand-filter."
        )

matches(entity: dict[str, Any]) -> bool

RawFilter cannot be evaluated client-side.

Source code in affinity/filters.py

def matches(self, entity: dict[str, Any]) -> bool:
    """RawFilter cannot be evaluated client-side."""
    raise NotImplementedError(
        "RawFilter cannot be evaluated client-side. "
        "Use structured filter expressions for --expand-filter."
    )

RawToken dataclass

A raw token inserted into a filter expression without quoting.

Used for special Affinity Filtering Language literals like *.

Source code in affinity/filters.py

@dataclass(frozen=True)
class RawToken:
    """
    A raw token inserted into a filter expression without quoting.

    Used for special Affinity Filtering Language literals like `*`.
    """

    token: str

parse(filter_string: str) -> FilterExpression

Parse a filter string into a FilterExpression AST.

This function converts a human-readable filter string into a structured FilterExpression that can be used for client-side filtering with matches().

Parameters:

Name	Type	Description	Default
filter_string	str	The filter expression to parse	required

Returns:

Type	Description
FilterExpression	A FilterExpression AST representing the filter

Raises:

Type	Description
ValueError	If the filter string is invalid

Examples:

>>> expr = parse('name = "Alice"')
>>> expr.matches({"name": "Alice"})
True

>>> expr = parse('status = Active | status = Pending')
>>> expr.matches({"status": "Active"})
True

>>> expr = parse('email = *')  # IS NOT NULL
>>> expr.matches({"email": "test@example.com"})
True

>>> expr = parse('email != *')  # IS NULL
>>> expr.matches({"email": None})
True

Source code in affinity/filters.py

def parse(filter_string: str) -> FilterExpression:
    """
    Parse a filter string into a FilterExpression AST.

    This function converts a human-readable filter string into a structured
    FilterExpression that can be used for client-side filtering with matches().

    Args:
        filter_string: The filter expression to parse

    Returns:
        A FilterExpression AST representing the filter

    Raises:
        ValueError: If the filter string is invalid

    Examples:
        >>> expr = parse('name = "Alice"')
        >>> expr.matches({"name": "Alice"})
        True

        >>> expr = parse('status = Active | status = Pending')
        >>> expr.matches({"status": "Active"})
        True

        >>> expr = parse('email = *')  # IS NOT NULL
        >>> expr.matches({"email": "test@example.com"})
        True

        >>> expr = parse('email != *')  # IS NULL
        >>> expr.matches({"email": None})
        True
    """
    if not filter_string or not filter_string.strip():
        raise ValueError("Empty filter expression")

    tokenizer = _Tokenizer(filter_string)
    tokens = tokenizer.tokenize()
    parser = _Parser(tokens)
    return parser.parse()

Search criteria (affinity.criteria)

Conditions and sorts for Affinity's search: companies.find, persons.find and lists.entries(list_id).find. Affinity applies them on its side, so only matching records are downloaded. affinity.filters (above) matches records after they're downloaded.

• Condition(field_id, operator, value=..., value_type=None, attribute_id=None): one test on a field, with Affinity's operator name (is-any-of, has-any-of, contains, is-greater-than, is-between, is-after, is-within-the-last, is-empty, ...).
• all_of(...) / any_of(...): AND / OR groups (up to 50 items each), nestable.
• Sort(field_id, "asc" | "desc"): up to 5 per search.
• SearchCriteria: filters, sorts and a search term together.

See Companies service for the find methods, paging and permissions.

Search criteria for the V2 search endpoints (beta): companies.find, persons.find and lists.entries(list_id).find.

Affinity applies these filters and sorts on its side, so only matching records are downloaded.

Example

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

page = client.companies.find( all_of( Condition("field-498220", "is-any-of", [3, 4]), # dropdown option ids any_of( Condition("affinity-data-last-funding-date", "is-after", date(2025, 1, 1)), Condition("dealroom-twitter-followers-last-month", "is-greater-than", 1000), ), ), sort=[Sort("dealroom-twitter-followers-last-month", "desc")], )

Each condition needs the field's filter value type. When it isn't given, it's read from the fields metadata (one cached request), which also checks the operator against the field.

This is separate from :mod:affinity.filters (the text filter language that filters records after they're downloaded).

Condition dataclass

One filter: field_id, an operator and (except for is-empty / is-not-empty) a value.

Values are written plainly and wrapped for the API:

• company / person / list fields: ids (7 or [7, 8])
• dropdown / ranked-dropdown fields: option ids
• location fields: dicts with country, state, city, continent, streetAddress
• filterable-text fields: strings; text fields: one string
• numbers: a number, or (low, high) for is-between
• dates: :class:~datetime.date (or YYYY-MM-DD), (start, end) for is-between; relative windows as (amount, unit), e.g. (6, "month"); is-between-relative as (start, end, unit)

value_type is the filter type (date for datetime fields); omitted, it's read from the fields metadata. attribute_id is required for fields filtered by an attribute (relationship-intelligence fields such as date-of-activity).

Source code in affinity/criteria.py

@dataclass(frozen=True)
class Condition:
    """One filter: ``field_id``, an ``operator`` and (except for is-empty / is-not-empty) a
    ``value``.

    Values are written plainly and wrapped for the API:

    - company / person / list fields: ids (``7`` or ``[7, 8]``)
    - dropdown / ranked-dropdown fields: option ids
    - location fields: dicts with ``country``, ``state``, ``city``, ``continent``,
      ``streetAddress``
    - filterable-text fields: strings; text fields: one string
    - numbers: a number, or ``(low, high)`` for ``is-between``
    - dates: :class:`~datetime.date` (or ``YYYY-MM-DD``), ``(start, end)`` for ``is-between``;
      relative windows as ``(amount, unit)``, e.g. ``(6, "month")``; ``is-between-relative`` as
      ``(start, end, unit)``

    ``value_type`` is the filter type (``date`` for datetime fields); omitted, it's read from
    the fields metadata. ``attribute_id`` is required for fields filtered by an attribute
    (relationship-intelligence fields such as ``date-of-activity``).
    """

    field_id: str
    operator: str
    value: Any = _UNSET
    value_type: str | None = None
    attribute_id: str | None = None

    def __post_init__(self) -> None:
        object.__setattr__(self, "field_id", str(self.field_id))

FilterGroup dataclass

Conditions or groups combined with and / or (at most 50 items).

Source code in affinity/criteria.py

@dataclass(frozen=True)
class FilterGroup:
    """Conditions or groups combined with ``and`` / ``or`` (at most 50 items)."""

    operator: Literal["and", "or"]
    items: tuple[Condition | FilterGroup, ...]

    def __post_init__(self) -> None:
        if self.operator not in ("and", "or"):
            raise ValueError("FilterGroup operator must be 'and' or 'or'")
        if len(self.items) > MAX_GROUP_ITEMS:
            raise ValueError(f"A filter group takes at most {MAX_GROUP_ITEMS} items")
        for item in self.items:
            if not isinstance(item, (Condition, FilterGroup)):
                raise TypeError(f"Expected a Condition or a group, got {type(item).__name__}")

SearchCriteria dataclass

What to search for: filters, sorts and a keyword (term, 3+ characters, matched against names, domains and emails, plus term_fields).

Source code in affinity/criteria.py

@dataclass(frozen=True)
class SearchCriteria:
    """What to search for: filters, sorts and a keyword (``term``, 3+ characters, matched
    against names, domains and emails, plus ``term_fields``)."""

    where: Condition | FilterGroup | None = None
    sort: tuple[Sort, ...] = field(default_factory=tuple)
    term: str | None = None
    term_fields: tuple[str, ...] = field(default_factory=tuple)

    def __post_init__(self) -> None:
        object.__setattr__(self, "sort", tuple(self.sort))
        object.__setattr__(self, "term_fields", tuple(str(f) for f in self.term_fields))
        if len(self.sort) > MAX_SORTS:
            raise ValueError(f"At most {MAX_SORTS} sorts")
        if self.term is not None and len(self.term.strip()) < MIN_TERM_LENGTH:
            raise ValueError(f"'term' must be at least {MIN_TERM_LENGTH} characters")
        if self.term_fields and self.term is None:
            raise ValueError("'term_fields' needs a 'term'")
        if len(self.term_fields) > MAX_VALUES:
            raise ValueError(f"At most {MAX_VALUES} 'term_fields'")

    def conditions(self) -> list[Condition]:
        """Every condition in ``where``."""
        return list(_walk(self.where)) if self.where is not None else []

    def needs_fields(self) -> bool:
        """True when a condition has no ``value_type`` (the fields metadata is needed)."""
        return any(c.value_type is None for c in self.conditions())

    def to_body(self, fields: Mapping[str, FieldMetadata] | None = None) -> dict[str, Any]:
        """The request body. ``fields`` (by field id, with ``filterability``) supplies missing
        value types and checks each operator against the field."""
        body: dict[str, Any] = {}
        if self.where is not None:
            group = self.where if isinstance(self.where, FilterGroup) else all_of(self.where)
            body["filters"] = _group_body(group, fields)
        if self.sort:
            body["sorts"] = [s.to_body() for s in self.sort]
        if self.term is not None:
            term: dict[str, Any] = {"term": self.term}
            if self.term_fields:
                term["fieldIds"] = list(self.term_fields)
            body["search"] = term
        return body

conditions() -> list[Condition]

Every condition in where.

Source code in affinity/criteria.py

def conditions(self) -> list[Condition]:
    """Every condition in ``where``."""
    return list(_walk(self.where)) if self.where is not None else []

needs_fields() -> bool

True when a condition has no value_type (the fields metadata is needed).

Source code in affinity/criteria.py

def needs_fields(self) -> bool:
    """True when a condition has no ``value_type`` (the fields metadata is needed)."""
    return any(c.value_type is None for c in self.conditions())

to_body(fields: Mapping[str, FieldMetadata] | None = None) -> dict[str, Any]

The request body. fields (by field id, with filterability) supplies missing value types and checks each operator against the field.

Source code in affinity/criteria.py

def to_body(self, fields: Mapping[str, FieldMetadata] | None = None) -> dict[str, Any]:
    """The request body. ``fields`` (by field id, with ``filterability``) supplies missing
    value types and checks each operator against the field."""
    body: dict[str, Any] = {}
    if self.where is not None:
        group = self.where if isinstance(self.where, FilterGroup) else all_of(self.where)
        body["filters"] = _group_body(group, fields)
    if self.sort:
        body["sorts"] = [s.to_body() for s in self.sort]
    if self.term is not None:
        term: dict[str, Any] = {"term": self.term}
        if self.term_fields:
            term["fieldIds"] = list(self.term_fields)
        body["search"] = term
    return body

Sort dataclass

Sort by a field (at most 5 sorts, applied in order). Empty values come last in both directions; ties are ordered by id.

Source code in affinity/criteria.py

@dataclass(frozen=True)
class Sort:
    """Sort by a field (at most 5 sorts, applied in order). Empty values come last in both
    directions; ties are ordered by id."""

    field_id: str
    direction: Literal["asc", "desc"] = "asc"
    attribute_id: str | None = None

    def __post_init__(self) -> None:
        object.__setattr__(self, "field_id", str(self.field_id))
        if self.direction not in ("asc", "desc"):
            raise ValueError("Sort direction must be 'asc' or 'desc'")

    def to_body(self) -> dict[str, Any]:
        body: dict[str, Any] = {"fieldId": self.field_id, "direction": self.direction}
        if self.attribute_id is not None:
            body["attributeId"] = self.attribute_id
        return body

all_of(*items: Condition | FilterGroup) -> FilterGroup

Every item must match.

Source code in affinity/criteria.py

def all_of(*items: Condition | FilterGroup) -> FilterGroup:
    """Every item must match."""
    return FilterGroup("and", tuple(items))

any_of(*items: Condition | FilterGroup) -> FilterGroup

At least one item must match.

Source code in affinity/criteria.py

def any_of(*items: Condition | FilterGroup) -> FilterGroup:
    """At least one item must match."""
    return FilterGroup("or", tuple(items))

filter_value_type(field_value_type: str) -> str | None

The search filter value type for a field's metadata value type, or None when that type has no value filter (e.g. interaction: filter it by an attribute instead).

Source code in affinity/criteria.py

def filter_value_type(field_value_type: str) -> str | None:
    """The search filter value type for a field's metadata value type, or ``None`` when that
    type has no value filter (e.g. ``interaction``: filter it by an attribute instead)."""
    vt = _FIELD_TO_FILTER_TYPE.get(field_value_type, field_value_type)
    return vt if vt in OPERATORS else None