Method replace

lib/sqlalchemy/sql/base.py:2433–2521

add the given column to this collection, removing unaliased versions of this column as well as existing columns with the same key. e.g.:: t = Table("sometable", metadata, Column("col1", Integer)) t.columns.replace(Column("col1", Integer, key="column

(
        self,
        column: _NAMEDCOL,
        *,
        extra_remove: Optional[Iterable[_NAMEDCOL]] = None,
        index: Optional[int] = None,
    )

        del self._index[len(self._collection)]

    def replace(
        self,
        column: _NAMEDCOL,
        *,
        extra_remove: Optional[Iterable[_NAMEDCOL]] = None,
        index: Optional[int] = None,
    ) -> None:
        """add the given column to this collection, removing unaliased
        versions of this column  as well as existing columns with the
        same key.

        e.g.::

            t = Table("sometable", metadata, Column("col1", Integer))
            t.columns.replace(Column("col1", Integer, key="columnone"))

        will remove the original 'col1' from the collection, and add
        the new column under the name 'columnname'.

        Used by schema.Column to override columns during table reflection.

        """

        if extra_remove:
            remove_col = set(extra_remove)
        else:
            remove_col = set()
        # remove up to two columns based on matches of name as well as key
        if column.name in self._index and column.key != column.name:
            other = self._index[column.name][1]
            if other.name == other.key:
                remove_col.add(other)

        if column.key in self._index:
            remove_col.add(self._index[column.key][1])

        if not remove_col:
            self._append_new_column(column.key, column, index=index)
            return
        new_cols: List[Tuple[str, _NAMEDCOL, _ColumnMetrics[_NAMEDCOL]]] = []
        replace_index = None

        for idx, (k, col, metrics) in enumerate(self._collection):
            if col in remove_col:
                if replace_index is None:
                    replace_index = idx
                    new_cols.append(
                        (column.key, column, _ColumnMetrics(self, column))
                    )
            else:
                new_cols.append((k, col, metrics))

        if remove_col:
            self._colset.difference_update(remove_col)

            for rc in remove_col:
                for metrics in self._proxy_index.get(rc, ()):
                    metrics.dispose(self)