popoto.fields.indexed_field_mixin

popoto.fields.indexed_field_mixin

Indexed Field Mixin - Secondary Indexing for Non-Key Fields

This module provides the IndexedFieldMixin class, which enables exact-match secondary indexing on fields that are NOT part of the model's Redis key.

Design Philosophy

KeyField conflates two concerns: identity (forming the Redis storage key) and indexing (enabling queries). IndexedFieldMixin decouples these by providing Set-based indexing without making the field part of the Redis key (identity).

This allows developers to query on fields like email, status, or category without making them part of the model's identity.

The implementation follows the exact same pattern as KeyFieldMixin: - on_save(): Maintains a Redis Set at $IndexF:Model:field_name:value - on_delete(): Removes from the Set - filter_query(): Uses SMEMBERS/SUNION for lookups - get_filter_query_params(): Declares supported query lookups

Index Key Pattern

$IndexF:ModelName:field_name:value -> Set of redis_keys

This mirrors the $KeyF pattern used by KeyFieldMixin but uses the $IndexF prefix (auto-generated by FieldBase metaclass via field_class_key).

Usage

from popoto import Model, Field
from popoto.fields.shortcuts import IndexedField, UniqueField

class User(Model):
    user_id = AutoKeyField()
    email = UniqueField(type=str)       # indexed + unique
    status = IndexedField(type=str)     # indexed, not unique

# Efficient exact-match queries without making email a KeyField
User.query.filter(email="alice@example.com")
User.query.filter(status="active")
User.query.filter(status__in=["active", "pending"])

IndexedFieldMixin

Mixin that provides Set-based secondary indexing for non-key fields.

When mixed with Field, this mixin maintains Redis Sets that track which model instances have a given value for the indexed field. This enables efficient exact-match queries without making the field part of the model's Redis key (identity).

Supports the same query lookups as KeyFieldMixin: - Exact match: filter(field=value) - IN queries: filter(field__in=[v1, v2]) - Null checks: filter(field__isnull=True/False) - Pattern matching: filter(field__startswith="prefix") - Pattern matching: filter(field__endswith="suffix")

Uniqueness enforcement is available via unique=True. The check is performed server-side inside the atomic INDEX_SWAP_LUA script, eliminating the classic check-then-act race condition.

When an external pipeline is provided, a best-effort pre-check is made before queuing the EVAL. The authoritative uniqueness guarantee is only enforced at pipeline.execute() time, when the Lua script runs atomically on the server.

Attributes:

Name	Type	Description
indexed	bool	Always True for indexed fields.

Source code in src/popoto/fields/indexed_field_mixin.py

class IndexedFieldMixin:
    """
    Mixin that provides Set-based secondary indexing for non-key fields.

    When mixed with Field, this mixin maintains Redis Sets that track which
    model instances have a given value for the indexed field. This enables
    efficient exact-match queries without making the field part of the
    model's Redis key (identity).

    Supports the same query lookups as KeyFieldMixin:
    - Exact match: ``filter(field=value)``
    - IN queries: ``filter(field__in=[v1, v2])``
    - Null checks: ``filter(field__isnull=True/False)``
    - Pattern matching: ``filter(field__startswith="prefix")``
    - Pattern matching: ``filter(field__endswith="suffix")``

    Uniqueness enforcement is available via ``unique=True``. The check is
    performed server-side inside the atomic INDEX_SWAP_LUA script, eliminating
    the classic check-then-act race condition.

    When an external pipeline is provided, a best-effort pre-check is made
    before queuing the EVAL. The authoritative uniqueness guarantee is only
    enforced at pipeline.execute() time, when the Lua script runs atomically
    on the server.

    Attributes:
        indexed (bool): Always True for indexed fields.
    """

    indexed: bool = True

    def __init__(self, **kwargs):
        """
        Initialize IndexedFieldMixin defaults.

        Sets ``indexed=True`` and updates field_defaults for introspection.
        Calls super().__init__() for cooperative multiple inheritance.
        """
        super().__init__(**kwargs)
        indexed_defaults = {
            "indexed": True,
        }
        self.field_defaults.update(indexed_defaults)
        for k, v in indexed_defaults.items():
            setattr(self, k, kwargs.get(k, v))

    @classmethod
    def on_save(
        cls,
        model_instance: "Model",
        field_name: str,
        field_value,
        pipeline: redis.client.Pipeline = None,
        **kwargs,
    ):
        """
        Maintain the secondary index Set when a model instance is saved.

        Internal path (no pipeline): queues INDEX_SWAP_LUA into an internal
        MULTI/EXEC pipeline. The Lua script atomically reads the old-set
        pointer from the model hash, removes the member from the old Set,
        enforces uniqueness (if configured), SADD to the new Set, and writes
        the server-authoritative pointer and field bytes — all as a single
        Redis command. No separate round-trips; no race window.

        External path (pipeline provided): performs a best-effort SMEMBERS
        pre-check for fast early raise on uniqueness violation, then queues
        the EVAL into the caller's pipeline. The authoritative guarantee is
        enforced at execute() time when the Lua script runs on the server.
        The pre-check reduces latency for the common conflict case but cannot
        close the TOCTOU window because the caller controls execute().

        Args:
            model_instance: The Model instance being saved.
            field_name: The name of this field on the model.
            field_value: The value being saved for this field.
            pipeline: Optional Redis pipeline for batched operations.
                When provided, EVAL is queued into the caller's pipeline.
                When omitted, an internal pipeline is used for atomicity.

        Returns:
            The pipeline (if provided) or 1 (EVAL result from internal pipeline).

        Raises:
            ModelException: If unique=True and the value is already taken
                by a different instance. Raised immediately on the internal
                path; raised at execute() time on the external path.
        """
        field = model_instance._meta.fields[field_name]
        member_key = model_instance.db_key.redis_key

        # Compute the new index Set key for this field value
        new_set_key = DB_key(
            cls.get_special_use_field_db_key(model_instance, field_name),
            field_value,
        ).redis_key

        # Server-authoritative pointer field name stored in the model hash.
        # The \x00 byte ensures the decode path skips it (never surfaces as
        # a model attribute). See encoding.py decode exclusion.
        ptr_field = f"{field_name}\x00idxset"

        # Legacy-old-set hint: used on first post-upgrade save of records
        # that predate the server-authoritative pointer. The Lua script uses
        # this only when the pointer is absent (i.e., never been written by
        # the new code path).
        saved_values = getattr(model_instance, "_saved_field_values", {})
        old_value = saved_values.get(field_name, _SENTINEL)
        if old_value is not _SENTINEL and old_value != field_value:
            # Use the saved value (even if it was None) as the legacy SREM hint.
            # This handles legacy records (no server-authoritative pointer) where
            # the field was previously stored as None.
            legacy_old_set = DB_key(
                cls.get_special_use_field_db_key(model_instance, field_name),
                old_value,
            ).redis_key
        else:
            legacy_old_set = ""

        # Pack field value with msgpack to match encode_popoto_model_obj() output
        new_bytes = msgpack.packb(field_value)
        is_unique = "1" if getattr(field, "unique", False) else "0"

        if isinstance(pipeline, redis.client.Pipeline):
            # External pipeline path: best-effort pre-check for fast early raise
            if field_value is not None and getattr(field, "unique", False):
                existing_members = POPOTO_REDIS_DB.smembers(new_set_key)
                other_members = {
                    m
                    for m in existing_members
                    if m != member_key.encode() and m != member_key
                }
                if other_members:
                    raise ModelException(
                        f"Uniqueness violation on "
                        f"{model_instance.__class__.__name__}.{field_name}: "
                        f"value '{field_value}' is already taken by another instance"
                    )
            # Queue EVAL into caller's pipeline — authoritative check at execute()
            pipeline.eval(
                INDEX_SWAP_LUA,
                2,
                member_key,  # KEYS[1]: the model hash key (same as record redis_key)
                new_set_key,  # KEYS[2]: new value Set key
                field_name,
                member_key,
                new_bytes,
                is_unique,
                ptr_field,
                legacy_old_set,
            )
            return pipeline
        else:
            # Internal path: execute EVAL via POPOTO_REDIS_DB directly.
            # This runs atomically on the Redis server; no client-side race.
            try:
                result = POPOTO_REDIS_DB.eval(
                    INDEX_SWAP_LUA,
                    2,
                    member_key,  # KEYS[1]: the model hash key (same as record redis_key)
                    new_set_key,  # KEYS[2]: new value Set key
                    field_name,
                    member_key,
                    new_bytes,
                    is_unique,
                    ptr_field,
                    legacy_old_set,
                )
            except Exception as e:
                if "POPOTO_UNIQUE_CONFLICT" in str(e):
                    raise ModelException(
                        f"Uniqueness violation on "
                        f"{model_instance.__class__.__name__}.{field_name}: "
                        f"value '{field_value}' is already taken by another instance"
                    )
                raise
            return result

    @classmethod
    def on_delete(
        cls,
        model_instance: "Model",
        field_name: str,
        field_value,
        pipeline: redis.client.Pipeline = None,
        **kwargs,
    ):
        """
        Remove the model instance from the index Set on delete.

        Uses the server-authoritative {field}\x00idxset pointer (if present) to
        determine which value-Set to SREM from. Falls back to field_value-derived
        key for legacy records without the pointer.

        Args:
            model_instance: The Model instance being deleted.
            field_name: The name of this field on the model.
            field_value: The value stored for this field.
            pipeline: Optional Redis pipeline for batched operations.

        Returns:
            The pipeline (if provided) or the SREM result.
        """
        member_key = kwargs.get("saved_redis_key", model_instance.db_key.redis_key)
        model_hash_key = member_key  # model hash key = redis_key for this record

        # Read server-authoritative pointer to find the exact set this member is in
        ptr_field = f"{field_name}\x00idxset"
        ptr_value = POPOTO_REDIS_DB.hget(model_hash_key, ptr_field)

        if ptr_value:
            # Pointer present: SREM from the set it names
            index_set_key = (
                ptr_value.decode() if isinstance(ptr_value, bytes) else ptr_value
            )
        else:
            # Legacy record (no pointer): fall back to field_value-derived key
            index_set_key = DB_key(
                cls.get_special_use_field_db_key(model_instance, field_name),
                field_value,
            ).redis_key

        if pipeline:
            return pipeline.srem(index_set_key, member_key)
        else:
            return POPOTO_REDIS_DB.srem(index_set_key, member_key)

    def get_filter_query_params(self, field_name: str) -> set:
        """
        Return the set of valid query parameter names for filtering on this field.

        Supports the same lookups as KeyFieldMixin:
        - exact match
        - __in
        - __isnull
        - __startswith
        - __endswith

        Args:
            field_name: The name of this field on the model.

        Returns:
            set: Valid query parameter strings.
        """
        return (
            super()
            .get_filter_query_params(field_name)
            .union(
                {
                    f"{field_name}",
                    f"{field_name}__isnull",
                    f"{field_name}__startswith",
                    f"{field_name}__endswith",
                    f"{field_name}__in",
                }
            )
        )

    @classmethod
    def filter_query(cls, model: "Model", field_name: str, **query_params) -> set:
        """
        Execute a filter query on this indexed field and return matching Redis keys.

        Uses Set-based lookups for exact match and __in queries. Falls back
        to SCAN for pattern queries (__startswith, __endswith).

        Args:
            model: The Model class being queried.
            field_name: The name of this field on the model.
            **query_params: Dict mapping query parameter names to values.

        Returns:
            set: Redis keys (as bytes) of matching model instances.

        Raises:
            QueryException: If __isnull receives a non-boolean value.
        """
        keys_lists_to_intersect = list()

        redis_set_key_prefix = model._meta.fields[
            field_name
        ].get_special_use_field_db_key(model, field_name)

        for query_param, query_value in query_params.items():
            if query_param.endswith("__in"):
                # Use SUNION for efficient server-side set union
                set_keys = [
                    DB_key(redis_set_key_prefix, query_value_elem).redis_key
                    for query_value_elem in query_value
                ]
                if set_keys:
                    keys_lists_to_intersect.append(POPOTO_REDIS_DB.sunion(set_keys))
                else:
                    keys_lists_to_intersect.append(set())

            elif query_param == f"{field_name}":
                keys_lists_to_intersect.append(
                    POPOTO_REDIS_DB.smembers(
                        DB_key(redis_set_key_prefix, query_value).redis_key
                    )
                )

            elif query_param.endswith("__isnull"):
                if query_value is True:
                    keys_lists_to_intersect.append(
                        POPOTO_REDIS_DB.smembers(
                            DB_key(redis_set_key_prefix, None).redis_key
                        )
                    )
                elif query_value is False:
                    # Use SCAN to find all index keys for this field, then union
                    pattern = f"{redis_set_key_prefix.redis_key}:*"
                    all_keys = scan_keys(pattern)
                    # Filter out the None set
                    none_key = DB_key(redis_set_key_prefix, None).redis_key
                    non_null_keys = set()
                    for key in all_keys:
                        key_str = key.decode() if isinstance(key, bytes) else key
                        if key_str != none_key:
                            non_null_keys.update(POPOTO_REDIS_DB.smembers(key_str))
                    keys_lists_to_intersect.append(non_null_keys)
                else:
                    raise QueryException(f"{query_param} filter must be True or False")

            elif query_param.endswith("__startswith"):
                # Scan index keys matching the prefix pattern
                pattern = (
                    f"{redis_set_key_prefix.redis_key}:{DB_key.clean(query_value)}*"
                )
                matching_index_keys = scan_keys(pattern)
                members = set()
                for idx_key in matching_index_keys:
                    members.update(POPOTO_REDIS_DB.smembers(idx_key))
                keys_lists_to_intersect.append(members)

            elif query_param.endswith("__endswith"):
                # Scan index keys matching the suffix pattern
                pattern = (
                    f"{redis_set_key_prefix.redis_key}:*{DB_key.clean(query_value)}"
                )
                matching_index_keys = scan_keys(pattern)
                members = set()
                for idx_key in matching_index_keys:
                    members.update(POPOTO_REDIS_DB.smembers(idx_key))
                keys_lists_to_intersect.append(members)

        logger.debug(keys_lists_to_intersect)
        if len(keys_lists_to_intersect):
            return set.intersection(
                *[set(key_list) for key_list in keys_lists_to_intersect]
            )
        return set()

on_save(model_instance, field_name, field_value, pipeline=None, **kwargs) classmethod

Maintain the secondary index Set when a model instance is saved.

Internal path (no pipeline): queues INDEX_SWAP_LUA into an internal MULTI/EXEC pipeline. The Lua script atomically reads the old-set pointer from the model hash, removes the member from the old Set, enforces uniqueness (if configured), SADD to the new Set, and writes the server-authoritative pointer and field bytes — all as a single Redis command. No separate round-trips; no race window.

External path (pipeline provided): performs a best-effort SMEMBERS pre-check for fast early raise on uniqueness violation, then queues the EVAL into the caller's pipeline. The authoritative guarantee is enforced at execute() time when the Lua script runs on the server. The pre-check reduces latency for the common conflict case but cannot close the TOCTOU window because the caller controls execute().

Parameters:

Name	Type	Description	Default
model_instance	Model	The Model instance being saved.	required
field_name	str	The name of this field on the model.	required
field_value		The value being saved for this field.	required
pipeline	Pipeline	Optional Redis pipeline for batched operations. When provided, EVAL is queued into the caller's pipeline. When omitted, an internal pipeline is used for atomicity.	None

Returns:

Type	Description
	The pipeline (if provided) or 1 (EVAL result from internal pipeline).

Raises:

Type	Description
ModelException	If unique=True and the value is already taken by a different instance. Raised immediately on the internal path; raised at execute() time on the external path.

Source code in src/popoto/fields/indexed_field_mixin.py

@classmethod
def on_save(
    cls,
    model_instance: "Model",
    field_name: str,
    field_value,
    pipeline: redis.client.Pipeline = None,
    **kwargs,
):
    """
    Maintain the secondary index Set when a model instance is saved.

    Internal path (no pipeline): queues INDEX_SWAP_LUA into an internal
    MULTI/EXEC pipeline. The Lua script atomically reads the old-set
    pointer from the model hash, removes the member from the old Set,
    enforces uniqueness (if configured), SADD to the new Set, and writes
    the server-authoritative pointer and field bytes — all as a single
    Redis command. No separate round-trips; no race window.

    External path (pipeline provided): performs a best-effort SMEMBERS
    pre-check for fast early raise on uniqueness violation, then queues
    the EVAL into the caller's pipeline. The authoritative guarantee is
    enforced at execute() time when the Lua script runs on the server.
    The pre-check reduces latency for the common conflict case but cannot
    close the TOCTOU window because the caller controls execute().

    Args:
        model_instance: The Model instance being saved.
        field_name: The name of this field on the model.
        field_value: The value being saved for this field.
        pipeline: Optional Redis pipeline for batched operations.
            When provided, EVAL is queued into the caller's pipeline.
            When omitted, an internal pipeline is used for atomicity.

    Returns:
        The pipeline (if provided) or 1 (EVAL result from internal pipeline).

    Raises:
        ModelException: If unique=True and the value is already taken
            by a different instance. Raised immediately on the internal
            path; raised at execute() time on the external path.
    """
    field = model_instance._meta.fields[field_name]
    member_key = model_instance.db_key.redis_key

    # Compute the new index Set key for this field value
    new_set_key = DB_key(
        cls.get_special_use_field_db_key(model_instance, field_name),
        field_value,
    ).redis_key

    # Server-authoritative pointer field name stored in the model hash.
    # The \x00 byte ensures the decode path skips it (never surfaces as
    # a model attribute). See encoding.py decode exclusion.
    ptr_field = f"{field_name}\x00idxset"

    # Legacy-old-set hint: used on first post-upgrade save of records
    # that predate the server-authoritative pointer. The Lua script uses
    # this only when the pointer is absent (i.e., never been written by
    # the new code path).
    saved_values = getattr(model_instance, "_saved_field_values", {})
    old_value = saved_values.get(field_name, _SENTINEL)
    if old_value is not _SENTINEL and old_value != field_value:
        # Use the saved value (even if it was None) as the legacy SREM hint.
        # This handles legacy records (no server-authoritative pointer) where
        # the field was previously stored as None.
        legacy_old_set = DB_key(
            cls.get_special_use_field_db_key(model_instance, field_name),
            old_value,
        ).redis_key
    else:
        legacy_old_set = ""

    # Pack field value with msgpack to match encode_popoto_model_obj() output
    new_bytes = msgpack.packb(field_value)
    is_unique = "1" if getattr(field, "unique", False) else "0"

    if isinstance(pipeline, redis.client.Pipeline):
        # External pipeline path: best-effort pre-check for fast early raise
        if field_value is not None and getattr(field, "unique", False):
            existing_members = POPOTO_REDIS_DB.smembers(new_set_key)
            other_members = {
                m
                for m in existing_members
                if m != member_key.encode() and m != member_key
            }
            if other_members:
                raise ModelException(
                    f"Uniqueness violation on "
                    f"{model_instance.__class__.__name__}.{field_name}: "
                    f"value '{field_value}' is already taken by another instance"
                )
        # Queue EVAL into caller's pipeline — authoritative check at execute()
        pipeline.eval(
            INDEX_SWAP_LUA,
            2,
            member_key,  # KEYS[1]: the model hash key (same as record redis_key)
            new_set_key,  # KEYS[2]: new value Set key
            field_name,
            member_key,
            new_bytes,
            is_unique,
            ptr_field,
            legacy_old_set,
        )
        return pipeline
    else:
        # Internal path: execute EVAL via POPOTO_REDIS_DB directly.
        # This runs atomically on the Redis server; no client-side race.
        try:
            result = POPOTO_REDIS_DB.eval(
                INDEX_SWAP_LUA,
                2,
                member_key,  # KEYS[1]: the model hash key (same as record redis_key)
                new_set_key,  # KEYS[2]: new value Set key
                field_name,
                member_key,
                new_bytes,
                is_unique,
                ptr_field,
                legacy_old_set,
            )
        except Exception as e:
            if "POPOTO_UNIQUE_CONFLICT" in str(e):
                raise ModelException(
                    f"Uniqueness violation on "
                    f"{model_instance.__class__.__name__}.{field_name}: "
                    f"value '{field_value}' is already taken by another instance"
                )
            raise
        return result

on_delete(model_instance, field_name, field_value, pipeline=None, **kwargs) classmethod

Remove the model instance from the index Set on delete.

Uses the server-authoritative {field}idxset pointer (if present) to determine which value-Set to SREM from. Falls back to field_value-derived key for legacy records without the pointer.

Parameters:

Name	Type	Description	Default
model_instance	Model	The Model instance being deleted.	required
field_name	str	The name of this field on the model.	required
field_value		The value stored for this field.	required
pipeline	Pipeline	Optional Redis pipeline for batched operations.	None

Returns:

Type	Description
	The pipeline (if provided) or the SREM result.

Source code in src/popoto/fields/indexed_field_mixin.py

@classmethod
def on_delete(
    cls,
    model_instance: "Model",
    field_name: str,
    field_value,
    pipeline: redis.client.Pipeline = None,
    **kwargs,
):
    """
    Remove the model instance from the index Set on delete.

    Uses the server-authoritative {field}\x00idxset pointer (if present) to
    determine which value-Set to SREM from. Falls back to field_value-derived
    key for legacy records without the pointer.

    Args:
        model_instance: The Model instance being deleted.
        field_name: The name of this field on the model.
        field_value: The value stored for this field.
        pipeline: Optional Redis pipeline for batched operations.

    Returns:
        The pipeline (if provided) or the SREM result.
    """
    member_key = kwargs.get("saved_redis_key", model_instance.db_key.redis_key)
    model_hash_key = member_key  # model hash key = redis_key for this record

    # Read server-authoritative pointer to find the exact set this member is in
    ptr_field = f"{field_name}\x00idxset"
    ptr_value = POPOTO_REDIS_DB.hget(model_hash_key, ptr_field)

    if ptr_value:
        # Pointer present: SREM from the set it names
        index_set_key = (
            ptr_value.decode() if isinstance(ptr_value, bytes) else ptr_value
        )
    else:
        # Legacy record (no pointer): fall back to field_value-derived key
        index_set_key = DB_key(
            cls.get_special_use_field_db_key(model_instance, field_name),
            field_value,
        ).redis_key

    if pipeline:
        return pipeline.srem(index_set_key, member_key)
    else:
        return POPOTO_REDIS_DB.srem(index_set_key, member_key)

get_filter_query_params(field_name)

Return the set of valid query parameter names for filtering on this field.

Supports the same lookups as KeyFieldMixin: - exact match - __in - __isnull - __startswith - __endswith

Parameters:

Name	Type	Description	Default
field_name	str	The name of this field on the model.	required

Returns:

Name	Type	Description
set	set	Valid query parameter strings.

Source code in src/popoto/fields/indexed_field_mixin.py

def get_filter_query_params(self, field_name: str) -> set:
    """
    Return the set of valid query parameter names for filtering on this field.

    Supports the same lookups as KeyFieldMixin:
    - exact match
    - __in
    - __isnull
    - __startswith
    - __endswith

    Args:
        field_name: The name of this field on the model.

    Returns:
        set: Valid query parameter strings.
    """
    return (
        super()
        .get_filter_query_params(field_name)
        .union(
            {
                f"{field_name}",
                f"{field_name}__isnull",
                f"{field_name}__startswith",
                f"{field_name}__endswith",
                f"{field_name}__in",
            }
        )
    )

filter_query(model, field_name, **query_params) classmethod

Execute a filter query on this indexed field and return matching Redis keys.

Uses Set-based lookups for exact match and __in queries. Falls back to SCAN for pattern queries (__startswith, __endswith).

Parameters:

Name	Type	Description	Default
model	Model	The Model class being queried.	required
field_name	str	The name of this field on the model.	required
**query_params		Dict mapping query parameter names to values.	{}

Returns:

Name	Type	Description
set	set	Redis keys (as bytes) of matching model instances.

Raises:

Type	Description
QueryException	If __isnull receives a non-boolean value.

Source code in src/popoto/fields/indexed_field_mixin.py

@classmethod
def filter_query(cls, model: "Model", field_name: str, **query_params) -> set:
    """
    Execute a filter query on this indexed field and return matching Redis keys.

    Uses Set-based lookups for exact match and __in queries. Falls back
    to SCAN for pattern queries (__startswith, __endswith).

    Args:
        model: The Model class being queried.
        field_name: The name of this field on the model.
        **query_params: Dict mapping query parameter names to values.

    Returns:
        set: Redis keys (as bytes) of matching model instances.

    Raises:
        QueryException: If __isnull receives a non-boolean value.
    """
    keys_lists_to_intersect = list()

    redis_set_key_prefix = model._meta.fields[
        field_name
    ].get_special_use_field_db_key(model, field_name)

    for query_param, query_value in query_params.items():
        if query_param.endswith("__in"):
            # Use SUNION for efficient server-side set union
            set_keys = [
                DB_key(redis_set_key_prefix, query_value_elem).redis_key
                for query_value_elem in query_value
            ]
            if set_keys:
                keys_lists_to_intersect.append(POPOTO_REDIS_DB.sunion(set_keys))
            else:
                keys_lists_to_intersect.append(set())

        elif query_param == f"{field_name}":
            keys_lists_to_intersect.append(
                POPOTO_REDIS_DB.smembers(
                    DB_key(redis_set_key_prefix, query_value).redis_key
                )
            )

        elif query_param.endswith("__isnull"):
            if query_value is True:
                keys_lists_to_intersect.append(
                    POPOTO_REDIS_DB.smembers(
                        DB_key(redis_set_key_prefix, None).redis_key
                    )
                )
            elif query_value is False:
                # Use SCAN to find all index keys for this field, then union
                pattern = f"{redis_set_key_prefix.redis_key}:*"
                all_keys = scan_keys(pattern)
                # Filter out the None set
                none_key = DB_key(redis_set_key_prefix, None).redis_key
                non_null_keys = set()
                for key in all_keys:
                    key_str = key.decode() if isinstance(key, bytes) else key
                    if key_str != none_key:
                        non_null_keys.update(POPOTO_REDIS_DB.smembers(key_str))
                keys_lists_to_intersect.append(non_null_keys)
            else:
                raise QueryException(f"{query_param} filter must be True or False")

        elif query_param.endswith("__startswith"):
            # Scan index keys matching the prefix pattern
            pattern = (
                f"{redis_set_key_prefix.redis_key}:{DB_key.clean(query_value)}*"
            )
            matching_index_keys = scan_keys(pattern)
            members = set()
            for idx_key in matching_index_keys:
                members.update(POPOTO_REDIS_DB.smembers(idx_key))
            keys_lists_to_intersect.append(members)

        elif query_param.endswith("__endswith"):
            # Scan index keys matching the suffix pattern
            pattern = (
                f"{redis_set_key_prefix.redis_key}:*{DB_key.clean(query_value)}"
            )
            matching_index_keys = scan_keys(pattern)
            members = set()
            for idx_key in matching_index_keys:
                members.update(POPOTO_REDIS_DB.smembers(idx_key))
            keys_lists_to_intersect.append(members)

    logger.debug(keys_lists_to_intersect)
    if len(keys_lists_to_intersect):
        return set.intersection(
            *[set(key_list) for key_list in keys_lists_to_intersect]
        )
    return set()