django_cte_force_hashjoin.py

from django_cte.raw import raw_cte_sql

from django_cte import CTEQuerySet, With

class CustomQuerySet(CTEQuerySet):
    def filter_by_uuids(self, uuids: Sequence):
        valid_uuids = [value for value in uuids if is_valid_uuid(value)]
        placeholders = ",".join(["%s"] * len(valid_uuids))
        uuid_table = With(
            raw_cte_sql(
                f"SELECT uuid_field FROM UNNEST(ARRAY[{placeholders}]::uuid[]) AS uuid_field",
                valid_uuids,
                {"uuid_field": models.UUIDField()},
            ),
            name="uuids_table",
        )
        return uuid_table.join(
            self, uuid=uuid_table.col.uuid_field, _join_type=INNER
        ).with_cte(uuid_table)