codekingpro/portable-devtools

1# testing/entities.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# mypy: ignore-errors
8
9
10from __future__ import annotations
11
12import sqlalchemy as sa
13from .. import exc as sa_exc
14from ..orm.writeonly import WriteOnlyCollection
15
16_repr_stack = set()
17
18
19class BasicEntity:
20    def __init__(self, **kw):
21        for key, value in kw.items():
22            setattr(self, key, value)
23
24    def __repr__(self):
25        if id(self) in _repr_stack:
26            return object.__repr__(self)
27        _repr_stack.add(id(self))
28        try:
29            return "%s(%s)" % (
30                (self.__class__.__name__),
31                ", ".join(
32                    [
33                        "%s=%r" % (key, getattr(self, key))
34                        for key in sorted(self.__dict__.keys())
35                        if not key.startswith("_")
36                    ]
37                ),
38            )
39        finally:
40            _repr_stack.remove(id(self))
41
42
43_recursion_stack = set()
44
45
46class ComparableMixin:
47    def __ne__(self, other):
48        return not self.__eq__(other)
49
50    def __eq__(self, other):
51        """'Deep, sparse compare.
52
53        Deeply compare two entities, following the non-None attributes of the
54        non-persisted object, if possible.
55
56        """
57        if other is self:
58            return True
59        elif not self.__class__ == other.__class__:
60            return False
61
62        if id(self) in _recursion_stack:
63            return True
64        _recursion_stack.add(id(self))
65
66        try:
67            # pick the entity that's not SA persisted as the source
68            try:
69                self_key = sa.orm.attributes.instance_state(self).key
70            except sa.orm.exc.NO_STATE:
71                self_key = None
72
73            if other is None:
74                a = self
75                b = other
76            elif self_key is not None:
77                a = other
78                b = self
79            else:
80                a = self
81                b = other
82
83            for attr in list(a.__dict__):
84                if attr.startswith("_"):
85                    continue
86
87                value = getattr(a, attr)
88
89                if isinstance(value, WriteOnlyCollection):
90                    continue
91
92                try:
93                    # handle lazy loader errors
94                    battr = getattr(b, attr)
95                except (AttributeError, sa_exc.UnboundExecutionError):
96                    return False
97
98                if hasattr(value, "__iter__") and not isinstance(value, str):
99                    if hasattr(value, "__getitem__") and not hasattr(
100                        value, "keys"
101                    ):
102                        if list(value) != list(battr):
103                            return False
104                    else:
105                        if set(value) != set(battr):
106                            return False
107                else:
108                    if value is not None and value != battr:
109                        return False
110            return True
111        finally:
112            _recursion_stack.remove(id(self))
113
114
115class ComparableEntity(ComparableMixin, BasicEntity):
116    def __hash__(self):
117        return hash(self.__class__)
118