A :class:`_sql.ColumnCollection` that allows mutation operations. This is the writable form of :class:`_sql.ColumnCollection` that implements methods such as :meth:`.add`, :meth:`.remove`, :meth:`.update`, and :meth:`.clear`. This class is used internally for building column collec
| 2065 | |
| 2066 | |
| 2067 | class WriteableColumnCollection(ColumnCollection[_COLKEY, _COL_co]): |
| 2068 | """A :class:`_sql.ColumnCollection` that allows mutation operations. |
| 2069 | |
| 2070 | This is the writable form of :class:`_sql.ColumnCollection` that |
| 2071 | implements methods such as :meth:`.add`, :meth:`.remove`, :meth:`.update`, |
| 2072 | and :meth:`.clear`. |
| 2073 | |
| 2074 | This class is used internally for building column collections during |
| 2075 | construction of SQL constructs. For schema-level objects that require |
| 2076 | deduplication behavior, use :class:`.DedupeColumnCollection`. |
| 2077 | |
| 2078 | .. versionadded:: 2.1 |
| 2079 | |
| 2080 | """ |
| 2081 | |
| 2082 | __slots__ = () |
| 2083 | |
| 2084 | def __init__( |
| 2085 | self, columns: Optional[Iterable[Tuple[_COLKEY, _COL_co]]] = None |
| 2086 | ): |
| 2087 | object.__setattr__(self, "_colset", set()) |
| 2088 | object.__setattr__(self, "_index", {}) |
| 2089 | object.__setattr__( |
| 2090 | self, "_proxy_index", collections.defaultdict(util.OrderedSet) |
| 2091 | ) |
| 2092 | object.__setattr__(self, "_collection", []) |
| 2093 | if columns: |
| 2094 | self._initial_populate(columns) |
| 2095 | |
| 2096 | def _initial_populate( |
| 2097 | self, iter_: Iterable[Tuple[_COLKEY, _COL_co]] |
| 2098 | ) -> None: |
| 2099 | self._populate_separate_keys(iter_) |
| 2100 | |
| 2101 | def _populate_separate_keys( |
| 2102 | self, iter_: Iterable[Tuple[_COLKEY, _COL_co]] |
| 2103 | ) -> None: |
| 2104 | """populate from an iterator of (key, column)""" |
| 2105 | |
| 2106 | self._collection[:] = collection = [ |
| 2107 | (k, c, _ColumnMetrics(self, c)) for k, c in iter_ |
| 2108 | ] |
| 2109 | self._colset.update(c._deannotate() for _, c, _ in collection) |
| 2110 | self._index.update( |
| 2111 | {idx: (k, c) for idx, (k, c, _) in enumerate(collection)} |
| 2112 | ) |
| 2113 | self._index.update({k: (k, col) for k, col, _ in reversed(collection)}) |
| 2114 | |
| 2115 | def __getstate__(self) -> Dict[str, Any]: |
| 2116 | return { |
| 2117 | "_collection": [(k, c) for k, c, _ in self._collection], |
| 2118 | "_index": self._index, |
| 2119 | } |
| 2120 | |
| 2121 | def __setstate__(self, state: Dict[str, Any]) -> None: |
| 2122 | object.__setattr__(self, "_index", state["_index"]) |
| 2123 | object.__setattr__( |
| 2124 | self, "_proxy_index", collections.defaultdict(util.OrderedSet) |
no outgoing calls
no test coverage detected