codekingpro/portable-devtools

1# testing/schema.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
9from __future__ import annotations
10
11import sys
12
13from . import config
14from . import exclusions
15from .. import event
16from .. import schema
17from .. import types as sqltypes
18from ..orm import mapped_column as _orm_mapped_column
19from ..util import OrderedDict
20
21__all__ = ["Table", "Column"]
22
23table_options = {}
24
25
26def Table(*args, **kw) -> schema.Table:
27    """A schema.Table wrapper/hook for dialect-specific tweaks."""
28
29    test_opts = {k: kw.pop(k) for k in list(kw) if k.startswith("test_")}
30
31    kw.update(table_options)
32
33    if exclusions.against(config._current, "mysql"):
34        if (
35            "mysql_engine" not in kw
36            and "mysql_type" not in kw
37            and "autoload_with" not in kw
38        ):
39            if "test_needs_fk" in test_opts or "test_needs_acid" in test_opts:
40                kw["mysql_engine"] = "InnoDB"
41            else:
42                # there are in fact test fixtures that rely upon MyISAM,
43                # due to MySQL / MariaDB having poor FK behavior under innodb,
44                # such as a self-referential table can't be deleted from at
45                # once without attending to per-row dependencies.  We'd need to
46                # add special steps to some fixtures if we want to not
47                # explicitly state MyISAM here
48                kw["mysql_engine"] = "MyISAM"
49    elif exclusions.against(config._current, "mariadb"):
50        if (
51            "mariadb_engine" not in kw
52            and "mariadb_type" not in kw
53            and "autoload_with" not in kw
54        ):
55            if "test_needs_fk" in test_opts or "test_needs_acid" in test_opts:
56                kw["mariadb_engine"] = "InnoDB"
57            else:
58                kw["mariadb_engine"] = "MyISAM"
59
60    return schema.Table(*args, **kw)
61
62
63def mapped_column(*args, **kw):
64    """An orm.mapped_column wrapper/hook for dialect-specific tweaks."""
65
66    return _schema_column(_orm_mapped_column, args, kw)
67
68
69def Column(*args, **kw):
70    """A schema.Column wrapper/hook for dialect-specific tweaks."""
71
72    return _schema_column(schema.Column, args, kw)
73
74
75def _schema_column(factory, args, kw):
76    test_opts = {k: kw.pop(k) for k in list(kw) if k.startswith("test_")}
77
78    if not config.requirements.foreign_key_ddl.enabled_for_config(config):
79        args = [arg for arg in args if not isinstance(arg, schema.ForeignKey)]
80
81    construct = factory(*args, **kw)
82
83    if factory is schema.Column:
84        col = construct
85    else:
86        col = construct.column
87
88    if test_opts.get("test_needs_autoincrement", False) and kw.get(
89        "primary_key", False
90    ):
91        if col.default is None and col.server_default is None:
92            col.autoincrement = True
93
94        # allow any test suite to pick up on this
95        col.info["test_needs_autoincrement"] = True
96
97        # hardcoded rule for oracle; this should
98        # be moved out
99        if exclusions.against(config._current, "oracle"):
100
101            def add_seq(c, tbl):
102                c._init_items(
103                    schema.Sequence(
104                        _truncate_name(
105                            config.db.dialect, tbl.name + "_" + c.name + "_seq"
106                        ),
107                        optional=True,
108                    )
109                )
110
111            event.listen(col, "after_parent_attach", add_seq, propagate=True)
112    return construct
113
114
115class eq_type_affinity:
116    """Helper to compare types inside of datastructures based on affinity.
117
118    E.g.::
119
120        eq_(
121            inspect(connection).get_columns("foo"),
122            [
123                {
124                    "name": "id",
125                    "type": testing.eq_type_affinity(sqltypes.INTEGER),
126                    "nullable": False,
127                    "default": None,
128                    "autoincrement": False,
129                },
130                {
131                    "name": "data",
132                    "type": testing.eq_type_affinity(sqltypes.NullType),
133                    "nullable": True,
134                    "default": None,
135                    "autoincrement": False,
136                },
137            ],
138        )
139
140    """
141
142    def __init__(self, target):
143        self.target = sqltypes.to_instance(target)
144
145    def __eq__(self, other):
146        return self.target._type_affinity is other._type_affinity
147
148    def __ne__(self, other):
149        return self.target._type_affinity is not other._type_affinity
150
151
152class eq_compile_type:
153    """similar to eq_type_affinity but uses compile"""
154
155    def __init__(self, target):
156        self.target = target
157
158    def __eq__(self, other):
159        return self.target == other.compile()
160
161    def __ne__(self, other):
162        return self.target != other.compile()
163
164
165class eq_clause_element:
166    """Helper to compare SQL structures based on compare()"""
167
168    def __init__(self, target):
169        self.target = target
170
171    def __eq__(self, other):
172        return self.target.compare(other)
173
174    def __ne__(self, other):
175        return not self.target.compare(other)
176
177
178def _truncate_name(dialect, name):
179    if len(name) > dialect.max_identifier_length:
180        return (
181            name[0 : max(dialect.max_identifier_length - 6, 0)]
182            + "_"
183            + hex(hash(name) % 64)[2:]
184        )
185    else:
186        return name
187
188
189def pep435_enum(name):
190    # Implements PEP 435 in the minimal fashion needed by SQLAlchemy
191    __members__ = OrderedDict()
192
193    def __init__(self, name, value, alias=None):
194        self.name = name
195        self.value = value
196        self.__members__[name] = self
197        value_to_member[value] = self
198        setattr(self.__class__, name, self)
199        if alias:
200            self.__members__[alias] = self
201            setattr(self.__class__, alias, self)
202
203    value_to_member = {}
204
205    @classmethod
206    def get(cls, value):
207        return value_to_member[value]
208
209    someenum = type(
210        name,
211        (object,),
212        {"__members__": __members__, "__init__": __init__, "get": get},
213    )
214
215    # getframe() trick for pickling I don't understand courtesy
216    # Python namedtuple()
217    try:
218        module = sys._getframe(1).f_globals.get("__name__", "__main__")
219    except (AttributeError, ValueError):
220        pass
221    if module is not None:
222        someenum.__module__ = module
223
224    return someenum
225