Loading django/db/backends/postgresql_psycopg2/introspection.py +17 −0 Original line number Diff line number Diff line Loading @@ -66,6 +66,23 @@ class DatabaseIntrospection(BaseDatabaseIntrospection): relations[row[0][0] - 1] = (row[1][0] - 1, row[2]) return relations def get_key_columns(self, cursor, table_name): key_columns = [] cursor.execute(""" SELECT kcu.column_name, ccu.table_name AS referenced_table, ccu.column_name AS referenced_column FROM information_schema.constraint_column_usage ccu LEFT JOIN information_schema.key_column_usage kcu ON ccu.constraint_catalog = kcu.constraint_catalog AND ccu.constraint_schema = kcu.constraint_schema AND ccu.constraint_name = kcu.constraint_name LEFT JOIN information_schema.table_constraints tc ON ccu.constraint_catalog = tc.constraint_catalog AND ccu.constraint_schema = tc.constraint_schema AND ccu.constraint_name = tc.constraint_name WHERE kcu.table_name = %s AND tc.constraint_type = 'FOREIGN KEY'""" , [table_name]) key_columns.extend(cursor.fetchall()) return key_columns def get_indexes(self, cursor, table_name): # This query retrieves each index on the given table, including the # first associated field name Loading tests/regressiontests/introspection/tests.py +3 −33 Original line number Diff line number Diff line from __future__ import absolute_import, unicode_literals from functools import update_wrapper from django.db import connection from django.test import TestCase, skipUnlessDBFeature, skipIfDBFeature from django.utils import six, unittest from django.utils import unittest from .models import Reporter, Article Loading @@ -14,36 +12,7 @@ else: expectedFailureOnOracle = lambda f: f # The introspection module is optional, so methods tested here might raise # NotImplementedError. This is perfectly acceptable behavior for the backend # in question, but the tests need to handle this without failing. Ideally we'd # skip these tests, but until #4788 is done we'll just ignore them. # # The easiest way to accomplish this is to decorate every test case with a # wrapper that ignores the exception. # # The metaclass is just for fun. def ignore_not_implemented(func): def _inner(*args, **kwargs): try: return func(*args, **kwargs) except NotImplementedError: return None update_wrapper(_inner, func) return _inner class IgnoreNotimplementedError(type): def __new__(cls, name, bases, attrs): for k, v in attrs.items(): if k.startswith('test'): attrs[k] = ignore_not_implemented(v) return type.__new__(cls, name, bases, attrs) class IntrospectionTests(six.with_metaclass(IgnoreNotimplementedError, TestCase)): class IntrospectionTests(TestCase): def test_table_names(self): tl = connection.introspection.table_names() self.assertEqual(tl, sorted(tl)) Loading Loading @@ -139,6 +108,7 @@ class IntrospectionTests(six.with_metaclass(IgnoreNotimplementedError, TestCase) # That's {field_index: (field_index_other_table, other_table)} self.assertEqual(relations, {3: (0, Reporter._meta.db_table)}) @skipUnlessDBFeature('can_introspect_foreign_keys') def test_get_key_columns(self): cursor = connection.cursor() key_columns = connection.introspection.get_key_columns(cursor, Article._meta.db_table) Loading Loading
django/db/backends/postgresql_psycopg2/introspection.py +17 −0 Original line number Diff line number Diff line Loading @@ -66,6 +66,23 @@ class DatabaseIntrospection(BaseDatabaseIntrospection): relations[row[0][0] - 1] = (row[1][0] - 1, row[2]) return relations def get_key_columns(self, cursor, table_name): key_columns = [] cursor.execute(""" SELECT kcu.column_name, ccu.table_name AS referenced_table, ccu.column_name AS referenced_column FROM information_schema.constraint_column_usage ccu LEFT JOIN information_schema.key_column_usage kcu ON ccu.constraint_catalog = kcu.constraint_catalog AND ccu.constraint_schema = kcu.constraint_schema AND ccu.constraint_name = kcu.constraint_name LEFT JOIN information_schema.table_constraints tc ON ccu.constraint_catalog = tc.constraint_catalog AND ccu.constraint_schema = tc.constraint_schema AND ccu.constraint_name = tc.constraint_name WHERE kcu.table_name = %s AND tc.constraint_type = 'FOREIGN KEY'""" , [table_name]) key_columns.extend(cursor.fetchall()) return key_columns def get_indexes(self, cursor, table_name): # This query retrieves each index on the given table, including the # first associated field name Loading
tests/regressiontests/introspection/tests.py +3 −33 Original line number Diff line number Diff line from __future__ import absolute_import, unicode_literals from functools import update_wrapper from django.db import connection from django.test import TestCase, skipUnlessDBFeature, skipIfDBFeature from django.utils import six, unittest from django.utils import unittest from .models import Reporter, Article Loading @@ -14,36 +12,7 @@ else: expectedFailureOnOracle = lambda f: f # The introspection module is optional, so methods tested here might raise # NotImplementedError. This is perfectly acceptable behavior for the backend # in question, but the tests need to handle this without failing. Ideally we'd # skip these tests, but until #4788 is done we'll just ignore them. # # The easiest way to accomplish this is to decorate every test case with a # wrapper that ignores the exception. # # The metaclass is just for fun. def ignore_not_implemented(func): def _inner(*args, **kwargs): try: return func(*args, **kwargs) except NotImplementedError: return None update_wrapper(_inner, func) return _inner class IgnoreNotimplementedError(type): def __new__(cls, name, bases, attrs): for k, v in attrs.items(): if k.startswith('test'): attrs[k] = ignore_not_implemented(v) return type.__new__(cls, name, bases, attrs) class IntrospectionTests(six.with_metaclass(IgnoreNotimplementedError, TestCase)): class IntrospectionTests(TestCase): def test_table_names(self): tl = connection.introspection.table_names() self.assertEqual(tl, sorted(tl)) Loading Loading @@ -139,6 +108,7 @@ class IntrospectionTests(six.with_metaclass(IgnoreNotimplementedError, TestCase) # That's {field_index: (field_index_other_table, other_table)} self.assertEqual(relations, {3: (0, Reporter._meta.db_table)}) @skipUnlessDBFeature('can_introspect_foreign_keys') def test_get_key_columns(self): cursor = connection.cursor() key_columns = connection.introspection.get_key_columns(cursor, Article._meta.db_table) Loading