Class DedupeColumnCollection

lib/sqlalchemy/sql/base.py:2027–2199

A :class:`_expression.ColumnCollection` that maintains deduplicating behavior. This is useful by schema level objects such as :class:`_schema.Table` and :class:`.PrimaryKeyConstraint`. The collection includes more sophisticated mutator methods as well to suit schema objects which

class DedupeColumnCollection(ColumnCollection[str, _NAMEDCOL]):
    """A :class:`_expression.ColumnCollection`
    that maintains deduplicating behavior.

    This is useful by schema level objects such as :class:`_schema.Table` and
    :class:`.PrimaryKeyConstraint`.    The collection includes more
    sophisticated mutator methods as well to suit schema objects which
    require mutable column collections.

    .. versionadded:: 1.4

    """

    def add(  # type: ignore[override]
        self, column: _NAMEDCOL, key: Optional[str] = None
    ) -> None:
        if key is not None and column.key != key:
            raise exc.ArgumentError(
                "DedupeColumnCollection requires columns be under "
                "the same key as their .key"
            )
        key = column.key

        if key is None:
            raise exc.ArgumentError(
                "Can't add unnamed column to column collection"
            )

        if key in self._index:
            existing = self._index[key][1]

            if existing is column:
                return

            self.replace(column)

            # pop out memoized proxy_set as this
            # operation may very well be occurring
            # in a _make_proxy operation
            util.memoized_property.reset(column, "proxy_set")
        else:
            self._append_new_column(key, column)

    def _append_new_column(self, key: str, named_column: _NAMEDCOL) -> None:
        l = len(self._collection)
        self._collection.append(
            (key, named_column, _ColumnMetrics(self, named_column))
        )
        self._colset.add(named_column._deannotate())
        self._index[l] = (key, named_column)
        self._index[key] = (key, named_column)

    def _populate_separate_keys(
        self, iter_: Iterable[Tuple[str, _NAMEDCOL]]
    ) -> None:
        """populate from an iterator of (key, column)"""
        cols = list(iter_)