MCPcopy
hub / github.com/sqlalchemy/sqlalchemy / WriteableColumnCollection

Class WriteableColumnCollection

lib/sqlalchemy/sql/base.py:2067–2297  ·  view source on GitHub ↗

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

Source from the content-addressed store, hash-verified

2065
2066
2067class 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)

Callers 8

_setup_collectionsMethod · 0.85
selected_columnsMethod · 0.85
selected_columnsMethod · 0.85
exported_columnsMethod · 0.85
__getitem__Method · 0.85
cMethod · 0.85
__init__Method · 0.85

Calls

no outgoing calls

Tested by

no test coverage detected