codekingpro/portable-devtools

1# testing/schema.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
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    # pop out local options; these are not used at the moment
30    _ = {k: kw.pop(k) for k in list(kw) if k.startswith("test_")}
31
32    kw.update(table_options)
33
34    return schema.Table(*args, **kw)
35
36
37def mapped_column(*args, **kw):
38    """An orm.mapped_column wrapper/hook for dialect-specific tweaks."""
39
40    return _schema_column(_orm_mapped_column, args, kw)
41
42
43def Column(*args, **kw):
44    """A schema.Column wrapper/hook for dialect-specific tweaks."""
45
46    return _schema_column(schema.Column, args, kw)
47
48
49def _schema_column(factory, args, kw):
50    test_opts = {k: kw.pop(k) for k in list(kw) if k.startswith("test_")}
51
52    if not config.requirements.foreign_key_ddl.enabled_for_config(config):
53        args = [arg for arg in args if not isinstance(arg, schema.ForeignKey)]
54
55    construct = factory(*args, **kw)
56
57    if factory is schema.Column:
58        col = construct
59    else:
60        col = construct.column
61
62    if test_opts.get("test_needs_autoincrement", False) and kw.get(
63        "primary_key", False
64    ):
65        if col.default is None and col.server_default is None:
66            col.autoincrement = True
67
68        # allow any test suite to pick up on this
69        col.info["test_needs_autoincrement"] = True
70
71        # hardcoded rule for oracle; this should
72        # be moved out
73        if exclusions.against(config._current, "oracle"):
74
75            def add_seq(c, tbl):
76                c._init_items(
77                    schema.Sequence(
78                        _truncate_name(
79                            config.db.dialect, tbl.name + "_" + c.name + "_seq"
80                        ),
81                        optional=True,
82                    )
83                )
84
85            event.listen(col, "after_parent_attach", add_seq, propagate=True)
86    return construct
87
88
89class eq_type_affinity:
90    """Helper to compare types inside of datastructures based on affinity.
91
92    E.g.::
93
94        eq_(
95            inspect(connection).get_columns("foo"),
96            [
97                {
98                    "name": "id",
99                    "type": testing.eq_type_affinity(sqltypes.INTEGER),
100                    "nullable": False,
101                    "default": None,
102                    "autoincrement": False,
103                },
104                {
105                    "name": "data",
106                    "type": testing.eq_type_affinity(sqltypes.NullType),
107                    "nullable": True,
108                    "default": None,
109                    "autoincrement": False,
110                },
111            ],
112        )
113
114    """
115
116    def __init__(self, target):
117        self.target = sqltypes.to_instance(target)
118
119    def __eq__(self, other):
120        return self.target._type_affinity is other._type_affinity
121
122    def __ne__(self, other):
123        return self.target._type_affinity is not other._type_affinity
124
125
126class eq_compile_type:
127    """similar to eq_type_affinity but uses compile"""
128
129    def __init__(self, target):
130        self.target = target
131
132    def __eq__(self, other):
133        return self.target == other.compile()
134
135    def __ne__(self, other):
136        return self.target != other.compile()
137
138
139class eq_clause_element:
140    """Helper to compare SQL structures based on compare()"""
141
142    def __init__(self, target):
143        self.target = target
144
145    def __eq__(self, other):
146        return self.target.compare(other)
147
148    def __ne__(self, other):
149        return not self.target.compare(other)
150
151
152def _truncate_name(dialect, name):
153    if len(name) > dialect.max_identifier_length:
154        return (
155            name[0 : max(dialect.max_identifier_length - 6, 0)]
156            + "_"
157            + hex(hash(name) % 64)[2:]
158        )
159    else:
160        return name
161
162
163def pep435_enum(name):
164    # Implements PEP 435 in the minimal fashion needed by SQLAlchemy
165    __members__ = OrderedDict()
166
167    def __init__(self, name, value, alias=None):
168        self.name = name
169        self.value = value
170        self.__members__[name] = self
171        value_to_member[value] = self
172        setattr(self.__class__, name, self)
173        if alias:
174            self.__members__[alias] = self
175            setattr(self.__class__, alias, self)
176
177    value_to_member = {}
178
179    @classmethod
180    def get(cls, value):
181        return value_to_member[value]
182
183    someenum = type(
184        name,
185        (object,),
186        {"__members__": __members__, "__init__": __init__, "get": get},
187    )
188
189    # getframe() trick for pickling I don't understand courtesy
190    # Python namedtuple()
191    try:
192        module = sys._getframe(1).f_globals.get("__name__", "__main__")
193    except (AttributeError, ValueError):
194        pass
195    if module is not None:
196        someenum.__module__ = module
197
198    return someenum
199