codekingpro/portable-devtools
115k
1# testing/suite/test_reflection.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
9import contextlib
10import operator
11import re
12
13import sqlalchemy as sa
14from .. import config
15from .. import engines
16from .. import eq_
17from .. import eq_regex
18from .. import expect_raises
19from .. import expect_raises_message
20from .. import expect_warnings
21from .. import fixtures
22from .. import is_
23from ..provision import get_temp_table_name
24from ..provision import temp_table_keyword_args
25from ..schema import Column
26from ..schema import Table
27from ... import Boolean
28from ... import DateTime
29from ... import event
30from ... import ForeignKey
31from ... import func
32from ... import Identity
33from ... import inspect
34from ... import Integer
35from ... import MetaData
36from ... import String
37from ... import testing
38from ... import types as sql_types
39from ...engine import Inspector
40from ...engine import ObjectKind
41from ...engine import ObjectScope
42from ...exc import NoSuchTableError
43from ...exc import UnreflectableTableError
44from ...schema import DDL
45from ...schema import Index
46from ...sql.elements import quoted_name
47from ...sql.schema import BLANK_SCHEMA
48from ...testing import ComparesIndexes
49from ...testing import ComparesTables
50from ...testing import is_false
51from ...testing import is_none
52from ...testing import is_true
53from ...testing import mock
54
55
56metadata, users = None, None
57
58
59class OneConnectionTablesTest(fixtures.TablesTest):
60 @classmethod
61 def setup_bind(cls):
62 # TODO: when temp tables are subject to server reset,
63 # this will also have to disable that server reset from
64 # happening
65 if config.requirements.independent_connections.enabled:
66 from sqlalchemy import pool
67
68 return engines.testing_engine(
69 options=dict(poolclass=pool.StaticPool, scope="class"),
70 )
71 else:
72 return config.db
73
74
75class HasTableTest(OneConnectionTablesTest):
76 __sparse_driver_backend__ = True
77
78 run_deletes = None
79
80 @classmethod
81 def define_tables(cls, metadata):
82 Table(
83 "test_table",
84 metadata,
85 Column("id", Integer, primary_key=True),
86 Column("data", String(50)),
87 )
88 if testing.requires.schemas.enabled:
89 Table(
90 "test_table_s",
91 metadata,
92 Column("id", Integer, primary_key=True),
93 Column("data", String(50)),
94 schema=config.test_schema,
95 )
96
97 if testing.requires.view_reflection:
98 cls.define_views(metadata)
99 if testing.requires.has_temp_table.enabled:
100 cls.define_temp_tables(metadata)
101
102 @classmethod
103 def define_views(cls, metadata):
104 query = "CREATE VIEW vv AS SELECT id, data FROM test_table"
105
106 event.listen(metadata, "after_create", DDL(query))
107 event.listen(metadata, "before_drop", DDL("DROP VIEW vv"))
108
109 if testing.requires.schemas.enabled:
110 query = (
111 "CREATE VIEW %s.vv AS SELECT id, data FROM %s.test_table_s"
112 % (
113 config.test_schema,
114 config.test_schema,
115 )
116 )
117 event.listen(metadata, "after_create", DDL(query))
118 event.listen(
119 metadata,
120 "before_drop",
121 DDL("DROP VIEW %s.vv" % (config.test_schema)),
122 )
123
124 @classmethod
125 def temp_table_name(cls):
126 return get_temp_table_name(
127 config, config.db, f"user_tmp_{config.ident}"
128 )
129
130 @classmethod
131 def define_temp_tables(cls, metadata):
132 kw = temp_table_keyword_args(config, config.db)
133 table_name = cls.temp_table_name()
134 user_tmp = Table(
135 table_name,
136 metadata,
137 Column("id", sa.INT, primary_key=True),
138 Column("name", sa.VARCHAR(50)),
139 **kw,
140 )
141 if (
142 testing.requires.view_reflection.enabled
143 and testing.requires.temporary_views.enabled
144 ):
145 event.listen(
146 user_tmp,
147 "after_create",
148 DDL(
149 "create temporary view user_tmp_v as "
150 "select * from user_tmp_%s" % config.ident
151 ),
152 )
153 event.listen(user_tmp, "before_drop", DDL("drop view user_tmp_v"))
154
155 def test_has_table(self):
156 with config.db.begin() as conn:
157 is_true(config.db.dialect.has_table(conn, "test_table"))
158 is_false(config.db.dialect.has_table(conn, "test_table_s"))
159 is_false(config.db.dialect.has_table(conn, "nonexistent_table"))
160
161 def test_has_table_cache(self, metadata):
162 insp = inspect(config.db)
163 is_true(insp.has_table("test_table"))
164 nt = Table("new_table", metadata, Column("col", Integer))
165 is_false(insp.has_table("new_table"))
166 nt.create(config.db)
167 try:
168 is_false(insp.has_table("new_table"))
169 insp.clear_cache()
170 is_true(insp.has_table("new_table"))
171 finally:
172 nt.drop(config.db)
173
174 @testing.requires.schemas
175 def test_has_table_schema(self):
176 with config.db.begin() as conn:
177 is_false(
178 config.db.dialect.has_table(
179 conn, "test_table", schema=config.test_schema
180 )
181 )
182 is_true(
183 config.db.dialect.has_table(
184 conn, "test_table_s", schema=config.test_schema
185 )
186 )
187 is_false(
188 config.db.dialect.has_table(
189 conn, "nonexistent_table", schema=config.test_schema
190 )
191 )
192
193 @testing.requires.schemas
194 def test_has_table_nonexistent_schema(self):
195 with config.db.begin() as conn:
196 is_false(
197 config.db.dialect.has_table(
198 conn, "test_table", schema="nonexistent_schema"
199 )
200 )
201
202 @testing.requires.views
203 def test_has_table_view(self, connection):
204 insp = inspect(connection)
205 is_true(insp.has_table("vv"))
206
207 @testing.requires.has_temp_table
208 def test_has_table_temp_table(self, connection):
209 insp = inspect(connection)
210 temp_table_name = self.temp_table_name()
211 is_true(insp.has_table(temp_table_name))
212
213 @testing.requires.has_temp_table
214 @testing.requires.view_reflection
215 @testing.requires.temporary_views
216 def test_has_table_temp_view(self, connection):
217 insp = inspect(connection)
218 is_true(insp.has_table("user_tmp_v"))
219
220 @testing.requires.views
221 @testing.requires.schemas
222 def test_has_table_view_schema(self, connection):
223 insp = inspect(connection)
224 is_true(insp.has_table("vv", config.test_schema))
225
226
227class HasIndexTest(fixtures.TablesTest):
228 __sparse_driver_backend__ = True
229 __requires__ = ("index_reflection",)
230
231 @classmethod
232 def define_tables(cls, metadata):
233 tt = Table(
234 "test_table",
235 metadata,
236 Column("id", Integer, primary_key=True),
237 Column("data", String(50)),
238 Column("data2", String(50)),
239 )
240 Index("my_idx", tt.c.data)
241
242 if testing.requires.schemas.enabled:
243 tt = Table(
244 "test_table",
245 metadata,
246 Column("id", Integer, primary_key=True),
247 Column("data", String(50)),
248 schema=config.test_schema,
249 )
250 Index("my_idx_s", tt.c.data)
251
252 kind = testing.combinations("dialect", "inspector", argnames="kind")
253
254 def _has_index(self, kind, conn):
255 if kind == "dialect":
256 return lambda *a, **k: config.db.dialect.has_index(conn, *a, **k)
257 else:
258 return inspect(conn).has_index
259
260 @kind
261 def test_has_index(self, kind, connection, metadata):
262 meth = self._has_index(kind, connection)
263 assert meth("test_table", "my_idx")
264 assert not meth("test_table", "my_idx_s")
265 assert not meth("nonexistent_table", "my_idx")
266 assert not meth("test_table", "nonexistent_idx")
267
268 assert not meth("test_table", "my_idx_2")
269 assert not meth("test_table_2", "my_idx_3")
270 idx = Index("my_idx_2", self.tables.test_table.c.data2)
271 tbl = Table(
272 "test_table_2",
273 metadata,
274 Column("foo", Integer),
275 Index("my_idx_3", "foo"),
276 )
277 idx.create(connection)
278 tbl.create(connection)
279 try:
280 if kind == "inspector":
281 assert not meth("test_table", "my_idx_2")
282 assert not meth("test_table_2", "my_idx_3")
283 meth.__self__.clear_cache()
284 assert meth("test_table", "my_idx_2") is True
285 assert meth("test_table_2", "my_idx_3") is True
286 finally:
287 tbl.drop(connection)
288 idx.drop(connection)
289
290 @testing.requires.schemas
291 @kind
292 def test_has_index_schema(self, kind, connection):
293 meth = self._has_index(kind, connection)
294 assert meth("test_table", "my_idx_s", schema=config.test_schema)
295 assert not meth("test_table", "my_idx", schema=config.test_schema)
296 assert not meth(
297 "nonexistent_table", "my_idx_s", schema=config.test_schema
298 )
299 assert not meth(
300 "test_table", "nonexistent_idx_s", schema=config.test_schema
301 )
302
303
304class BizarroCharacterTest(fixtures.TestBase):
305
306 __sparse_driver_backend__ = True
307
308 def column_names():
309 return testing.combinations(
310 ("plainname",),
311 ("(3)",),
312 ("col%p",),
313 ("[brack]",),
314 argnames="columnname",
315 )
316
317 def table_names():
318 return testing.combinations(
319 ("plain",),
320 ("(2)",),
321 ("per % cent",),
322 ("[brackets]",),
323 argnames="tablename",
324 )
325
326 @testing.variation("use_composite", [True, False])
327 @column_names()
328 @table_names()
329 @testing.requires.foreign_key_constraint_reflection
330 def test_fk_ref(
331 self, connection, metadata, use_composite, tablename, columnname
332 ):
333 """tests for #10275"""
334 tt = Table(
335 tablename,
336 metadata,
337 Column(columnname, Integer, key="id", primary_key=True),
338 test_needs_fk=True,
339 )
340 if use_composite:
341 tt.append_column(Column("id2", Integer, primary_key=True))
342
343 if use_composite:
344 Table(
345 "other",
346 metadata,
347 Column("id", Integer, primary_key=True),
348 Column("ref", Integer),
349 Column("ref2", Integer),
350 sa.ForeignKeyConstraint(["ref", "ref2"], [tt.c.id, tt.c.id2]),
351 test_needs_fk=True,
352 )
353 else:
354 Table(
355 "other",
356 metadata,
357 Column("id", Integer, primary_key=True),
358 Column("ref", ForeignKey(tt.c.id)),
359 test_needs_fk=True,
360 )
361
362 metadata.create_all(connection)
363
364 m2 = MetaData()
365
366 o2 = Table("other", m2, autoload_with=connection)
367 t1 = m2.tables[tablename]
368
369 assert o2.c.ref.references(t1.c[0])
370 if use_composite:
371 assert o2.c.ref2.references(t1.c[1])
372
373 @column_names()
374 @table_names()
375 @testing.requires.identity_columns
376 def test_reflect_identity(
377 self, tablename, columnname, connection, metadata
378 ):
379 Table(
380 tablename,
381 metadata,
382 Column(columnname, Integer, Identity(), primary_key=True),
383 )
384 metadata.create_all(connection)
385 insp = inspect(connection)
386
387 eq_(insp.get_columns(tablename)[0]["identity"]["start"], 1)
388
389 @column_names()
390 @table_names()
391 @testing.requires.comment_reflection
392 def test_reflect_comments(
393 self, tablename, columnname, connection, metadata
394 ):
395 Table(
396 tablename,
397 metadata,
398 Column("id", Integer, primary_key=True),
399 Column(columnname, Integer, comment="some comment"),
400 )
401 metadata.create_all(connection)
402 insp = inspect(connection)
403
404 eq_(insp.get_columns(tablename)[1]["comment"], "some comment")
405
406
407class TempTableElementsTest(fixtures.TestBase):
408
409 __sparse_driver_backend__ = True
410
411 __requires__ = ("temp_table_reflection",)
412
413 @testing.fixture
414 def tablename(self):
415 return get_temp_table_name(
416 config, config.db, f"ident_tmp_{config.ident}"
417 )
418
419 @testing.requires.identity_columns
420 def test_reflect_identity(self, tablename, connection, metadata):
421 Table(
422 tablename,
423 metadata,
424 Column("id", Integer, Identity(), primary_key=True),
425 )
426 metadata.create_all(connection)
427 insp = inspect(connection)
428
429 eq_(insp.get_columns(tablename)[0]["identity"]["start"], 1)
430
431 @testing.requires.temp_table_comment_reflection
432 def test_reflect_comments(self, tablename, connection, metadata):
433 Table(
434 tablename,
435 metadata,
436 Column("id", Integer, primary_key=True),
437 Column("foobar", Integer, comment="some comment"),
438 )
439 metadata.create_all(connection)
440 insp = inspect(connection)
441
442 eq_(insp.get_columns(tablename)[1]["comment"], "some comment")
443
444
445class QuotedNameArgumentTest(fixtures.TablesTest):
446 run_create_tables = "once"
447 __sparse_driver_backend__ = True
448
449 @classmethod
450 def define_tables(cls, metadata):
451 Table(
452 "quote ' one",
453 metadata,
454 Column("id", Integer),
455 Column("name", String(50)),
456 Column("data", String(50)),
457 Column("related_id", Integer),
458 sa.PrimaryKeyConstraint("id", name="pk quote ' one"),
459 sa.Index("ix quote ' one", "name"),
460 sa.UniqueConstraint(
461 "data",
462 name="uq quote' one",
463 ),
464 sa.ForeignKeyConstraint(
465 ["id"], ["related.id"], name="fk quote ' one"
466 ),
467 sa.CheckConstraint("name != 'foo'", name="ck quote ' one"),
468 comment=r"""quote ' one comment""",
469 test_needs_fk=True,
470 )
471
472 if testing.requires.symbol_names_w_double_quote.enabled:
473 Table(
474 'quote " two',
475 metadata,
476 Column("id", Integer),
477 Column("name", String(50)),
478 Column("data", String(50)),
479 Column("related_id", Integer),
480 sa.PrimaryKeyConstraint("id", name='pk quote " two'),
481 sa.Index('ix quote " two', "name"),
482 sa.UniqueConstraint(
483 "data",
484 name='uq quote" two',
485 ),
486 sa.ForeignKeyConstraint(
487 ["id"], ["related.id"], name='fk quote " two'
488 ),
489 sa.CheckConstraint("name != 'foo'", name='ck quote " two '),
490 comment=r"""quote " two comment""",
491 test_needs_fk=True,
492 )
493
494 Table(
495 "related",
496 metadata,
497 Column("id", Integer, primary_key=True),
498 Column("related", Integer),
499 test_needs_fk=True,
500 )
501
502 if testing.requires.view_column_reflection.enabled:
503 if testing.requires.symbol_names_w_double_quote.enabled:
504 names = [
505 "quote ' one",
506 'quote " two',
507 ]
508 else:
509 names = [
510 "quote ' one",
511 ]
512 for name in names:
513 query = "CREATE VIEW %s AS SELECT * FROM %s" % (
514 config.db.dialect.identifier_preparer.quote(
515 "view %s" % name
516 ),
517 config.db.dialect.identifier_preparer.quote(name),
518 )
519
520 event.listen(metadata, "after_create", DDL(query))
521 event.listen(
522 metadata,
523 "before_drop",
524 DDL(
525 "DROP VIEW %s"
526 % config.db.dialect.identifier_preparer.quote(
527 "view %s" % name
528 )
529 ),
530 )
531
532 def quote_fixtures(fn):
533 return testing.combinations(
534 ("quote ' one",),
535 ('quote " two', testing.requires.symbol_names_w_double_quote),
536 )(fn)
537
538 @quote_fixtures
539 def test_get_table_options(self, name):
540 insp = inspect(config.db)
541
542 if testing.requires.reflect_table_options.enabled:
543 res = insp.get_table_options(name)
544 is_true(isinstance(res, dict))
545 else:
546 with expect_raises(NotImplementedError):
547 insp.get_table_options(name)
548
549 @quote_fixtures
550 @testing.requires.view_column_reflection
551 def test_get_view_definition(self, name):
552 insp = inspect(config.db)
553 assert insp.get_view_definition("view %s" % name)
554
555 @quote_fixtures
556 def test_get_columns(self, name):
557 insp = inspect(config.db)
558 assert insp.get_columns(name)
559
560 @quote_fixtures
561 def test_get_pk_constraint(self, name):
562 insp = inspect(config.db)
563 assert insp.get_pk_constraint(name)
564
565 @quote_fixtures
566 @testing.requires.foreign_key_constraint_reflection
567 def test_get_foreign_keys(self, name):
568 insp = inspect(config.db)
569 assert insp.get_foreign_keys(name)
570
571 @quote_fixtures
572 @testing.requires.index_reflection
573 def test_get_indexes(self, name):
574 insp = inspect(config.db)
575 assert insp.get_indexes(name)
576
577 @quote_fixtures
578 @testing.requires.unique_constraint_reflection
579 def test_get_unique_constraints(self, name):
580 insp = inspect(config.db)
581 assert insp.get_unique_constraints(name)
582
583 @quote_fixtures
584 @testing.requires.comment_reflection
585 def test_get_table_comment(self, name):
586 insp = inspect(config.db)
587 assert insp.get_table_comment(name)
588
589 @quote_fixtures
590 @testing.requires.check_constraint_reflection
591 def test_get_check_constraints(self, name):
592 insp = inspect(config.db)
593 assert insp.get_check_constraints(name)
594
595
596def _multi_combination(fn):
597 schema = testing.combinations(
598 None,
599 (
600 lambda: config.test_schema,
601 testing.requires.schemas,
602 ),
603 argnames="schema",
604 )
605 scope = testing.combinations(
606 ObjectScope.DEFAULT,
607 ObjectScope.TEMPORARY,
608 ObjectScope.ANY,
609 argnames="scope",
610 )
611 kind = testing.combinations(
612 ObjectKind.TABLE,
613 ObjectKind.VIEW,
614 ObjectKind.MATERIALIZED_VIEW,
615 ObjectKind.ANY,
616 ObjectKind.ANY_VIEW,
617 ObjectKind.TABLE | ObjectKind.VIEW,
618 ObjectKind.TABLE | ObjectKind.MATERIALIZED_VIEW,
619 argnames="kind",
620 )
621 filter_names = testing.combinations(True, False, argnames="use_filter")
622
623 return schema(scope(kind(filter_names(fn))))
624
625
626class ComponentReflectionTest(ComparesTables, OneConnectionTablesTest):
627 run_inserts = run_deletes = None
628
629 __sparse_driver_backend__ = True
630
631 @classmethod
632 def define_tables(cls, metadata):
633 cls.define_reflected_tables(metadata, None)
634 if testing.requires.schemas.enabled:
635 cls.define_reflected_tables(metadata, testing.config.test_schema)
636
637 @classmethod
638 def define_reflected_tables(cls, metadata, schema):
639 if schema:
640 schema_prefix = schema + "."
641 else:
642 schema_prefix = ""
643
644 if testing.requires.self_referential_foreign_keys.enabled:
645 parent_id_args = (
646 ForeignKey(
647 "%susers.user_id" % schema_prefix, name="user_id_fk"
648 ),
649 )
650 else:
651 parent_id_args = ()
652 users = Table(
653 "users",
654 metadata,
655 Column("user_id", sa.INT, primary_key=True),
656 Column("test1", sa.CHAR(5), nullable=False),
657 Column("test2", sa.Float(), nullable=False),
658 Column("parent_user_id", sa.Integer, *parent_id_args),
659 sa.CheckConstraint(
660 "test2 > 0",
661 name="zz_test2_gt_zero",
662 comment="users check constraint",
663 ),
664 sa.CheckConstraint("test2 <= 1000"),
665 schema=schema,
666 test_needs_fk=True,
667 )
668
669 Table(
670 "dingalings",
671 metadata,
672 Column("dingaling_id", sa.Integer, primary_key=True),
673 Column(
674 "address_id",
675 sa.Integer,
676 ForeignKey(
677 "%semail_addresses.address_id" % schema_prefix,
678 name="zz_email_add_id_fg",
679 comment="di fk comment",
680 ),
681 ),
682 Column(
683 "id_user",
684 sa.Integer,
685 ForeignKey("%susers.user_id" % schema_prefix),
686 ),
687 Column("data", sa.String(30), unique=True),
688 sa.CheckConstraint(
689 "address_id > 0 AND address_id < 1000",
690 name="address_id_gt_zero",
691 ),
692 sa.UniqueConstraint(
693 "address_id",
694 "dingaling_id",
695 name="zz_dingalings_multiple",
696 comment="di unique comment",
697 ),
698 schema=schema,
699 test_needs_fk=True,
700 )
701 Table(
702 "email_addresses",
703 metadata,
704 Column("address_id", sa.Integer),
705 Column("remote_user_id", sa.Integer, ForeignKey(users.c.user_id)),
706 Column("email_address", sa.String(20), index=True),
707 sa.PrimaryKeyConstraint(
708 "address_id", name="email_ad_pk", comment="ea pk comment"
709 ),
710 schema=schema,
711 test_needs_fk=True,
712 )
713 Table(
714 "comment_test",
715 metadata,
716 Column("id", sa.Integer, primary_key=True, comment="id comment"),
717 Column("data", sa.String(20), comment="data % comment"),
718 Column(
719 "d2",
720 sa.String(20),
721 comment=r"""Comment types type speedily ' " \ '' Fun!""",
722 ),
723 Column("d3", sa.String(42), comment="Comment\nwith\rescapes"),
724 schema=schema,
725 comment=r"""the test % ' " \ table comment""",
726 )
727 Table(
728 "no_constraints",
729 metadata,
730 Column("data", sa.String(20)),
731 schema=schema,
732 comment="no\nconstraints\rhas\fescaped\vcomment",
733 )
734
735 if testing.requires.cross_schema_fk_reflection.enabled:
736 if schema is None:
737 Table(
738 "local_table",
739 metadata,
740 Column("id", sa.Integer, primary_key=True),
741 Column("data", sa.String(20)),
742 Column(
743 "remote_id",
744 ForeignKey(
745 "%s.remote_table_2.id" % testing.config.test_schema
746 ),
747 ),
748 test_needs_fk=True,
749 schema=config.db.dialect.default_schema_name,
750 )
751 else:
752 Table(
753 "remote_table",
754 metadata,
755 Column("id", sa.Integer, primary_key=True),
756 Column(
757 "local_id",
758 ForeignKey(
759 "%s.local_table.id"
760 % config.db.dialect.default_schema_name
761 ),
762 ),
763 Column("data", sa.String(20)),
764 schema=schema,
765 test_needs_fk=True,
766 )
767 Table(
768 "remote_table_2",
769 metadata,
770 Column("id", sa.Integer, primary_key=True),
771 Column("data", sa.String(20)),
772 schema=schema,
773 test_needs_fk=True,
774 )
775
776 if testing.requires.index_reflection.enabled:
777 Index("users_t_idx", users.c.test1, users.c.test2, unique=True)
778 Index(
779 "users_all_idx", users.c.user_id, users.c.test2, users.c.test1
780 )
781
782 if not schema:
783 # test_needs_fk is at the moment to force MySQL InnoDB
784 noncol_idx_test_nopk = Table(
785 "noncol_idx_test_nopk",
786 metadata,
787 Column("q", sa.String(5)),
788 test_needs_fk=True,
789 )
790
791 noncol_idx_test_pk = Table(
792 "noncol_idx_test_pk",
793 metadata,
794 Column("id", sa.Integer, primary_key=True),
795 Column("q", sa.String(5)),
796 test_needs_fk=True,
797 )
798
799 if (
800 testing.requires.indexes_with_ascdesc.enabled
801 and testing.requires.reflect_indexes_with_ascdesc.enabled
802 ):
803 Index("noncol_idx_nopk", noncol_idx_test_nopk.c.q.desc())
804 Index("noncol_idx_pk", noncol_idx_test_pk.c.q.desc())
805
806 if testing.requires.view_column_reflection.enabled:
807 cls.define_views(metadata, schema)
808 if not schema and testing.requires.temp_table_reflection.enabled:
809 cls.define_temp_tables(metadata)
810
811 @classmethod
812 def temp_table_name(cls):
813 return get_temp_table_name(
814 config, config.db, f"user_tmp_{config.ident}"
815 )
816
817 @classmethod
818 def define_temp_tables(cls, metadata):
819 kw = temp_table_keyword_args(config, config.db)
820 table_name = cls.temp_table_name()
821 user_tmp = Table(
822 table_name,
823 metadata,
824 Column("id", sa.INT, primary_key=True),
825 Column("name", sa.VARCHAR(50)),
826 Column("foo", sa.INT),
827 # disambiguate temp table unique constraint names. this is
828 # pretty arbitrary for a generic dialect however we are doing
829 # it to suit SQL Server which will produce name conflicts for
830 # unique constraints created against temp tables in different
831 # databases.
832 # https://www.arbinada.com/en/node/1645
833 sa.UniqueConstraint("name", name=f"user_tmp_uq_{config.ident}"),
834 sa.Index("user_tmp_ix", "foo"),
835 **kw,
836 )
837 if (
838 testing.requires.view_reflection.enabled
839 and testing.requires.temporary_views.enabled
840 ):
841 event.listen(
842 user_tmp,
843 "after_create",
844 DDL(
845 "create temporary view user_tmp_v as "
846 "select * from user_tmp_%s" % config.ident
847 ),
848 )
849 event.listen(user_tmp, "before_drop", DDL("drop view user_tmp_v"))
850
851 @classmethod
852 def define_views(cls, metadata, schema):
853 if testing.requires.materialized_views.enabled:
854 materialized = {"dingalings"}
855 else:
856 materialized = set()
857 for table_name in ("users", "email_addresses", "dingalings"):
858 fullname = table_name
859 if schema:
860 fullname = f"{schema}.{table_name}"
861 view_name = fullname + "_v"
862 prefix = "MATERIALIZED " if table_name in materialized else ""
863 query = (
864 f"CREATE {prefix}VIEW {view_name} AS SELECT * FROM {fullname}"
865 )
866
867 event.listen(metadata, "after_create", DDL(query))
868 if table_name in materialized:
869 index_name = "mat_index"
870 if schema and testing.against("oracle"):
871 index_name = f"{schema}.{index_name}"
872 idx = f"CREATE INDEX {index_name} ON {view_name}(data)"
873 event.listen(metadata, "after_create", DDL(idx))
874 event.listen(
875 metadata, "before_drop", DDL(f"DROP {prefix}VIEW {view_name}")
876 )
877
878 def _resolve_kind(self, kind, tables, views, materialized):
879 res = {}
880 if ObjectKind.TABLE in kind:
881 res.update(tables)
882 if ObjectKind.VIEW in kind:
883 res.update(views)
884 if ObjectKind.MATERIALIZED_VIEW in kind:
885 res.update(materialized)
886 return res
887
888 def _resolve_views(self, views, materialized):
889 if not testing.requires.view_column_reflection.enabled:
890 materialized.clear()
891 views.clear()
892 elif not testing.requires.materialized_views.enabled:
893 views.update(materialized)
894 materialized.clear()
895
896 def _resolve_names(self, schema, scope, filter_names, values):
897 scope_filter = lambda _: True # noqa: E731
898 if scope is ObjectScope.DEFAULT:
899 scope_filter = lambda k: "tmp" not in k[1] # noqa: E731
900 if scope is ObjectScope.TEMPORARY:
901 scope_filter = lambda k: "tmp" in k[1] # noqa: E731
902
903 removed = {
904 None: {"remote_table", "remote_table_2"},
905 testing.config.test_schema: {
906 "local_table",
907 "noncol_idx_test_nopk",
908 "noncol_idx_test_pk",
909 "user_tmp_v",
910 self.temp_table_name(),
911 },
912 }
913 if not testing.requires.cross_schema_fk_reflection.enabled:
914 removed[None].add("local_table")
915 removed[testing.config.test_schema].update(
916 ["remote_table", "remote_table_2"]
917 )
918 if not testing.requires.index_reflection.enabled:
919 removed[None].update(
920 ["noncol_idx_test_nopk", "noncol_idx_test_pk"]
921 )
922 if (
923 not testing.requires.temp_table_reflection.enabled
924 or not testing.requires.temp_table_names.enabled
925 ):
926 removed[None].update(["user_tmp_v", self.temp_table_name()])
927 if not testing.requires.temporary_views.enabled:
928 removed[None].update(["user_tmp_v"])
929
930 res = {
931 k: v
932 for k, v in values.items()
933 if scope_filter(k)
934 and k[1] not in removed[schema]
935 and (not filter_names or k[1] in filter_names)
936 }
937 return res
938
939 def exp_options(
940 self,
941 schema=None,
942 scope=ObjectScope.ANY,
943 kind=ObjectKind.ANY,
944 filter_names=None,
945 ):
946 materialized = {(schema, "dingalings_v"): mock.ANY}
947 views = {
948 (schema, "email_addresses_v"): mock.ANY,
949 (schema, "users_v"): mock.ANY,
950 (schema, "user_tmp_v"): mock.ANY,
951 }
952 self._resolve_views(views, materialized)
953 tables = {
954 (schema, "users"): mock.ANY,
955 (schema, "dingalings"): mock.ANY,
956 (schema, "email_addresses"): mock.ANY,
957 (schema, "comment_test"): mock.ANY,
958 (schema, "no_constraints"): mock.ANY,
959 (schema, "local_table"): mock.ANY,
960 (schema, "remote_table"): mock.ANY,
961 (schema, "remote_table_2"): mock.ANY,
962 (schema, "noncol_idx_test_nopk"): mock.ANY,
963 (schema, "noncol_idx_test_pk"): mock.ANY,
964 (schema, self.temp_table_name()): mock.ANY,
965 }
966 res = self._resolve_kind(kind, tables, views, materialized)
967 res = self._resolve_names(schema, scope, filter_names, res)
968 return res
969
970 def exp_comments(
971 self,
972 schema=None,
973 scope=ObjectScope.ANY,
974 kind=ObjectKind.ANY,
975 filter_names=None,
976 ):
977 empty = {"text": None}
978 materialized = {(schema, "dingalings_v"): empty}
979 views = {
980 (schema, "email_addresses_v"): empty,
981 (schema, "users_v"): empty,
982 (schema, "user_tmp_v"): empty,
983 }
984 self._resolve_views(views, materialized)
985 tables = {
986 (schema, "users"): empty,
987 (schema, "dingalings"): empty,
988 (schema, "email_addresses"): empty,
989 (schema, "comment_test"): {
990 "text": r"""the test % ' " \ table comment"""
991 },
992 (schema, "no_constraints"): {
993 "text": "no\nconstraints\rhas\fescaped\vcomment"
994 },
995 (schema, "local_table"): empty,
996 (schema, "remote_table"): empty,
997 (schema, "remote_table_2"): empty,
998 (schema, "noncol_idx_test_nopk"): empty,
999 (schema, "noncol_idx_test_pk"): empty,
1000 (schema, self.temp_table_name()): empty,
1001 }
1002 res = self._resolve_kind(kind, tables, views, materialized)
1003 res = self._resolve_names(schema, scope, filter_names, res)
1004 return res
1005
1006 def exp_columns(
1007 self,
1008 schema=None,
1009 scope=ObjectScope.ANY,
1010 kind=ObjectKind.ANY,
1011 filter_names=None,
1012 ):
1013 def col(
1014 name, auto=False, default=mock.ANY, comment=None, nullable=True
1015 ):
1016 res = {
1017 "name": name,
1018 "autoincrement": auto,
1019 "type": mock.ANY,
1020 "default": default,
1021 "comment": comment,
1022 "nullable": nullable,
1023 }
1024 if auto == "omit":
1025 res.pop("autoincrement")
1026 return res
1027
1028 def pk(name, **kw):
1029 kw = {"auto": True, "default": mock.ANY, "nullable": False, **kw}
1030 return col(name, **kw)
1031
1032 materialized = {
1033 (schema, "dingalings_v"): [
1034 col("dingaling_id", auto="omit", nullable=mock.ANY),
1035 col("address_id"),
1036 col("id_user"),
1037 col("data"),
1038 ]
1039 }
1040 views = {
1041 (schema, "email_addresses_v"): [
1042 col("address_id", auto="omit", nullable=mock.ANY),
1043 col("remote_user_id"),
1044 col("email_address"),
1045 ],
1046 (schema, "users_v"): [
1047 col("user_id", auto="omit", nullable=mock.ANY),
1048 col("test1", nullable=mock.ANY),
1049 col("test2", nullable=mock.ANY),
1050 col("parent_user_id"),
1051 ],
1052 (schema, "user_tmp_v"): [
1053 col("id", auto="omit", nullable=mock.ANY),
1054 col("name"),
1055 col("foo"),
1056 ],
1057 }
1058 self._resolve_views(views, materialized)
1059 tables = {
1060 (schema, "users"): [
1061 pk("user_id"),
1062 col("test1", nullable=False),
1063 col("test2", nullable=False),
1064 col("parent_user_id"),
1065 ],
1066 (schema, "dingalings"): [
1067 pk("dingaling_id"),
1068 col("address_id"),
1069 col("id_user"),
1070 col("data"),
1071 ],
1072 (schema, "email_addresses"): [
1073 pk("address_id"),
1074 col("remote_user_id"),
1075 col("email_address"),
1076 ],
1077 (schema, "comment_test"): [
1078 pk("id", comment="id comment"),
1079 col("data", comment="data % comment"),
1080 col(
1081 "d2",
1082 comment=r"""Comment types type speedily ' " \ '' Fun!""",
1083 ),
1084 col("d3", comment="Comment\nwith\rescapes"),
1085 ],
1086 (schema, "no_constraints"): [col("data")],
1087 (schema, "local_table"): [pk("id"), col("data"), col("remote_id")],
1088 (schema, "remote_table"): [pk("id"), col("local_id"), col("data")],
1089 (schema, "remote_table_2"): [pk("id"), col("data")],
1090 (schema, "noncol_idx_test_nopk"): [col("q")],
1091 (schema, "noncol_idx_test_pk"): [pk("id"), col("q")],
1092 (schema, self.temp_table_name()): [
1093 pk("id"),
1094 col("name"),
1095 col("foo"),
1096 ],
1097 }
1098 res = self._resolve_kind(kind, tables, views, materialized)
1099 res = self._resolve_names(schema, scope, filter_names, res)
1100 return res
1101
1102 @property
1103 def _required_column_keys(self):
1104 return {"name", "type", "nullable", "default"}
1105
1106 def exp_pks(
1107 self,
1108 schema=None,
1109 scope=ObjectScope.ANY,
1110 kind=ObjectKind.ANY,
1111 filter_names=None,
1112 ):
1113 def pk(*cols, name=mock.ANY, comment=None):
1114 return {
1115 "constrained_columns": list(cols),
1116 "name": name,
1117 "comment": comment,
1118 }
1119
1120 empty = pk(name=None)
1121 if testing.requires.materialized_views_reflect_pk.enabled:
1122 materialized = {(schema, "dingalings_v"): pk("dingaling_id")}
1123 else:
1124 materialized = {(schema, "dingalings_v"): empty}
1125 views = {
1126 (schema, "email_addresses_v"): empty,
1127 (schema, "users_v"): empty,
1128 (schema, "user_tmp_v"): empty,
1129 }
1130 self._resolve_views(views, materialized)
1131 tables = {
1132 (schema, "users"): pk("user_id"),
1133 (schema, "dingalings"): pk("dingaling_id"),
1134 (schema, "email_addresses"): pk(
1135 "address_id", name="email_ad_pk", comment="ea pk comment"
1136 ),
1137 (schema, "comment_test"): pk("id"),
1138 (schema, "no_constraints"): empty,
1139 (schema, "local_table"): pk("id"),
1140 (schema, "remote_table"): pk("id"),
1141 (schema, "remote_table_2"): pk("id"),
1142 (schema, "noncol_idx_test_nopk"): empty,
1143 (schema, "noncol_idx_test_pk"): pk("id"),
1144 (schema, self.temp_table_name()): pk("id"),
1145 }
1146 if not testing.requires.reflects_pk_names.enabled:
1147 for val in tables.values():
1148 if val["name"] is not None:
1149 val["name"] = mock.ANY
1150 res = self._resolve_kind(kind, tables, views, materialized)
1151 res = self._resolve_names(schema, scope, filter_names, res)
1152 return res
1153
1154 @property
1155 def _required_pk_keys(self):
1156 return {"name", "constrained_columns"}
1157
1158 def exp_fks(
1159 self,
1160 schema=None,
1161 scope=ObjectScope.ANY,
1162 kind=ObjectKind.ANY,
1163 filter_names=None,
1164 ):
1165 class tt:
1166 def __eq__(self, other):
1167 return (
1168 other is None
1169 or config.db.dialect.default_schema_name == other
1170 )
1171
1172 def fk(
1173 cols,
1174 ref_col,
1175 ref_table,
1176 ref_schema=schema,
1177 name=mock.ANY,
1178 comment=None,
1179 ):
1180 return {
1181 "constrained_columns": cols,
1182 "referred_columns": ref_col,
1183 "name": name,
1184 "options": mock.ANY,
1185 "referred_schema": (
1186 ref_schema if ref_schema is not None else tt()
1187 ),
1188 "referred_table": ref_table,
1189 "comment": comment,
1190 }
1191
1192 materialized = {(schema, "dingalings_v"): []}
1193 views = {
1194 (schema, "email_addresses_v"): [],
1195 (schema, "users_v"): [],
1196 (schema, "user_tmp_v"): [],
1197 }
1198 self._resolve_views(views, materialized)
1199 tables = {
1200 (schema, "users"): [