codekingpro/portable-devtools

1# ext/orderinglist.py
2# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: https://www.opensource.org/licenses/mit-license.php
7
8"""A custom list that manages index/position information for contained
9elements.
10
11:author: Jason Kirtland
12
13``orderinglist`` is a helper for mutable ordered relationships.  It will
14intercept list operations performed on a :func:`_orm.relationship`-managed
15collection and
16automatically synchronize changes in list position onto a target scalar
17attribute.
18
19Example: A ``slide`` table, where each row refers to zero or more entries
20in a related ``bullet`` table.   The bullets within a slide are
21displayed in order based on the value of the ``position`` column in the
22``bullet`` table.   As entries are reordered in memory, the value of the
23``position`` attribute should be updated to reflect the new sort order::
24
25
26    Base = declarative_base()
27
28
29    class Slide(Base):
30        __tablename__ = "slide"
31
32        id = Column(Integer, primary_key=True)
33        name = Column(String)
34
35        bullets = relationship("Bullet", order_by="Bullet.position")
36
37
38    class Bullet(Base):
39        __tablename__ = "bullet"
40        id = Column(Integer, primary_key=True)
41        slide_id = Column(Integer, ForeignKey("slide.id"))
42        position = Column(Integer)
43        text = Column(String)
44
45The standard relationship mapping will produce a list-like attribute on each
46``Slide`` containing all related ``Bullet`` objects,
47but coping with changes in ordering is not handled automatically.
48When appending a ``Bullet`` into ``Slide.bullets``, the ``Bullet.position``
49attribute will remain unset until manually assigned.   When the ``Bullet``
50is inserted into the middle of the list, the following ``Bullet`` objects
51will also need to be renumbered.
52
53The :class:`.OrderingList` object automates this task, managing the
54``position`` attribute on all ``Bullet`` objects in the collection.  It is
55constructed using the :func:`.ordering_list` factory::
56
57    from sqlalchemy.ext.orderinglist import ordering_list
58
59    Base = declarative_base()
60
61
62    class Slide(Base):
63        __tablename__ = "slide"
64
65        id = Column(Integer, primary_key=True)
66        name = Column(String)
67
68        bullets = relationship(
69            "Bullet",
70            order_by="Bullet.position",
71            collection_class=ordering_list("position"),
72        )
73
74
75    class Bullet(Base):
76        __tablename__ = "bullet"
77        id = Column(Integer, primary_key=True)
78        slide_id = Column(Integer, ForeignKey("slide.id"))
79        position = Column(Integer)
80        text = Column(String)
81
82With the above mapping the ``Bullet.position`` attribute is managed::
83
84    s = Slide()
85    s.bullets.append(Bullet())
86    s.bullets.append(Bullet())
87    s.bullets[1].position
88    >>> 1
89    s.bullets.insert(1, Bullet())
90    s.bullets[2].position
91    >>> 2
92
93The :class:`.OrderingList` construct only works with **changes** to a
94collection, and not the initial load from the database, and requires that the
95list be sorted when loaded.  Therefore, be sure to specify ``order_by`` on the
96:func:`_orm.relationship` against the target ordering attribute, so that the
97ordering is correct when first loaded.
98
99.. warning::
100
101  :class:`.OrderingList` only provides limited functionality when a primary
102  key column or unique column is the target of the sort.  Operations
103  that are unsupported or are problematic include:
104
105    * two entries must trade values.  This is not supported directly in the
106      case of a primary key or unique constraint because it means at least
107      one row would need to be temporarily removed first, or changed to
108      a third, neutral value while the switch occurs.
109
110    * an entry must be deleted in order to make room for a new entry.
111      SQLAlchemy's unit of work performs all INSERTs before DELETEs within a
112      single flush.  In the case of a primary key, it will trade
113      an INSERT/DELETE of the same primary key for an UPDATE statement in order
114      to lessen the impact of this limitation, however this does not take place
115      for a UNIQUE column.
116      A future feature will allow the "DELETE before INSERT" behavior to be
117      possible, alleviating this limitation, though this feature will require
118      explicit configuration at the mapper level for sets of columns that
119      are to be handled in this way.
120
121:func:`.ordering_list` takes the name of the related object's ordering
122attribute as an argument.  By default, the zero-based integer index of the
123object's position in the :func:`.ordering_list` is synchronized with the
124ordering attribute: index 0 will get position 0, index 1 position 1, etc.  To
125start numbering at 1 or some other integer, provide ``count_from=1``.
126
127
128"""
129from __future__ import annotations
130
131from typing import Any
132from typing import Callable
133from typing import Dict
134from typing import Iterable
135from typing import List
136from typing import Optional
137from typing import overload
138from typing import Sequence
139from typing import Type
140from typing import TypeVar
141from typing import Union
142
143from ..orm.collections import collection
144from ..orm.collections import collection_adapter
145from ..util.typing import SupportsIndex
146
147_T = TypeVar("_T")
148OrderingFunc = Callable[[int, Sequence[_T]], object]
149
150
151__all__ = ["ordering_list"]
152
153
154def ordering_list(
155    attr: str,
156    count_from: Optional[int] = None,
157    ordering_func: Optional[OrderingFunc[_T]] = None,
158    reorder_on_append: bool = False,
159) -> Callable[[], OrderingList[_T]]:
160    """Prepares an :class:`OrderingList` factory for use in mapper definitions.
161
162    Returns an object suitable for use as an argument to a Mapper
163    relationship's ``collection_class`` option.  e.g.::
164
165        from sqlalchemy.ext.orderinglist import ordering_list
166
167
168        class Slide(Base):
169            __tablename__ = "slide"
170
171            id = Column(Integer, primary_key=True)
172            name = Column(String)
173
174            bullets = relationship(
175                "Bullet",
176                order_by="Bullet.position",
177                collection_class=ordering_list("position"),
178            )
179
180    :param attr:
181      Name of the mapped attribute to use for storage and retrieval of
182      ordering information
183
184    :param count_from:
185      Set up an integer-based ordering, starting at ``count_from``.  For
186      example, ``ordering_list('pos', count_from=1)`` would create a 1-based
187      list in SQL, storing the value in the 'pos' column.  Ignored if
188      ``ordering_func`` is supplied.
189
190    Additional arguments are passed to the :class:`.OrderingList` constructor.
191
192    """
193
194    kw = _unsugar_count_from(
195        count_from=count_from,
196        ordering_func=ordering_func,
197        reorder_on_append=reorder_on_append,
198    )
199    return lambda: OrderingList(attr, **kw)
200
201
202# Ordering utility functions
203
204
205def count_from_0(index: int, collection: object) -> int:
206    """Numbering function: consecutive integers starting at 0."""
207
208    return index
209
210
211def count_from_1(index: int, collection: object) -> int:
212    """Numbering function: consecutive integers starting at 1."""
213
214    return index + 1
215
216
217def count_from_n_factory(start: int) -> OrderingFunc[Any]:
218    """Numbering function: consecutive integers starting at arbitrary start."""
219
220    def f(index: int, collection: object) -> int:
221        return index + start
222
223    try:
224        f.__name__ = "count_from_%i" % start
225    except TypeError:
226        pass
227    return f
228
229
230def _unsugar_count_from(**kw: Any) -> Dict[str, Any]:
231    """Builds counting functions from keyword arguments.
232
233    Keyword argument filter, prepares a simple ``ordering_func`` from a
234    ``count_from`` argument, otherwise passes ``ordering_func`` on unchanged.
235    """
236
237    count_from = kw.pop("count_from", None)
238    if kw.get("ordering_func", None) is None and count_from is not None:
239        if count_from == 0:
240            kw["ordering_func"] = count_from_0
241        elif count_from == 1:
242            kw["ordering_func"] = count_from_1
243        else:
244            kw["ordering_func"] = count_from_n_factory(count_from)
245    return kw
246
247
248class OrderingList(List[_T]):
249    """A custom list that manages position information for its children.
250
251    The :class:`.OrderingList` object is normally set up using the
252    :func:`.ordering_list` factory function, used in conjunction with
253    the :func:`_orm.relationship` function.
254
255    """
256
257    ordering_attr: str
258    ordering_func: OrderingFunc[_T]
259    reorder_on_append: bool
260
261    def __init__(
262        self,
263        ordering_attr: str,
264        ordering_func: Optional[OrderingFunc[_T]] = None,
265        reorder_on_append: bool = False,
266    ):
267        """A custom list that manages position information for its children.
268
269        ``OrderingList`` is a ``collection_class`` list implementation that
270        syncs position in a Python list with a position attribute on the
271        mapped objects.
272
273        This implementation relies on the list starting in the proper order,
274        so be **sure** to put an ``order_by`` on your relationship.
275
276        :param ordering_attr:
277          Name of the attribute that stores the object's order in the
278          relationship.
279
280        :param ordering_func: Optional.  A function that maps the position in
281          the Python list to a value to store in the
282          ``ordering_attr``.  Values returned are usually (but need not be!)
283          integers.
284
285          An ``ordering_func`` is called with two positional parameters: the
286          index of the element in the list, and the list itself.
287
288          If omitted, Python list indexes are used for the attribute values.
289          Two basic pre-built numbering functions are provided in this module:
290          ``count_from_0`` and ``count_from_1``.  For more exotic examples
291          like stepped numbering, alphabetical and Fibonacci numbering, see
292          the unit tests.
293
294        :param reorder_on_append:
295          Default False.  When appending an object with an existing (non-None)
296          ordering value, that value will be left untouched unless
297          ``reorder_on_append`` is true.  This is an optimization to avoid a
298          variety of dangerous unexpected database writes.
299
300          SQLAlchemy will add instances to the list via append() when your
301          object loads.  If for some reason the result set from the database
302          skips a step in the ordering (say, row '1' is missing but you get
303          '2', '3', and '4'), reorder_on_append=True would immediately
304          renumber the items to '1', '2', '3'.  If you have multiple sessions
305          making changes, any of whom happen to load this collection even in
306          passing, all of the sessions would try to "clean up" the numbering
307          in their commits, possibly causing all but one to fail with a
308          concurrent modification error.
309
310          Recommend leaving this with the default of False, and just call
311          ``reorder()`` if you're doing ``append()`` operations with
312          previously ordered instances or when doing some housekeeping after
313          manual sql operations.
314
315        """
316        self.ordering_attr = ordering_attr
317        if ordering_func is None:
318            ordering_func = count_from_0
319        self.ordering_func = ordering_func
320        self.reorder_on_append = reorder_on_append
321
322    # More complex serialization schemes (multi column, e.g.) are possible by
323    # subclassing and reimplementing these two methods.
324    def _get_order_value(self, entity: _T) -> Any:
325        return getattr(entity, self.ordering_attr)
326
327    def _set_order_value(self, entity: _T, value: Any) -> None:
328        setattr(entity, self.ordering_attr, value)
329
330    def reorder(self) -> None:
331        """Synchronize ordering for the entire collection.
332
333        Sweeps through the list and ensures that each object has accurate
334        ordering information set.
335
336        """
337        for index, entity in enumerate(self):
338            self._order_entity(index, entity, True)
339
340    # As of 0.5, _reorder is no longer semi-private
341    _reorder = reorder
342
343    def _order_entity(
344        self, index: int, entity: _T, reorder: bool = True
345    ) -> None:
346        have = self._get_order_value(entity)
347
348        # Don't disturb existing ordering if reorder is False
349        if have is not None and not reorder:
350            return
351
352        should_be = self.ordering_func(index, self)
353        if have != should_be:
354            self._set_order_value(entity, should_be)
355
356    def append(self, entity: _T) -> None:
357        super().append(entity)
358        self._order_entity(len(self) - 1, entity, self.reorder_on_append)
359
360    def _raw_append(self, entity: _T) -> None:
361        """Append without any ordering behavior."""
362
363        super().append(entity)
364
365    _raw_append = collection.adds(1)(_raw_append)
366
367    def insert(self, index: SupportsIndex, entity: _T) -> None:
368        super().insert(index, entity)
369        self._reorder()
370
371    def remove(self, entity: _T) -> None:
372        super().remove(entity)
373
374        adapter = collection_adapter(self)
375        if adapter and adapter._referenced_by_owner:
376            self._reorder()
377
378    def pop(self, index: SupportsIndex = -1) -> _T:
379        entity = super().pop(index)
380        self._reorder()
381        return entity
382
383    @overload
384    def __setitem__(self, index: SupportsIndex, entity: _T) -> None: ...
385
386    @overload
387    def __setitem__(self, index: slice, entity: Iterable[_T]) -> None: ...
388
389    def __setitem__(
390        self,
391        index: Union[SupportsIndex, slice],
392        entity: Union[_T, Iterable[_T]],
393    ) -> None:
394        if isinstance(index, slice):
395            step = index.step or 1
396            start = index.start or 0
397            if start < 0:
398                start += len(self)
399            stop = index.stop or len(self)
400            if stop < 0:
401                stop += len(self)
402            entities = list(entity)  # type: ignore[arg-type]
403            for i in range(start, stop, step):
404                self.__setitem__(i, entities[i])
405        else:
406            self._order_entity(int(index), entity, True)  # type: ignore[arg-type] # noqa: E501
407            super().__setitem__(index, entity)  # type: ignore[assignment]
408
409    def __delitem__(self, index: Union[SupportsIndex, slice]) -> None:
410        super().__delitem__(index)
411        self._reorder()
412
413    def __reduce__(self) -> Any:
414        return _reconstitute, (self.__class__, self.__dict__, list(self))
415
416    for func_name, func in list(locals().items()):
417        if (
418            callable(func)
419            and func.__name__ == func_name
420            and not func.__doc__
421            and hasattr(list, func_name)
422        ):
423            func.__doc__ = getattr(list, func_name).__doc__
424    del func_name, func
425
426
427def _reconstitute(
428    cls: Type[OrderingList[_T]], dict_: Dict[str, Any], items: List[_T]
429) -> OrderingList[_T]:
430    """Reconstitute an :class:`.OrderingList`.
431
432    This is the adjoint to :meth:`.OrderingList.__reduce__`.  It is used for
433    unpickling :class:`.OrderingList` objects.
434
435    """
436    obj = cls.__new__(cls)
437    obj.__dict__.update(dict_)
438    list.extend(obj, items)
439    return obj
440