The :mod:`~sqlalchemy.dialects.postgresql.psycopg2` dialect also offers the special level ``AUTOCOMMIT``. See
:ref:`psycopg2_isolation_level` for details.
-
-Remote / Cross-Schema Table Introspection
------------------------------------------
-
-Tables can be introspected from any accessible schema, including
-inter-schema foreign key relationships. However, care must be taken
-when specifying the "schema" argument for a given :class:`.Table`, when
-the given schema is also present in PostgreSQL's ``search_path`` variable
-for the current connection.
-
-If a FOREIGN KEY constraint reports that the remote table's schema is within
-the current ``search_path``, the "schema" attribute of the resulting
-:class:`.Table` will be set to ``None``, unless the actual schema of the
-remote table matches that of the referencing table, and the "schema" argument
-was explicitly stated on the referencing table.
-
-The best practice here is to not use the ``schema`` argument
-on :class:`.Table` for any schemas that are present in ``search_path``.
-``search_path`` defaults to "public", but care should be taken
-to inspect the actual value using::
-
- SHOW search_path;
-
-.. versionchanged:: 0.7.3
- Prior to this version, cross-schema foreign keys when the schemas
- were also in the ``search_path`` could make an incorrect assumption
- if the schemas were explicitly stated on each :class:`.Table`.
-
-Background on PG's ``search_path`` is at:
-http://www.postgresql.org/docs/9.0/static/ddl-schemas.html#DDL-SCHEMAS-PATH
+.. _postgresql_schema_reflection:
+
+Remote-Schema Table Introspection and Postgresql search_path
+------------------------------------------------------------
+
+The Postgresql dialect can reflect tables from any schema. The
+:paramref:`.Table.schema` argument, or alternatively the
+:paramref:`.MetaData.reflect.schema` argument determines which schema will
+be searched for the table or tables. The reflected :class:`.Table` objects
+will in all cases retain this ``.schema`` attribute as was specified. However,
+with regards to tables which these :class:`.Table` objects refer to via
+foreign key constraint, a decision must be made as to how the ``.schema``
+is represented in those remote tables, in the case where that remote
+schema name is also a member of the current
+`Postgresql search path <http://www.postgresql.org/docs/9.0/static/ddl-schemas.html#DDL-SCHEMAS-PATH>`_.
+
+By default, the Postgresql dialect mimics the behavior encouraged by
+Postgresql's own ``pg_get_constraintdef()`` builtin procedure. This function
+returns a sample definition for a particular foreign key constraint,
+omitting the referenced schema name from that definition when the name is
+also in the Postgresql schema search path. The interaction below
+illustrates this behavior::
+
+ test=> CREATE TABLE test_schema.referred(id INTEGER PRIMARY KEY);
+ CREATE TABLE
+ test=> CREATE TABLE referring(
+ test(> id INTEGER PRIMARY KEY,
+ test(> referred_id INTEGER REFERENCES test_schema.referred(id));
+ CREATE TABLE
+ test=> SET search_path TO public, test_schema;
+ test=> SELECT pg_catalog.pg_get_constraintdef(r.oid, true) FROM
+ test-> pg_catalog.pg_class c JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
+ test-> JOIN pg_catalog.pg_constraint r ON c.oid = r.conrelid
+ test-> WHERE c.relname='referring' AND r.contype = 'f'
+ test-> ;
+ pg_get_constraintdef
+ ---------------------------------------------------
+ FOREIGN KEY (referred_id) REFERENCES referred(id)
+ (1 row)
+
+Above, we created a table ``referred`` as a member of the remote schema ``test_schema``, however
+when we added ``test_schema`` to the PG ``search_path`` and then asked ``pg_get_constraintdef()``
+for the ``FOREIGN KEY`` syntax, ``test_schema`` was not included in the
+output of the function.
+
+On the other hand, if we set the search path back to the typical default
+of ``public``::
+
+ test=> SET search_path TO public;
+ SET
+
+The same query against ``pg_get_constraintdef()`` now returns the fully
+schema-qualified name for us::
+
+ test=> SELECT pg_catalog.pg_get_constraintdef(r.oid, true) FROM
+ test-> pg_catalog.pg_class c JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
+ test-> JOIN pg_catalog.pg_constraint r ON c.oid = r.conrelid
+ test-> WHERE c.relname='referring' AND r.contype = 'f';
+ pg_get_constraintdef
+ ---------------------------------------------------------------
+ FOREIGN KEY (referred_id) REFERENCES test_schema.referred(id)
+ (1 row)
+
+SQLAlchemy will by default use the return value of ``pg_get_constraintdef()``
+in order to determine the remote schema name. That is, if our ``search_path``
+were set to include ``test_schema``, and we invoked a table
+reflection process as follows::
+
+ >>> from sqlalchemy import Table, MetaData, create_engine
+ >>> engine = create_engine("postgresql://scott:tiger@localhost/test")
+ >>> with engine.connect() as conn:
+ ... conn.execute("SET search_path TO test_schema, public")
+ ... meta = MetaData()
+ ... referring = Table('referring', meta, autoload=True, autoload_with=conn)
+ ...
+ <sqlalchemy.engine.result.ResultProxy object at 0x101612ed0>
+
+The above process would deliver to the :attr:`.MetaData.tables` collection
+``referred`` table named **without** the schema::
+
+ >>> meta.tables['referred'].schema is None
+ True
+
+To alter the behavior of reflection such that the referred schema is maintained
+regardless of the ``search_path`` setting, use the ``postgresql_ignore_search_path``
+option, which can be specified as a dialect-specific argument to both
+:class:`.Table` as well as :meth:`.MetaData.reflect`::
+
+ >>> with engine.connect() as conn:
+ ... conn.execute("SET search_path TO test_schema, public")
+ ... meta = MetaData()
+ ... referring = Table('referring', meta, autoload=True, autoload_with=conn,
+ ... postgresql_ignore_search_path=True)
+ ...
+ <sqlalchemy.engine.result.ResultProxy object at 0x1016126d0>
+
+We will now have ``test_schema.referred`` stored as schema-qualified::
+
+ >>> meta.tables['test_schema.referred'].schema
+ 'test_schema'
+
+.. sidebar:: Best Practices for Postgresql Schema reflection
+
+ The description of Postgresql schema reflection behavior is complex, and is
+ the product of many years of dealing with widely varied use cases and user preferences.
+ But in fact, there's no need to understand any of it if you just stick to the simplest
+ use pattern: leave the ``search_path`` set to its default of ``public`` only, never refer
+ to the name ``public`` as an explicit schema name otherwise, and
+ refer to all other schema names explicitly when building
+ up a :class:`.Table` object. The options described here are only for those users
+ who can't, or prefer not to, stay within these guidelines.
+
+Note that **in all cases**, the "default" schema is always reflected as ``None``.
+The "default" schema on Postgresql is that which is returned by the
+Postgresql ``current_schema()`` function. On a typical Postgresql installation,
+this is the name ``public``. So a table that refers to another which is
+in the ``public`` (i.e. default) schema will always have the ``.schema`` attribute
+set to ``None``.
+
+.. versionadded:: 0.9.2 Added the ``postgresql_ignore_search_path``
+ dialect-level option accepted by :class:`.Table` and :meth:`.MetaData.reflect`.
+
+
+.. seealso::
+
+ `The Schema Search Path <http://www.postgresql.org/docs/9.0/static/ddl-schemas.html#DDL-SCHEMAS-PATH>`_ - on the Postgresql website.
INSERT/UPDATE...RETURNING
-------------------------
"using": False,
"where": None,
"ops": {}
+ }),
+ (schema.Table, {
+ "ignore_search_path": False
})
]
+ reflection_options = ('postgresql_ignore_search_path', )
+
_backslash_escapes = True
def __init__(self, isolation_level=None, json_serializer=None,
return {'constrained_columns': cols, 'name': name}
@reflection.cache
- def get_foreign_keys(self, connection, table_name, schema=None, **kw):
+ def get_foreign_keys(self, connection, table_name, schema=None,
+ postgresql_ignore_search_path=False, **kw):
preparer = self.identifier_preparer
table_oid = self.get_table_oid(connection, table_name, schema,
info_cache=kw.get('info_cache'))
fkeys = []
for conname, condef, conschema in c.fetchall():
m = re.search(FK_REGEX, condef).groups()
+
constrained_columns, referred_schema, \
referred_table, referred_columns, \
_, match, _, onupdate, _, ondelete, \
deferrable, _, initially = m
+
if deferrable is not None:
deferrable = True if deferrable == 'DEFERRABLE' else False
constrained_columns = [preparer._unquote_identifier(x)
for x in re.split(r'\s*,\s*', constrained_columns)]
- if referred_schema:
+ if postgresql_ignore_search_path:
+ # when ignoring search path, we use the actual schema
+ # provided it isn't the "default" schema
+ if conschema != self.default_schema_name:
+ referred_schema = conschema
+ else:
+ referred_schema = schema
+ elif referred_schema:
+ # referred_schema is the schema that we regexp'ed from
+ # pg_get_constraintdef(). If the schema is in the search
+ # path, pg_get_constraintdef() will give us None.
referred_schema = \
preparer._unquote_identifier(referred_schema)
elif schema is not None and schema == conschema:
- # no schema was returned by pg_get_constraintdef(). This
- # means the schema is in the search path. We will leave
- # it as None, unless the actual schema, which we pull out
- # from pg_namespace even though pg_get_constraintdef() doesn't
- # want to give it to us, matches that of the referencing table,
- # and an explicit schema was given for the referencing table.
+ # If the actual schema matches the schema of the table
+ # we're reflecting, then we will use that.
referred_schema = schema
+
referred_table = preparer._unquote_identifier(referred_table)
referred_columns = [preparer._unquote_identifier(x)
for x in re.split(r'\s*,\s', referred_columns)]
eq_([c.name for c in t2.primary_key], ['t_id'])
@testing.provide_metadata
- def test_schema_reflection(self):
- """note: this test requires that the 'test_schema' schema be
- separate and accessible by the test user"""
+ def test_cross_schema_reflection_one(self):
meta1 = self.metadata
- users = Table('users', meta1, Column('user_id', Integer,
- primary_key=True), Column('user_name',
- String(30), nullable=False), schema='test_schema')
- addresses = Table(
- 'email_addresses',
- meta1,
+ users = Table('users', meta1,
+ Column('user_id', Integer, primary_key=True),
+ Column('user_name', String(30), nullable=False),
+ schema='test_schema')
+ addresses = Table('email_addresses', meta1,
Column('address_id', Integer, primary_key=True),
- Column('remote_user_id', Integer,
- ForeignKey(users.c.user_id)),
+ Column('remote_user_id', Integer, ForeignKey(users.c.user_id)),
Column('email_address', String(20)),
- schema='test_schema',
- )
+ schema='test_schema')
meta1.create_all()
meta2 = MetaData(testing.db)
addresses = Table('email_addresses', meta2, autoload=True,
== addresses.c.remote_user_id).compare(j.onclause))
@testing.provide_metadata
- def test_schema_reflection_2(self):
+ def test_cross_schema_reflection_two(self):
meta1 = self.metadata
- subject = Table('subject', meta1, Column('id', Integer,
- primary_key=True))
- referer = Table('referer', meta1, Column('id', Integer,
- primary_key=True), Column('ref', Integer,
- ForeignKey('subject.id')), schema='test_schema')
+ subject = Table('subject', meta1,
+ Column('id', Integer, primary_key=True))
+ referer = Table('referer', meta1,
+ Column('id', Integer, primary_key=True),
+ Column('ref', Integer, ForeignKey('subject.id')),
+ schema='test_schema')
meta1.create_all()
meta2 = MetaData(testing.db)
subject = Table('subject', meta2, autoload=True)
subject.join(referer).onclause))
@testing.provide_metadata
- def test_schema_reflection_3(self):
+ def test_cross_schema_reflection_three(self):
meta1 = self.metadata
- subject = Table('subject', meta1, Column('id', Integer,
- primary_key=True), schema='test_schema_2')
- referer = Table('referer', meta1, Column('id', Integer,
- primary_key=True), Column('ref', Integer,
- ForeignKey('test_schema_2.subject.id')),
+ subject = Table('subject', meta1,
+ Column('id', Integer, primary_key=True),
+ schema='test_schema_2')
+ referer = Table('referer', meta1,
+ Column('id', Integer, primary_key=True),
+ Column('ref', Integer, ForeignKey('test_schema_2.subject.id')),
schema='test_schema')
meta1.create_all()
meta2 = MetaData(testing.db)
subject = Table('subject', meta2, autoload=True,
schema='test_schema_2')
- referer = Table('referer', meta2, schema='test_schema',
- autoload=True)
+ referer = Table('referer', meta2, autoload=True,
+ schema='test_schema')
self.assert_((subject.c.id
== referer.c.ref).compare(
subject.join(referer).onclause))
@testing.provide_metadata
- def test_uppercase_lowercase_table(self):
- metadata = self.metadata
-
- a_table = Table('a', metadata, Column('x', Integer))
- A_table = Table('A', metadata, Column('x', Integer))
-
- a_table.create()
- assert testing.db.has_table("a")
- assert not testing.db.has_table("A")
- A_table.create(checkfirst=True)
- assert testing.db.has_table("A")
-
- def test_uppercase_lowercase_sequence(self):
+ def test_cross_schema_reflection_four(self):
+ meta1 = self.metadata
+ subject = Table('subject', meta1,
+ Column('id', Integer, primary_key=True),
+ schema='test_schema_2')
+ referer = Table('referer', meta1,
+ Column('id', Integer, primary_key=True),
+ Column('ref', Integer, ForeignKey('test_schema_2.subject.id')),
+ schema='test_schema')
+ meta1.create_all()
- a_seq = Sequence('a')
- A_seq = Sequence('A')
+ conn = testing.db.connect()
+ conn.detach()
+ conn.execute("SET search_path TO test_schema, test_schema_2")
+ meta2 = MetaData(bind=conn)
+ subject = Table('subject', meta2, autoload=True,
+ schema='test_schema_2',
+ postgresql_ignore_search_path=True)
+ referer = Table('referer', meta2, autoload=True,
+ schema='test_schema',
+ postgresql_ignore_search_path=True)
+ self.assert_((subject.c.id
+ == referer.c.ref).compare(
+ subject.join(referer).onclause))
+ conn.close()
- a_seq.create(testing.db)
- assert testing.db.dialect.has_sequence(testing.db, "a")
- assert not testing.db.dialect.has_sequence(testing.db, "A")
- A_seq.create(testing.db, checkfirst=True)
- assert testing.db.dialect.has_sequence(testing.db, "A")
+ @testing.provide_metadata
+ def test_cross_schema_reflection_five(self):
+ meta1 = self.metadata
- a_seq.drop(testing.db)
- A_seq.drop(testing.db)
+ # we assume 'public'
+ default_schema = testing.db.dialect.default_schema_name
+ subject = Table('subject', meta1,
+ Column('id', Integer, primary_key=True))
+ referer = Table('referer', meta1,
+ Column('id', Integer, primary_key=True),
+ Column('ref', Integer, ForeignKey('subject.id')))
+ meta1.create_all()
- def test_schema_reflection_multi_search_path(self):
- """test the 'set the same schema' rule when
- multiple schemas/search paths are in effect."""
+ meta2 = MetaData(testing.db)
+ subject = Table('subject', meta2, autoload=True,
+ schema=default_schema,
+ postgresql_ignore_search_path=True
+ )
+ referer = Table('referer', meta2, autoload=True,
+ schema=default_schema,
+ postgresql_ignore_search_path=True
+ )
+ assert subject.schema == default_schema
+ self.assert_((subject.c.id
+ == referer.c.ref).compare(
+ subject.join(referer).onclause))
- db = engines.testing_engine()
- conn = db.connect()
- trans = conn.begin()
- try:
- conn.execute("set search_path to test_schema_2, "
- "test_schema, public")
- conn.dialect.default_schema_name = "test_schema_2"
+ @testing.provide_metadata
+ def test_cross_schema_reflection_six(self):
+ # test that the search path *is* taken into account
+ # by default
+ meta1 = self.metadata
- conn.execute("""
- CREATE TABLE test_schema.some_table (
- id SERIAL not null primary key
+ Table('some_table', meta1,
+ Column('id', Integer, primary_key=True),
+ schema='test_schema'
)
- """)
+ Table('some_other_table', meta1,
+ Column('id', Integer, primary_key=True),
+ Column('sid', Integer, ForeignKey('test_schema.some_table.id')),
+ schema='test_schema_2'
+ )
+ meta1.create_all()
+ with testing.db.connect() as conn:
+ conn.detach()
- conn.execute("""
- CREATE TABLE test_schema_2.some_other_table (
- id SERIAL not null primary key,
- sid INTEGER REFERENCES test_schema.some_table(id)
- )
- """)
+ conn.execute("set search_path to test_schema_2, test_schema, public")
- m1 = MetaData()
+ m1 = MetaData(conn)
- t2_schema = Table('some_other_table',
- m1,
- schema="test_schema_2",
- autoload=True,
- autoload_with=conn)
t1_schema = Table('some_table',
m1,
schema="test_schema",
- autoload=True,
- autoload_with=conn)
+ autoload=True)
+ t2_schema = Table('some_other_table',
+ m1,
+ schema="test_schema_2",
+ autoload=True)
t2_no_schema = Table('some_other_table',
m1,
- autoload=True,
- autoload_with=conn)
+ autoload=True)
t1_no_schema = Table('some_table',
m1,
+ autoload=True)
+
+ m2 = MetaData(conn)
+ t1_schema_isp = Table('some_table',
+ m2,
+ schema="test_schema",
autoload=True,
- autoload_with=conn)
+ postgresql_ignore_search_path=True)
+ t2_schema_isp = Table('some_other_table',
+ m2,
+ schema="test_schema_2",
+ autoload=True,
+ postgresql_ignore_search_path=True)
+
- # OK, this because, "test_schema" is
- # in the search path, and might as well be
- # the default too. why would we assign
- # a "schema" to the Table ?
+ # t2_schema refers to t1_schema, but since "test_schema"
+ # is in the search path, we instead link to t2_no_schema
assert t2_schema.c.sid.references(
t1_no_schema.c.id)
+ # the two no_schema tables refer to each other also.
assert t2_no_schema.c.sid.references(
t1_no_schema.c.id)
- finally:
- trans.rollback()
- conn.close()
- db.dispose()
+ # but if we're ignoring search path, then we maintain
+ # those explicit schemas vs. what the "default" schema is
+ assert t2_schema_isp.c.sid.references(t1_schema_isp.c.id)
+
+ @testing.provide_metadata
+ def test_cross_schema_reflection_seven(self):
+ # test that the search path *is* taken into account
+ # by default
+ meta1 = self.metadata
+
+ Table('some_table', meta1,
+ Column('id', Integer, primary_key=True),
+ schema='test_schema'
+ )
+ Table('some_other_table', meta1,
+ Column('id', Integer, primary_key=True),
+ Column('sid', Integer, ForeignKey('test_schema.some_table.id')),
+ schema='test_schema_2'
+ )
+ meta1.create_all()
+ with testing.db.connect() as conn:
+ conn.detach()
+
+ conn.execute("set search_path to test_schema_2, test_schema, public")
+ meta2 = MetaData(conn)
+ meta2.reflect(schema="test_schema_2")
+
+ eq_(set(meta2.tables), set(['test_schema_2.some_other_table', 'some_table']))
+
+ meta3 = MetaData(conn)
+ meta3.reflect(schema="test_schema_2", postgresql_ignore_search_path=True)
+
+ eq_(set(meta3.tables),
+ set(['test_schema_2.some_other_table', 'test_schema.some_table']))
+
+
+ @testing.provide_metadata
+ def test_uppercase_lowercase_table(self):
+ metadata = self.metadata
+
+ a_table = Table('a', metadata, Column('x', Integer))
+ A_table = Table('A', metadata, Column('x', Integer))
+
+ a_table.create()
+ assert testing.db.has_table("a")
+ assert not testing.db.has_table("A")
+ A_table.create(checkfirst=True)
+ assert testing.db.has_table("A")
+
+ def test_uppercase_lowercase_sequence(self):
+
+ a_seq = Sequence('a')
+ A_seq = Sequence('A')
+
+ a_seq.create(testing.db)
+ assert testing.db.dialect.has_sequence(testing.db, "a")
+ assert not testing.db.dialect.has_sequence(testing.db, "A")
+ A_seq.create(testing.db, checkfirst=True)
+ assert testing.db.dialect.has_sequence(testing.db, "A")
+
+ a_seq.drop(testing.db)
+ A_seq.drop(testing.db)
+
@testing.provide_metadata
def test_index_reflection(self):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
