codekingpro/portable-devtools

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