from contextlib import contextmanager

import pickle

import sqlalchemy as tsa

from sqlalchemy import ARRAY

from sqlalchemy import bindparam

from sqlalchemy import BLANK_SCHEMA

from sqlalchemy import Boolean

from sqlalchemy import CheckConstraint

from sqlalchemy import Column

from sqlalchemy import ColumnDefault

from sqlalchemy import desc

from sqlalchemy import Enum

from sqlalchemy import event

from sqlalchemy import exc

from sqlalchemy import ForeignKey

from sqlalchemy import ForeignKeyConstraint

from sqlalchemy import func

from sqlalchemy import Index

from sqlalchemy import Integer

from sqlalchemy import MetaData

from sqlalchemy import PrimaryKeyConstraint

from sqlalchemy import schema

from sqlalchemy import Sequence

from sqlalchemy import String

from sqlalchemy import Table

from sqlalchemy import testing

from sqlalchemy import text

from sqlalchemy import TypeDecorator

from sqlalchemy import types as sqltypes

from sqlalchemy import Unicode

from sqlalchemy import UniqueConstraint

from sqlalchemy.engine import default

from sqlalchemy.schema import AddConstraint

from sqlalchemy.schema import CreateIndex

from sqlalchemy.schema import DefaultClause

from sqlalchemy.schema import DropIndex

from sqlalchemy.sql import naming

from sqlalchemy.sql import operators

from sqlalchemy.sql.elements import _NONE_NAME

from sqlalchemy.testing import assert_raises

from sqlalchemy.testing import assert_raises_message

from sqlalchemy.testing import AssertsCompiledSQL

from sqlalchemy.testing import ComparesTables

from sqlalchemy.testing import emits_warning

from sqlalchemy.testing import eq_

from sqlalchemy.testing import fixtures

from sqlalchemy.testing import is_

from sqlalchemy.testing import is_false

from sqlalchemy.testing import is_true

from sqlalchemy.testing import mock

class MetaDataTest(fixtures.TestBase, ComparesTables):

def test_metadata_contains(self):

metadata = MetaData()

t1 = Table("t1", metadata, Column("x", Integer))

t2 = Table("t2", metadata, Column("x", Integer), schema="foo")

t3 = Table("t2", MetaData(), Column("x", Integer))

t4 = Table("t1", MetaData(), Column("x", Integer), schema="foo")

assert "t1" in metadata

assert "foo.t2" in metadata

assert "t2" not in metadata

assert "foo.t1" not in metadata

assert t1 in metadata

assert t2 in metadata

assert t3 not in metadata

assert t4 not in metadata

def test_uninitialized_column_copy(self):

for col in [

Column("foo", String(), nullable=False),

Column("baz", String(), unique=True),

Column(Integer(), primary_key=True),

Column(

"bar",

Integer(),

Sequence("foo_seq"),

primary_key=True,

key="bar",

),

Column(Integer(), ForeignKey("bat.blah"), doc="this is a col"),

Column(

"bar",

Integer(),

ForeignKey("bat.blah"),

primary_key=True,

key="bar",

),

Column("bar", Integer(), info={"foo": "bar"}),

]:

c2 = col.copy()

for attr in (

"name",

"type",

"nullable",

"primary_key",

"key",

"unique",

"info",

"doc",

):

eq_(getattr(col, attr), getattr(c2, attr))

eq_(len(col.foreign_keys), len(c2.foreign_keys))

if col.default:

eq_(c2.default.name, "foo_seq")

for a1, a2 in zip(col.foreign_keys, c2.foreign_keys):

assert a1 is not a2

eq_(a2._colspec, "bat.blah")

def test_col_subclass_copy(self):

class MyColumn(schema.Column):

def __init__(self, *args, **kw):

self.widget = kw.pop("widget", None)

super(MyColumn, self).__init__(*args, **kw)

def copy(self, *arg, **kw):

c = super(MyColumn, self).copy(*arg, **kw)

c.widget = self.widget

return c

c1 = MyColumn("foo", Integer, widget="x")

c2 = c1.copy()

assert isinstance(c2, MyColumn)

eq_(c2.widget, "x")

def test_uninitialized_column_copy_events(self):

msgs = []

def write(c, t):

msgs.append("attach %s.%s" % (t.name, c.name))

c1 = Column("foo", String())

m = MetaData()

for i in range(3):

cx = c1.copy()

# as of 0.7, these events no longer copy. its expected

# that listeners will be re-established from the

# natural construction of things.

cx._on_table_attach(write)

Table("foo%d" % i, m, cx)

eq_(msgs, ["attach foo0.foo", "attach foo1.foo", "attach foo2.foo"])

def test_schema_collection_add(self):

metadata = MetaData()

Table("t1", metadata, Column("x", Integer), schema="foo")

Table("t2", metadata, Column("x", Integer), schema="bar")

Table("t3", metadata, Column("x", Integer))

eq_(metadata._schemas, set(["foo", "bar"]))

eq_(len(metadata.tables), 3)

def test_schema_collection_remove(self):

metadata = MetaData()

t1 = Table("t1", metadata, Column("x", Integer), schema="foo")

Table("t2", metadata, Column("x", Integer), schema="bar")

t3 = Table("t3", metadata, Column("x", Integer), schema="bar")

metadata.remove(t3)

eq_(metadata._schemas, set(["foo", "bar"]))

eq_(len(metadata.tables), 2)

metadata.remove(t1)

eq_(metadata._schemas, set(["bar"]))

eq_(len(metadata.tables), 1)

def test_schema_collection_remove_all(self):

metadata = MetaData()

Table("t1", metadata, Column("x", Integer), schema="foo")

Table("t2", metadata, Column("x", Integer), schema="bar")

metadata.clear()

eq_(metadata._schemas, set())

eq_(len(metadata.tables), 0)

def test_metadata_tables_immutable(self):

# this use case was added due to #1917.

metadata = MetaData()

Table("t1", metadata, Column("x", Integer))

assert "t1" in metadata.tables

assert_raises(TypeError, lambda: metadata.tables.pop("t1"))

@testing.provide_metadata

def test_dupe_tables(self):

metadata = self.metadata

Table(

"table1",

metadata,

Column("col1", Integer, primary_key=True),

Column("col2", String(20)),

)

metadata.create_all()

Table("table1", metadata, autoload=True)

def go():

Table(

"table1",

metadata,

Column("col1", Integer, primary_key=True),

Column("col2", String(20)),

)

assert_raises_message(

tsa.exc.InvalidRequestError,

"Table 'table1' is already defined for this "

"MetaData instance. Specify 'extend_existing=True' "

"to redefine options and columns on an existing "

"Table object.",

go,

)

def test_fk_copy(self):

c1 = Column("foo", Integer)

c2 = Column("bar", Integer)

m = MetaData()

t1 = Table("t", m, c1, c2)

kw = dict(

onupdate="X",

ondelete="Y",

use_alter=True,

name="f1",

deferrable="Z",

initially="Q",

link_to_name=True,

)

fk1 = ForeignKey(c1, **kw)

fk2 = ForeignKeyConstraint((c1,), (c2,), **kw)

t1.append_constraint(fk2)

fk1c = fk1.copy()

fk2c = fk2.copy()

for k in kw:

eq_(getattr(fk1c, k), kw[k])

eq_(getattr(fk2c, k), kw[k])

def test_check_constraint_copy(self):

def r(x):

return x

c = CheckConstraint(

"foo bar",

name="name",

initially=True,

deferrable=True,

_create_rule=r,

)

c2 = c.copy()

eq_(c2.name, "name")

eq_(str(c2.sqltext), "foo bar")

eq_(c2.initially, True)

eq_(c2.deferrable, True)

assert c2._create_rule is r

def test_col_replace_w_constraint(self):

m = MetaData()

a = Table("a", m, Column("id", Integer, primary_key=True))

aid = Column("a_id", ForeignKey("a.id"))

b = Table("b", m, aid)

b.append_column(aid)

assert b.c.a_id.references(a.c.id)

eq_(len(b.constraints), 2)

def test_fk_construct(self):

c1 = Column("foo", Integer)

c2 = Column("bar", Integer)

m = MetaData()

t1 = Table("t", m, c1, c2)

fk1 = ForeignKeyConstraint(("foo",), ("bar",), table=t1)

assert fk1 in t1.constraints

def test_fk_constraint_col_collection_w_table(self):

c1 = Column("foo", Integer)

c2 = Column("bar", Integer)

m = MetaData()

t1 = Table("t", m, c1, c2)

fk1 = ForeignKeyConstraint(("foo",), ("bar",), table=t1)

eq_(dict(fk1.columns), {"foo": c1})

def test_fk_constraint_col_collection_no_table(self):

fk1 = ForeignKeyConstraint(("foo", "bat"), ("bar", "hoho"))

eq_(dict(fk1.columns), {})

eq_(fk1.column_keys, ["foo", "bat"])

eq_(fk1._col_description, "foo, bat")

eq_(fk1._elements, {"foo": fk1.elements[0], "bat": fk1.elements[1]})

def test_fk_constraint_col_collection_no_table_real_cols(self):

c1 = Column("foo", Integer)

c2 = Column("bar", Integer)

fk1 = ForeignKeyConstraint((c1,), (c2,))

eq_(dict(fk1.columns), {})

eq_(fk1.column_keys, ["foo"])

eq_(fk1._col_description, "foo")

eq_(fk1._elements, {"foo": fk1.elements[0]})

def test_fk_constraint_col_collection_added_to_table(self):

c1 = Column("foo", Integer)

m = MetaData()

fk1 = ForeignKeyConstraint(("foo",), ("bar",))

Table("t", m, c1, fk1)

eq_(dict(fk1.columns), {"foo": c1})

eq_(fk1._elements, {"foo": fk1.elements[0]})

def test_fk_constraint_col_collection_via_fk(self):

fk = ForeignKey("bar")

c1 = Column("foo", Integer, fk)

m = MetaData()

t1 = Table("t", m, c1)

fk1 = fk.constraint

eq_(fk1.column_keys, ["foo"])

assert fk1 in t1.constraints

eq_(fk1.column_keys, ["foo"])

eq_(dict(fk1.columns), {"foo": c1})

eq_(fk1._elements, {"foo": fk})

def test_fk_no_such_parent_col_error(self):

meta = MetaData()

a = Table("a", meta, Column("a", Integer))

Table("b", meta, Column("b", Integer))

def go():

a.append_constraint(ForeignKeyConstraint(["x"], ["b.b"]))

assert_raises_message(

exc.ArgumentError,

"Can't create ForeignKeyConstraint on "

"table 'a': no column named 'x' is present.",

go,

)

def test_fk_given_non_col(self):

not_a_col = bindparam("x")

assert_raises_message(

exc.ArgumentError,

"String column name or Column object for DDL foreign "

"key constraint expected, got .*Bind",

ForeignKey,

not_a_col,

)

def test_fk_given_non_col_clauseelem(self):

class Foo(object):

def __clause_element__(self):

return bindparam("x")

assert_raises_message(

exc.ArgumentError,

"String column name or Column object for DDL foreign "

"key constraint expected, got .*Foo",

ForeignKey,

Foo(),

)

def test_fk_given_col_non_table(self):

t = Table("t", MetaData(), Column("x", Integer))

xa = t.alias().c.x

assert_raises_message(

exc.ArgumentError,

"ForeignKey received Column not bound to a Table, got: .*Alias",

ForeignKey,

xa,

)

def test_fk_given_col_non_table_clauseelem(self):

t = Table("t", MetaData(), Column("x", Integer))

class Foo(object):

def __clause_element__(self):

return t.alias().c.x

assert_raises_message(

exc.ArgumentError,

"ForeignKey received Column not bound to a Table, got: .*Alias",

ForeignKey,

Foo(),

)

def test_fk_no_such_target_col_error_upfront(self):

meta = MetaData()

a = Table("a", meta, Column("a", Integer))

Table("b", meta, Column("b", Integer))

a.append_constraint(ForeignKeyConstraint(["a"], ["b.x"]))

assert_raises_message(

exc.NoReferencedColumnError,

"Could not initialize target column for ForeignKey 'b.x' on "

"table 'a': table 'b' has no column named 'x'",

getattr,

list(a.foreign_keys)[0],

"column",

)

def test_fk_no_such_target_col_error_delayed(self):

meta = MetaData()

a = Table("a", meta, Column("a", Integer))

a.append_constraint(ForeignKeyConstraint(["a"], ["b.x"]))

Table("b", meta, Column("b", Integer))

assert_raises_message(

exc.NoReferencedColumnError,

"Could not initialize target column for ForeignKey 'b.x' on "

"table 'a': table 'b' has no column named 'x'",

getattr,

list(a.foreign_keys)[0],

"column",

)

def test_fk_mismatched_local_remote_cols(self):

assert_raises_message(

exc.ArgumentError,

"ForeignKeyConstraint number of constrained columns must "

"match the number of referenced columns.",

ForeignKeyConstraint,

["a"],

["b.a", "b.b"],

)

assert_raises_message(

exc.ArgumentError,

"ForeignKeyConstraint number of constrained columns "

"must match the number of referenced columns.",

ForeignKeyConstraint,

["a", "b"],

["b.a"],

)

assert_raises_message(

exc.ArgumentError,

"ForeignKeyConstraint with duplicate source column "

"references are not supported.",

ForeignKeyConstraint,

["a", "a"],

["b.a", "b.b"],

)

def test_pickle_metadata_sequence_restated(self):

m1 = MetaData()

Table(

"a",

m1,

Column("id", Integer, primary_key=True),

Column("x", Integer, Sequence("x_seq")),

)

m2 = pickle.loads(pickle.dumps(m1))

s2 = Sequence("x_seq")

t2 = Table(

"a",

m2,

Column("id", Integer, primary_key=True),

Column("x", Integer, s2),

extend_existing=True,

)

assert m2._sequences["x_seq"] is t2.c.x.default

assert m2._sequences["x_seq"] is s2

def test_sequence_restated_replaced(self):

"""Test restatement of Sequence replaces."""

m1 = MetaData()

s1 = Sequence("x_seq")

t = Table("a", m1, Column("x", Integer, s1))

assert m1._sequences["x_seq"] is s1

s2 = Sequence("x_seq")

Table("a", m1, Column("x", Integer, s2), extend_existing=True)

assert t.c.x.default is s2

assert m1._sequences["x_seq"] is s2

def test_sequence_attach_to_table(self):

m1 = MetaData()

s1 = Sequence("s")

Table("a", m1, Column("x", Integer, s1))

assert s1.metadata is m1

def test_sequence_attach_to_existing_table(self):

m1 = MetaData()

s1 = Sequence("s")

t = Table("a", m1, Column("x", Integer))

t.c.x._init_items(s1)

assert s1.metadata is m1

def test_pickle_metadata_sequence_implicit(self):

m1 = MetaData()

Table(

"a",

m1,

Column("id", Integer, primary_key=True),

Column("x", Integer, Sequence("x_seq")),

)

m2 = pickle.loads(pickle.dumps(m1))

t2 = Table("a", m2, extend_existing=True)

eq_(m2._sequences, {"x_seq": t2.c.x.default})

def test_pickle_metadata_schema(self):

m1 = MetaData()

Table(

"a",

m1,

Column("id", Integer, primary_key=True),

Column("x", Integer, Sequence("x_seq")),

schema="y",

)

m2 = pickle.loads(pickle.dumps(m1))

Table("a", m2, schema="y", extend_existing=True)

eq_(m2._schemas, m1._schemas)

def test_metadata_schema_arg(self):

m1 = MetaData(schema="sch1")

m2 = MetaData(schema="sch1", quote_schema=True)

m3 = MetaData(schema="sch1", quote_schema=False)

m4 = MetaData()

for (

i,

(

name,

metadata,

schema_,

quote_schema,

exp_schema,

exp_quote_schema,

),

) in enumerate(

[

("t1", m1, None, None, "sch1", None),

("t2", m1, "sch2", None, "sch2", None),

("t3", m1, "sch2", True, "sch2", True),

("t4", m1, "sch1", None, "sch1", None),

("t5", m1, BLANK_SCHEMA, None, None, None),

("t1", m2, None, None, "sch1", True),

("t2", m2, "sch2", None, "sch2", None),

("t3", m2, "sch2", True, "sch2", True),

("t4", m2, "sch1", None, "sch1", None),

("t1", m3, None, None, "sch1", False),

("t2", m3, "sch2", None, "sch2", None),

("t3", m3, "sch2", True, "sch2", True),

("t4", m3, "sch1", None, "sch1", None),

("t1", m4, None, None, None, None),

("t2", m4, "sch2", None, "sch2", None),

("t3", m4, "sch2", True, "sch2", True),

("t4", m4, "sch1", None, "sch1", None),

("t5", m4, BLANK_SCHEMA, None, None, None),

]

):

kw = {}

if schema_ is not None:

kw["schema"] = schema_

if quote_schema is not None:

kw["quote_schema"] = quote_schema

t = Table(name, metadata, **kw)

eq_(t.schema, exp_schema, "test %d, table schema" % i)

eq_(

t.schema.quote if t.schema is not None else None,

exp_quote_schema,

"test %d, table quote_schema" % i,

)

seq = Sequence(name, metadata=metadata, **kw)

eq_(seq.schema, exp_schema, "test %d, seq schema" % i)

eq_(

seq.schema.quote if seq.schema is not None else None,

exp_quote_schema,

"test %d, seq quote_schema" % i,

)

def test_manual_dependencies(self):

meta = MetaData()

a = Table("a", meta, Column("foo", Integer))

b = Table("b", meta, Column("foo", Integer))

c = Table("c", meta, Column("foo", Integer))

d = Table("d", meta, Column("foo", Integer))

e = Table("e", meta, Column("foo", Integer))

e.add_is_dependent_on(c)

a.add_is_dependent_on(b)

b.add_is_dependent_on(d)

e.add_is_dependent_on(b)

c.add_is_dependent_on(a)

eq_(meta.sorted_tables, [d, b, a, c, e])

def test_deterministic_order(self):

meta = MetaData()

a = Table("a", meta, Column("foo", Integer))

b = Table("b", meta, Column("foo", Integer))

c = Table("c", meta, Column("foo", Integer))

d = Table("d", meta, Column("foo", Integer))

e = Table("e", meta, Column("foo", Integer))

e.add_is_dependent_on(c)

a.add_is_dependent_on(b)

eq_(meta.sorted_tables, [b, c, d, a, e])

def test_fks_deterministic_order(self):

meta = MetaData()

a = Table("a", meta, Column("foo", Integer, ForeignKey("b.foo")))

b = Table("b", meta, Column("foo", Integer))

c = Table("c", meta, Column("foo", Integer))

d = Table("d", meta, Column("foo", Integer))

e = Table("e", meta, Column("foo", Integer, ForeignKey("c.foo")))

eq_(meta.sorted_tables, [b, c, d, a, e])

def test_cycles_fks_warning_one(self):

meta = MetaData()

a = Table("a", meta, Column("foo", Integer, ForeignKey("b.foo")))

b = Table("b", meta, Column("foo", Integer, ForeignKey("d.foo")))

c = Table("c", meta, Column("foo", Integer, ForeignKey("b.foo")))

d = Table("d", meta, Column("foo", Integer, ForeignKey("c.foo")))

e = Table("e", meta, Column("foo", Integer))

with testing.expect_warnings(

"Cannot correctly sort tables; there are unresolvable cycles "

'between tables "b, c, d", which is usually caused by mutually '

"dependent foreign key constraints. "

"Foreign key constraints involving these tables will not be "

"considered"

):

eq_(meta.sorted_tables, [b, c, d, e, a])

def test_cycles_fks_warning_two(self):

meta = MetaData()

a = Table("a", meta, Column("foo", Integer, ForeignKey("b.foo")))

b = Table("b", meta, Column("foo", Integer, ForeignKey("a.foo")))

c = Table("c", meta, Column("foo", Integer, ForeignKey("e.foo")))

d = Table("d", meta, Column("foo", Integer))

e = Table("e", meta, Column("foo", Integer, ForeignKey("d.foo")))

with testing.expect_warnings(

"Cannot correctly sort tables; there are unresolvable cycles "

'between tables "a, b", which is usually caused by mutually '

"dependent foreign key constraints. "

"Foreign key constraints involving these tables will not be "

"considered"

):

eq_(meta.sorted_tables, [a, b, d, e, c])

def test_cycles_fks_fks_delivered_separately(self):

meta = MetaData()

a = Table("a", meta, Column("foo", Integer, ForeignKey("b.foo")))

b = Table("b", meta, Column("foo", Integer, ForeignKey("a.foo")))

c = Table("c", meta, Column("foo", Integer, ForeignKey("e.foo")))

d = Table("d", meta, Column("foo", Integer))

e = Table("e", meta, Column("foo", Integer, ForeignKey("d.foo")))

results = schema.sort_tables_and_constraints(

sorted(meta.tables.values(), key=lambda t: t.key)

)

results[-1] = (None, set(results[-1][-1]))

eq_(

results,

[

(a, set()),

(b, set()),

(d, {fk.constraint for fk in d.foreign_keys}),

(e, {fk.constraint for fk in e.foreign_keys}),

(c, {fk.constraint for fk in c.foreign_keys}),

(

None,

{fk.constraint for fk in a.foreign_keys}.union(

fk.constraint for fk in b.foreign_keys

),

),

],

)

def test_cycles_fks_usealter(self):

meta = MetaData()

a = Table("a", meta, Column("foo", Integer, ForeignKey("b.foo")))

b = Table(

"b",

meta,

Column("foo", Integer, ForeignKey("d.foo", use_alter=True)),

)

c = Table("c", meta, Column("foo", Integer, ForeignKey("b.foo")))

d = Table("d", meta, Column("foo", Integer, ForeignKey("c.foo")))

e = Table("e", meta, Column("foo", Integer))

eq_(meta.sorted_tables, [b, e, a, c, d])

def test_nonexistent(self):

assert_raises(

tsa.exc.NoSuchTableError,

Table,

"fake_table",

MetaData(testing.db),

autoload=True,

)

def test_assorted_repr(self):

t1 = Table("foo", MetaData(), Column("x", Integer))

i1 = Index("bar", t1.c.x)

ck = schema.CheckConstraint("x > y", name="someconstraint")

for const, exp in (

(Sequence("my_seq"), "Sequence('my_seq')"),

(Sequence("my_seq", start=5), "Sequence('my_seq', start=5)"),

(Column("foo", Integer), "Column('foo', Integer(), table=None)"),

(

Column(

"foo",

Integer,

primary_key=True,

nullable=False,

onupdate=1,

default=42,

server_default="42",

comment="foo",

),

"Column('foo', Integer(), table=None, primary_key=True, "

"nullable=False, onupdate=%s, default=%s, server_default=%s, "

"comment='foo')"

% (ColumnDefault(1), ColumnDefault(42), DefaultClause("42"),),

),

(

Table("bar", MetaData(), Column("x", String)),

"Table('bar', MetaData(bind=None), "

"Column('x', String(), table=<bar>), schema=None)",

),

(

schema.DefaultGenerator(for_update=True),

"DefaultGenerator(for_update=True)",

),

(schema.Index("bar", "c"), "Index('bar', 'c')"),

(i1, "Index('bar', Column('x', Integer(), table=<foo>))"),

(schema.FetchedValue(), "FetchedValue()"),

(

ck,

"CheckConstraint("

"%s"

", name='someconstraint')" % repr(ck.sqltext),

),

(ColumnDefault(("foo", "bar")), "ColumnDefault(('foo', 'bar'))"),

):

eq_(repr(const), exp)

class ToMetaDataTest(fixtures.TestBase, ComparesTables):

@testing.requires.check_constraints

def test_copy(self):

# TODO: modernize this test for 2.0

from sqlalchemy.testing.schema import Table

meta = MetaData()

table = Table(

"mytable",

meta,

Column("myid", Integer, Sequence("foo_id_seq"), primary_key=True),

Column("name", String(40), nullable=True),

Column(

"foo",

String(40),

nullable=False,

server_default="x",

server_onupdate="q",

),

Column(

"bar", String(40), nullable=False, default="y", onupdate="z"

),

Column(

"description", String(30), CheckConstraint("description='hi'")

),

UniqueConstraint("name"),

test_needs_fk=True,

)

table2 = Table(

"othertable",

meta,

Column("id", Integer, Sequence("foo_seq"), primary_key=True),

Column("myid", Integer, ForeignKey("mytable.myid")),

test_needs_fk=True,

)

table3 = Table(

"has_comments",

meta,

Column("foo", Integer, comment="some column"),

comment="table comment",

)

def test_to_metadata():

meta2 = MetaData()

table_c = table.tometadata(meta2)

table2_c = table2.tometadata(meta2)

table3_c = table3.tometadata(meta2)

return (table_c, table2_c, table3_c)

def test_pickle():

meta.bind = testing.db

meta2 = pickle.loads(pickle.dumps(meta))

assert meta2.bind is None

pickle.loads(pickle.dumps(meta2))

return (

meta2.tables["mytable"],

meta2.tables["othertable"],

meta2.tables["has_comments"],

)

def test_pickle_via_reflect():

# this is the most common use case, pickling the results of a

# database reflection

meta2 = MetaData(bind=testing.db)

t1 = Table("mytable", meta2, autoload=True)

Table("othertable", meta2, autoload=True)

Table("has_comments", meta2, autoload=True)

meta3 = pickle.loads(pickle.dumps(meta2))

assert meta3.bind is None

assert meta3.tables["mytable"] is not t1

return (

meta3.tables["mytable"],

meta3.tables["othertable"],

meta3.tables["has_comments"],

)

meta.create_all(testing.db)

try:

for test, has_constraints, reflect in (

(test_to_metadata, True, False),

(test_pickle, True, False),

(test_pickle_via_reflect, False, True),

):

table_c, table2_c, table3_c = test()

self.assert_tables_equal(table, table_c)

self.assert_tables_equal(table2, table2_c)

assert table is not table_c

assert table.primary_key is not table_c.primary_key

assert (

list(table2_c.c.myid.foreign_keys)[0].column

is table_c.c.myid

)

assert (

list(table2_c.c.myid.foreign_keys)[0].column

is not table.c.myid

)

assert "x" in str(table_c.c.foo.server_default.arg)

if not reflect:

assert isinstance(table_c.c.myid.default, Sequence)

assert str(table_c.c.foo.server_onupdate.arg) == "q"

assert str(table_c.c.bar.default.arg) == "y"

assert (

getattr(

table_c.c.bar.onupdate.arg,

"arg",

table_c.c.bar.onupdate.arg,

)

== "z"

)

assert isinstance(table2_c.c.id.default, Sequence)

# constraints don't get reflected for any dialect right

# now

if has_constraints:

for c in table_c.c.description.constraints:

if isinstance(c, CheckConstraint):

break

else:

assert False

assert str(c.sqltext) == "description='hi'"

for c in table_c.constraints:

if isinstance(c, UniqueConstraint):

break

else:

assert False

assert c.columns.contains_column(table_c.c.name)

assert not c.columns.contains_column(table.c.name)

if testing.requires.comment_reflection.enabled:

eq_(table3_c.comment, "table comment")

eq_(table3_c.c.foo.comment, "some column")

finally:

meta.drop_all(testing.db)

def test_col_key_fk_parent(self):

# test #2643

m1 = MetaData()

a = Table("a", m1, Column("x", Integer))

b = Table("b", m1, Column("x", Integer, ForeignKey("a.x"), key="y"))

assert b.c.y.references(a.c.x)

m2 = MetaData()

b2 = b.tometadata(m2)

a2 = a.tometadata(m2)

assert b2.c.y.references(a2.c.x)

def test_change_schema(self):

meta = MetaData()

table = Table(

"mytable",

meta,

Column("myid", Integer, primary_key=True),

Column("name", String(40), nullable=True),

Column(

"description", String(30), CheckConstraint("description='hi'")

),

UniqueConstraint("name"),

)

table2 = Table(

"othertable",

meta,

Column("id", Integer, primary_key=True),

Column("myid", Integer, ForeignKey("mytable.myid")),

)

meta2 = MetaData()

table_c = table.tometadata(meta2, schema="someschema")

table2_c = table2.tometadata(meta2, schema="someschema")

eq_(

str(table_c.join(table2_c).onclause),

str(table_c.c.myid == table2_c.c.myid),

)

eq_(

str(table_c.join(table2_c).onclause),

"someschema.mytable.myid = someschema.othertable.myid",

)

def test_retain_table_schema(self):

meta = MetaData()

table = Table(

"mytable",

meta,

Column("myid", Integer, primary_key=True),

Column("name", String(40), nullable=True),

Column(

"description", String(30), CheckConstraint("description='hi'")

),

UniqueConstraint("name"),

schema="myschema",

)

table2 = Table(

"othertable",

meta,

Column("id", Integer, primary_key=True),

Column("myid", Integer, ForeignKey("myschema.mytable.myid")),

schema="myschema",

)

meta2 = MetaData()

table_c = table.tometadata(meta2)

table2_c = table2.tometadata(meta2)

eq_(

str(table_c.join(table2_c).onclause),

str(table_c.c.myid == table2_c.c.myid),

)

eq_(

str(table_c.join(table2_c).onclause),

"myschema.mytable.myid = myschema.othertable.myid",

)

def test_change_name_retain_metadata(self):

meta = MetaData()

table = Table(

"mytable",

meta,

Column("myid", Integer, primary_key=True),

Column("name", String(40), nullable=True),

Column(

"description", String(30), CheckConstraint("description='hi'")

),

UniqueConstraint("name"),

schema="myschema",

)

table2 = table.tometadata(table.metadata, name="newtable")

table3 = table.tometadata(

table.metadata, schema="newschema", name="newtable"

)