underlying CREATE INDEX command, so it *must* be a valid index type for your
version of PostgreSQL.
+Special Reflection Options
+--------------------------
+
+The :class:`.Inspector` used for the Postgresql backend is an instance
+of :class:`.PGInspector`, which offers additional methods::
+
+ from sqlalchemy import create_engine, inspect
+
+ engine = create_engine("postgresql+psycopg2://localhost/test")
+ insp = inspect(engine) # will be a PGInspector
+
+ print(insp.get_enums())
+
+.. autoclass:: PGInspector
+ :members:
+
+
"""
from collections import defaultdict
import re
reflection.Inspector.__init__(self, conn)
def get_table_oid(self, table_name, schema=None):
- """Return the oid from `table_name` and `schema`."""
+ """Return the OID for the given table name."""
return self.dialect.get_table_oid(self.bind, table_name, schema,
info_cache=self.info_cache)
- def load_enums(self, conn, schema=None):
- """Return a enums list.
- A per-database and per-schema enums list."""
+ def get_enums(self, schema=None):
+ """Return a list of ENUM objects.
+
+ Each member is a dictionary containing these fields:
+
+ * name - name of the enum
+ * schema - the schema name for the enum.
+ * visible - boolean, whether or not this enum is visible
+ in the default search path.
+ * labels - a list of string labels that apply to the enum.
+
+ :param schema: schema name. If None, the default schema
+ (typically 'public') is used. May also be set to '*' to
+ indicate load enums for all schemas.
+
+ .. versionadded:: 1.0.0
+
+ """
schema = schema or self.default_schema_name
- return self.dialect._load_enums(conn, schema)
+ return self.dialect._load_enums(self.bind, schema)
class CreateEnumType(schema._CreateDropBase):
c = connection.execute(s, table_oid=table_oid)
rows = c.fetchall()
domains = self._load_domains(connection)
- enums = self._load_enums(connection)
+ enums = dict(
+ (
+ "%s.%s" % (rec['schema'], rec['name'])
+ if not rec['visible'] else rec['name'], rec) for rec in
+ self._load_enums(connection, schema='*')
+ )
# format columns
columns = []
elif attype in enums:
enum = enums[attype]
coltype = ENUM
- if "." in attype:
- kwargs['schema'], kwargs['name'] = attype.split('.')
- else:
- kwargs['name'] = attype
+ kwargs['name'] = enum['name']
+ if not enum['visible']:
+ kwargs['schema'] = enum['schema']
args = tuple(enum['labels'])
break
elif attype in domains:
LEFT JOIN pg_catalog.pg_namespace n ON n.oid = t.typnamespace
LEFT JOIN pg_catalog.pg_enum e ON t.oid = e.enumtypid
WHERE t.typtype = 'e'
- AND n.nspname = '%s'
- ORDER BY "name", e.oid -- e.oid gives us label order
- """ % schema
+ """
+
+ if schema != '*':
+ SQL_ENUMS += "AND n.nspname = :schema "
+
+ # e.oid gives us label order within an enum
+ SQL_ENUMS += 'ORDER BY "schema", "name", e.oid'
s = sql.text(SQL_ENUMS, typemap={
'attname': sqltypes.Unicode,
'label': sqltypes.Unicode})
+
+ if schema != '*':
+ s = s.bindparams(schema=schema)
+
c = connection.execute(s)
- enums = {}
+ enums = []
+ enum_by_name = {}
for enum in c.fetchall():
- if enum['visible']:
- # 'visible' just means whether or not the enum is in a
- # schema that's on the search path -- or not overridden by
- # a schema with higher precedence. If it's not visible,
- # it will be prefixed with the schema-name when it's used.
- name = enum['name']
- else:
- name = "%s.%s" % (enum['schema'], enum['name'])
-
- if name in enums:
- enums[name]['labels'].append(enum['label'])
+ key = (enum['schema'], enum['name'])
+ if key in enum_by_name:
+ enum_by_name[key]['labels'].append(enum['label'])
else:
- enums[name] = {
+ enum_by_name[key] = enum_rec = {
+ 'name': enum['name'],
+ 'schema': enum['schema'],
+ 'visible': enum['visible'],
'labels': [enum['label']],
}
+ enums.append(enum_rec)
return enums
eq_(fk, fk_ref[fk['name']])
@testing.provide_metadata
- def test_inspect_enums_custom_schema(self):
+ def test_inspect_enums_schema(self):
conn = testing.db.connect()
- enum_type = postgresql.ENUM('sad', 'ok', 'happy', name='mood',
- metadata=self.metadata, schema='test_schema')
+ enum_type = postgresql.ENUM(
+ 'sad', 'ok', 'happy', name='mood',
+ schema='test_schema',
+ metadata=self.metadata)
enum_type.create(conn)
inspector = reflection.Inspector.from_engine(conn.engine)
- eq_(inspector.load_enums(conn, 'test_schema'), {
- u'test_schema.mood': {'labels': [u'sad', u'ok', u'happy']}})
+ eq_(
+ inspector.get_enums('test_schema'), [{
+ 'visible': False,
+ 'name': 'mood',
+ 'schema': 'test_schema',
+ 'labels': ['sad', 'ok', 'happy']
+ }])
@testing.provide_metadata
- def test_inspect_enums_schema(self):
- conn = testing.db.connect()
- enum_type = postgresql.ENUM('cat', 'dog', 'rat', name='pet',
+ def test_inspect_enums(self):
+ enum_type = postgresql.ENUM(
+ 'cat', 'dog', 'rat', name='pet', metadata=self.metadata)
+ enum_type.create(testing.db)
+ inspector = reflection.Inspector.from_engine(testing.db)
+ eq_(inspector.get_enums(), [
+ {
+ 'visible': True,
+ 'labels': ['cat', 'dog', 'rat'],
+ 'name': 'pet',
+ 'schema': 'public'
+ }])
+
+ @testing.provide_metadata
+ def test_inspect_enums_star(self):
+ enum_type = postgresql.ENUM(
+ 'cat', 'dog', 'rat', name='pet', metadata=self.metadata)
+ schema_enum_type = postgresql.ENUM(
+ 'sad', 'ok', 'happy', name='mood',
+ schema='test_schema',
metadata=self.metadata)
- enum_type.create(conn)
- inspector = reflection.Inspector.from_engine(conn.engine)
- eq_(inspector.load_enums(conn), {
- u'pet': {'labels': [u'cat', u'dog', u'rat']}})
+ enum_type.create(testing.db)
+ schema_enum_type.create(testing.db)
+ inspector = reflection.Inspector.from_engine(testing.db)
+
+ eq_(inspector.get_enums(), [
+ {
+ 'visible': True,
+ 'labels': ['cat', 'dog', 'rat'],
+ 'name': 'pet',
+ 'schema': 'public'
+ }])
+
+ eq_(inspector.get_enums('*'), [
+ {
+ 'visible': True,
+ 'labels': ['cat', 'dog', 'rat'],
+ 'name': 'pet',
+ 'schema': 'public'
+ },
+ {
+ 'visible': False,
+ 'name': 'mood',
+ 'schema': 'test_schema',
+ 'labels': ['sad', 'ok', 'happy']
+ }])
class CustomTypeReflectionTest(fixtures.TestBase):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
