From 6d7ba58f880be618ade07f8ea080fe8c4bf8a896 Mon Sep 17 00:00:00 2001 From: cyfraeviolae Date: Wed, 3 Apr 2024 03:10:44 -0400 Subject: venv --- .../site-packages/sqlalchemy/testing/schema.py | 224 +++++++++++++++++++++ 1 file changed, 224 insertions(+) create mode 100644 venv/lib/python3.11/site-packages/sqlalchemy/testing/schema.py (limited to 'venv/lib/python3.11/site-packages/sqlalchemy/testing/schema.py') diff --git a/venv/lib/python3.11/site-packages/sqlalchemy/testing/schema.py b/venv/lib/python3.11/site-packages/sqlalchemy/testing/schema.py new file mode 100644 index 0000000..7dfd33d --- /dev/null +++ b/venv/lib/python3.11/site-packages/sqlalchemy/testing/schema.py @@ -0,0 +1,224 @@ +# testing/schema.py +# Copyright (C) 2005-2024 the SQLAlchemy authors and contributors +# +# +# This module is part of SQLAlchemy and is released under +# the MIT License: https://www.opensource.org/licenses/mit-license.php +# mypy: ignore-errors + +from __future__ import annotations + +import sys + +from . import config +from . import exclusions +from .. import event +from .. import schema +from .. import types as sqltypes +from ..orm import mapped_column as _orm_mapped_column +from ..util import OrderedDict + +__all__ = ["Table", "Column"] + +table_options = {} + + +def Table(*args, **kw) -> schema.Table: + """A schema.Table wrapper/hook for dialect-specific tweaks.""" + + test_opts = {k: kw.pop(k) for k in list(kw) if k.startswith("test_")} + + kw.update(table_options) + + if exclusions.against(config._current, "mysql"): + if ( + "mysql_engine" not in kw + and "mysql_type" not in kw + and "autoload_with" not in kw + ): + if "test_needs_fk" in test_opts or "test_needs_acid" in test_opts: + kw["mysql_engine"] = "InnoDB" + else: + # there are in fact test fixtures that rely upon MyISAM, + # due to MySQL / MariaDB having poor FK behavior under innodb, + # such as a self-referential table can't be deleted from at + # once without attending to per-row dependencies. We'd need to + # add special steps to some fixtures if we want to not + # explicitly state MyISAM here + kw["mysql_engine"] = "MyISAM" + elif exclusions.against(config._current, "mariadb"): + if ( + "mariadb_engine" not in kw + and "mariadb_type" not in kw + and "autoload_with" not in kw + ): + if "test_needs_fk" in test_opts or "test_needs_acid" in test_opts: + kw["mariadb_engine"] = "InnoDB" + else: + kw["mariadb_engine"] = "MyISAM" + + return schema.Table(*args, **kw) + + +def mapped_column(*args, **kw): + """An orm.mapped_column wrapper/hook for dialect-specific tweaks.""" + + return _schema_column(_orm_mapped_column, args, kw) + + +def Column(*args, **kw): + """A schema.Column wrapper/hook for dialect-specific tweaks.""" + + return _schema_column(schema.Column, args, kw) + + +def _schema_column(factory, args, kw): + test_opts = {k: kw.pop(k) for k in list(kw) if k.startswith("test_")} + + if not config.requirements.foreign_key_ddl.enabled_for_config(config): + args = [arg for arg in args if not isinstance(arg, schema.ForeignKey)] + + construct = factory(*args, **kw) + + if factory is schema.Column: + col = construct + else: + col = construct.column + + if test_opts.get("test_needs_autoincrement", False) and kw.get( + "primary_key", False + ): + if col.default is None and col.server_default is None: + col.autoincrement = True + + # allow any test suite to pick up on this + col.info["test_needs_autoincrement"] = True + + # hardcoded rule for oracle; this should + # be moved out + if exclusions.against(config._current, "oracle"): + + def add_seq(c, tbl): + c._init_items( + schema.Sequence( + _truncate_name( + config.db.dialect, tbl.name + "_" + c.name + "_seq" + ), + optional=True, + ) + ) + + event.listen(col, "after_parent_attach", add_seq, propagate=True) + return construct + + +class eq_type_affinity: + """Helper to compare types inside of datastructures based on affinity. + + E.g.:: + + eq_( + inspect(connection).get_columns("foo"), + [ + { + "name": "id", + "type": testing.eq_type_affinity(sqltypes.INTEGER), + "nullable": False, + "default": None, + "autoincrement": False, + }, + { + "name": "data", + "type": testing.eq_type_affinity(sqltypes.NullType), + "nullable": True, + "default": None, + "autoincrement": False, + }, + ], + ) + + """ + + def __init__(self, target): + self.target = sqltypes.to_instance(target) + + def __eq__(self, other): + return self.target._type_affinity is other._type_affinity + + def __ne__(self, other): + return self.target._type_affinity is not other._type_affinity + + +class eq_compile_type: + """similar to eq_type_affinity but uses compile""" + + def __init__(self, target): + self.target = target + + def __eq__(self, other): + return self.target == other.compile() + + def __ne__(self, other): + return self.target != other.compile() + + +class eq_clause_element: + """Helper to compare SQL structures based on compare()""" + + def __init__(self, target): + self.target = target + + def __eq__(self, other): + return self.target.compare(other) + + def __ne__(self, other): + return not self.target.compare(other) + + +def _truncate_name(dialect, name): + if len(name) > dialect.max_identifier_length: + return ( + name[0 : max(dialect.max_identifier_length - 6, 0)] + + "_" + + hex(hash(name) % 64)[2:] + ) + else: + return name + + +def pep435_enum(name): + # Implements PEP 435 in the minimal fashion needed by SQLAlchemy + __members__ = OrderedDict() + + def __init__(self, name, value, alias=None): + self.name = name + self.value = value + self.__members__[name] = self + value_to_member[value] = self + setattr(self.__class__, name, self) + if alias: + self.__members__[alias] = self + setattr(self.__class__, alias, self) + + value_to_member = {} + + @classmethod + def get(cls, value): + return value_to_member[value] + + someenum = type( + name, + (object,), + {"__members__": __members__, "__init__": __init__, "get": get}, + ) + + # getframe() trick for pickling I don't understand courtesy + # Python namedtuple() + try: + module = sys._getframe(1).f_globals.get("__name__", "__main__") + except (AttributeError, ValueError): + pass + if module is not None: + someenum.__module__ = module + + return someenum -- cgit v1.2.3