10568469992

Committed 26 Aug 2024 10:49PM UTC coverage: 90.187%. Remained the same

Build # 10568469992

Build Type

push

github

Committed by

lorinkoz

Commit Message

Bumped version

Run Details

1158 of 1284 relevant lines covered (90.19%)

10.81 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

77.78

/django_pgschemas/postgresql/introspection.py

from django.db.backends.base.introspection import FieldInfo, TableInfo
from django.utils.encoding import force_str

from django_pgschemas.schema import get_current_schema
from django_pgschemas.settings import get_base_backend_module, get_original_backend_module

from . import _constraints

try:
    module = get_original_backend_module("introspection")
except AttributeError:
    module = get_base_backend_module("introspection")


class DatabaseSchemaIntrospection(module.DatabaseIntrospection):  # pragma: no cover
    """
    database schema introspection class
    """

    _get_indexes_query = """
        SELECT attr.attname, idx.indkey, idx.indisunique, idx.indisprimary
        FROM pg_catalog.pg_class c,
            INNER JOIN pg_catalog.pg_index idx ON c.oid = idx.indrelid
            INNER JOIN pg_catalog.pg_class c2 ON idx.indexrelid = c2.oid
            INNER JOIN pg_catalog.pg_attribute attr ON attr.attrelid = c.oid and attr.attnum = idx.indkey[0]
            INNER JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
        WHERE c.relname = %s AND n.nspname = %s
    """

    def get_table_list(self, cursor):
        """
        Returns a list of table names in the current database and schema.
        """

        cursor.execute(
            """
            SELECT c.relname, c.relkind
            FROM pg_catalog.pg_class c
            LEFT JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
            WHERE c.relkind IN ('r', 'v', '')
                AND n.nspname = '%s'
                AND pg_catalog.pg_table_is_visible(c.oid)"""
            % get_current_schema().schema_name
        )

        return [
            TableInfo(row[0], {"r": "t", "v": "v"}.get(row[1]))
            for row in cursor.fetchall()
            if row[0] not in self.ignored_tables
        ]

    def get_table_description(self, cursor, table_name):
        "Returns a description of the table, with the DB-API cursor.description interface."
        # As cursor.description does not return reliably the nullable property,
        # we have to query the information_schema (#7783)
        cursor.execute(
            """
            SELECT column_name, is_nullable, column_default
            FROM information_schema.columns
            WHERE table_schema = %s and table_name = %s""",
            [get_current_schema().schema_name, table_name],
        )
        field_map = {line[0]: line[1:] for line in cursor.fetchall()}
        cursor.execute("SELECT * FROM %s LIMIT 1" % self.connection.ops.quote_name(table_name))
        return [
            FieldInfo(
                *(
                    (force_str(line[0]),)
                    + line[1:6]
                    + (field_map[force_str(line[0])][0] == "YES", field_map[force_str(line[0])][1])
                )
            )
            for line in cursor.description
        ]

    def get_indexes(self, cursor, table_name):
        # This query retrieves each index on the given table, including the
        # first associated field name
        cursor.execute(self._get_indexes_query, [table_name, get_current_schema().schema_name])
        indexes = {}
        for row in cursor.fetchall():
            # row[1] (idx.indkey) is stored in the DB as an array. It comes out as
            # a string of space-separated integers. This designates the field
            # indexes (1-based) of the fields that have indexes on the table.
            # Here, we skip any indexes across multiple fields.
            if " " in row[1]:
                continue
            if row[0] not in indexes:
                indexes[row[0]] = {"primary_key": False, "unique": False}
            # It's possible to have the unique and PK constraints in separate indexes.
            if row[3]:
                indexes[row[0]]["primary_key"] = True
            if row[2]:
                indexes[row[0]]["unique"] = True
        return indexes

    def get_relations(self, cursor, table_name):
        """
        Returns a dictionary of {field_name: (field_name_other_table, other_table)}
        representing all relationships to the given table.
        """
        cursor.execute(
            """
            SELECT c2.relname, a1.attname, a2.attname
            FROM pg_constraint con
            LEFT JOIN pg_class c1 ON con.conrelid = c1.oid
            LEFT JOIN pg_namespace n ON n.oid = c1.relnamespace
            LEFT JOIN pg_class c2 ON con.confrelid = c2.oid
            LEFT JOIN pg_attribute a1 ON c1.oid = a1.attrelid AND a1.attnum = con.conkey[1]
            LEFT JOIN pg_attribute a2 ON c2.oid = a2.attrelid AND a2.attnum = con.confkey[1]
            WHERE c1.relname = %s and n.nspname = %s
                AND con.contype = 'f'""",
            [table_name, get_current_schema().schema_name],
        )
        relations = {}
        for row in cursor.fetchall():
            relations[row[1]] = (row[2], row[0])
        return relations

    get_constraints = _constraints.get_constraints

    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' AND tc.table_schema = %s
        """,
            [table_name, get_current_schema().schema_name],
        )
        key_columns.extend(cursor.fetchall())
        return key_columns

1	from django.db.backends.base.introspection import FieldInfo, TableInfo	12✔
2	from django.utils.encoding import force_str	12✔
3
4	from django_pgschemas.schema import get_current_schema	12✔
5	from django_pgschemas.settings import get_base_backend_module, get_original_backend_module	12✔
6
7	from . import _constraints	12✔
8
9	try:	12✔
10	module = get_original_backend_module("introspection")	12✔
11	except AttributeError:	×
12	module = get_base_backend_module("introspection")	×
13
14
15	class DatabaseSchemaIntrospection(module.DatabaseIntrospection): # pragma: no cover
16	"""
17	database schema introspection class
18	"""
19
20	_get_indexes_query = """
21	SELECT attr.attname, idx.indkey, idx.indisunique, idx.indisprimary
22	FROM pg_catalog.pg_class c,
23	INNER JOIN pg_catalog.pg_index idx ON c.oid = idx.indrelid
24	INNER JOIN pg_catalog.pg_class c2 ON idx.indexrelid = c2.oid
25	INNER JOIN pg_catalog.pg_attribute attr ON attr.attrelid = c.oid and attr.attnum = idx.indkey[0]
26	INNER JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
27	WHERE c.relname = %s AND n.nspname = %s
28	"""
29
30	def get_table_list(self, cursor):
31	"""
32	Returns a list of table names in the current database and schema.
33	"""
34
35	cursor.execute(
36	"""
37	SELECT c.relname, c.relkind
38	FROM pg_catalog.pg_class c
39	LEFT JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
40	WHERE c.relkind IN ('r', 'v', '')
41	AND n.nspname = '%s'
42	AND pg_catalog.pg_table_is_visible(c.oid)"""
43	% get_current_schema().schema_name
44	)
45
46	return [
47	TableInfo(row[0], {"r": "t", "v": "v"}.get(row[1]))
48	for row in cursor.fetchall()
49	if row[0] not in self.ignored_tables
50	]
51
52	def get_table_description(self, cursor, table_name):
53	"Returns a description of the table, with the DB-API cursor.description interface."
54	# As cursor.description does not return reliably the nullable property,
55	# we have to query the information_schema (#7783)
56	cursor.execute(
57	"""
58	SELECT column_name, is_nullable, column_default
59	FROM information_schema.columns
60	WHERE table_schema = %s and table_name = %s""",
61	[get_current_schema().schema_name, table_name],
62	)
63	field_map = {line[0]: line[1:] for line in cursor.fetchall()}
64	cursor.execute("SELECT * FROM %s LIMIT 1" % self.connection.ops.quote_name(table_name))
65	return [
66	FieldInfo(
67	*(
68	(force_str(line[0]),)
69	+ line[1:6]
70	+ (field_map[force_str(line[0])][0] == "YES", field_map[force_str(line[0])][1])
71	)
72	)
73	for line in cursor.description
74	]
75
76	def get_indexes(self, cursor, table_name):
77	# This query retrieves each index on the given table, including the
78	# first associated field name
79	cursor.execute(self._get_indexes_query, [table_name, get_current_schema().schema_name])
80	indexes = {}
81	for row in cursor.fetchall():
82	# row[1] (idx.indkey) is stored in the DB as an array. It comes out as
83	# a string of space-separated integers. This designates the field
84	# indexes (1-based) of the fields that have indexes on the table.
85	# Here, we skip any indexes across multiple fields.
86	if " " in row[1]:
87	continue
88	if row[0] not in indexes:
89	indexes[row[0]] = {"primary_key": False, "unique": False}
90	# It's possible to have the unique and PK constraints in separate indexes.
91	if row[3]:
92	indexes[row[0]]["primary_key"] = True
93	if row[2]:
94	indexes[row[0]]["unique"] = True
95	return indexes
96
97	def get_relations(self, cursor, table_name):
98	"""
99	Returns a dictionary of {field_name: (field_name_other_table, other_table)}
100	representing all relationships to the given table.
101	"""
102	cursor.execute(
103	"""
104	SELECT c2.relname, a1.attname, a2.attname
105	FROM pg_constraint con
106	LEFT JOIN pg_class c1 ON con.conrelid = c1.oid
107	LEFT JOIN pg_namespace n ON n.oid = c1.relnamespace
108	LEFT JOIN pg_class c2 ON con.confrelid = c2.oid
109	LEFT JOIN pg_attribute a1 ON c1.oid = a1.attrelid AND a1.attnum = con.conkey[1]
110	LEFT JOIN pg_attribute a2 ON c2.oid = a2.attrelid AND a2.attnum = con.confkey[1]
111	WHERE c1.relname = %s and n.nspname = %s
112	AND con.contype = 'f'""",
113	[table_name, get_current_schema().schema_name],
114	)
115	relations = {}
116	for row in cursor.fetchall():
117	relations[row[1]] = (row[2], row[0])
118	return relations
119
120	get_constraints = _constraints.get_constraints
121
122	def get_key_columns(self, cursor, table_name):
123	key_columns = []
124	cursor.execute(
125	"""
126	SELECT kcu.column_name, ccu.table_name AS referenced_table, ccu.column_name AS referenced_column
127	FROM information_schema.constraint_column_usage ccu
128	LEFT JOIN information_schema.key_column_usage kcu
129	ON ccu.constraint_catalog = kcu.constraint_catalog
130	AND ccu.constraint_schema = kcu.constraint_schema
131	AND ccu.constraint_name = kcu.constraint_name
132	LEFT JOIN information_schema.table_constraints tc
133	ON ccu.constraint_catalog = tc.constraint_catalog
134	AND ccu.constraint_schema = tc.constraint_schema
135	AND ccu.constraint_name = tc.constraint_name
136	WHERE kcu.table_name = %s AND tc.constraint_type = 'FOREIGN KEY' AND tc.table_schema = %s
137	""",
138	[table_name, get_current_schema().schema_name],
139	)
140	key_columns.extend(cursor.fetchall())
141	return key_columns

lorinkoz / django-pgschemas / 10568469992

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous