collections.dedupe-by-key

from typing import Any, List, Mapping, Optional, Sequence, Set

from .collections_dedupe_by_key_types import DedupeKeep

# Largest integer JavaScript can hold exactly; beyond it the three languages disagree.
SAFE_INTEGER = 9007199254740991


def _key_of(value: Any, key: str) -> Optional[str]:
    """The identity of a record, or None when it has none.

    Rendered exactly as collections.group-by-key renders a group name, so the
    two capabilities agree on what "the same key" means. Each language's
    default string conversion differs (Python prints True, JavaScript true),
    so the rendering is spelled out rather than inherited.
    """
    # A record with no key is not a duplicate of anything: two records that
    # both lack an id are two unknowns, not one thing seen twice.
    if value is None:
        return None
    if isinstance(value, str):
        return value
    # bool before int: in Python True is an int.
    if isinstance(value, bool):
        return "true" if value else "false"
    if isinstance(value, int):
        if abs(value) > SAFE_INTEGER:
            raise ValueError('cannot dedupe by the out-of-range number %d at "%s"' % (value, key))
        return str(value)
    if isinstance(value, float):
        raise TypeError('cannot dedupe by the fractional number %r at "%s"' % (value, key))
    raise TypeError('cannot dedupe by the list or map at "%s"' % (key,))


def dedupe_by_key(
    records: Sequence[Mapping[str, Any]],
    key: str,
    keep: DedupeKeep,
) -> List[Mapping[str, Any]]:
    """Remove records that share the value at ``key``, keeping the first or
    the last of each set.

    Survivors keep their own positions: with "last", the survivor sits where
    the last occurrence was, which is what "latest wins" means in a change feed.
    """
    if isinstance(records, (str, bytes)) or not isinstance(records, (list, tuple)):
        raise TypeError("dedupe_by_key needs a list of records")
    if not isinstance(key, str) or key == "":
        raise TypeError("dedupe_by_key needs a non-empty key name")
    if keep not in ("first", "last"):
        raise ValueError('keep must be "first" or "last", received "%s"' % (keep,))

    # Every key is rendered up front, so a bad value raises whichever mode runs.
    keys = [_key_of(record.get(key) if isinstance(record, dict) else None, key) for record in records]

    seen: Set[str] = set()
    survives = [False] * len(records)
    # Walking backwards for "last" makes the last occurrence the first one seen.
    order = range(len(records) - 1, -1, -1) if keep == "last" else range(len(records))
    for i in order:
        k = keys[i]
        if k is None:
            survives[i] = True
        elif k not in seen:
            seen.add(k)
            survives[i] = True
    return [record for i, record in enumerate(records) if survives[i]]