Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with
or
.
Download ZIP
Browse files

Fixed #17574 -- Implemented missing get_key_columns in PostgreSQL bac…

…kend
  • Loading branch information...
commit 0171ba65dbbff377282c03b86c83036168c84b22 1 parent 223fc8e
@claudep claudep authored
View
17 django/db/backends/postgresql_psycopg2/introspection.py
@@ -66,6 +66,23 @@ def get_relations(self, cursor, table_name):
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
View
36 tests/regressiontests/introspection/tests.py
@@ -1,10 +1,8 @@
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
@@ -14,36 +12,7 @@
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))
@@ -139,6 +108,7 @@ def test_get_relations(self):
# 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)
Please sign in to comment.
Something went wrong with that request. Please try again.