6618841281

Committed 23 Oct 2023 08:47PM CUT coverage: 93.617%. Remained the same

Build # 6618841281

Build Type

Pull #187

github

Committed by

web-flow

Commit Message

Merge a81f35a98 into b4182e6e1

Pull Request Pull Request #187: Bump ruff from 0.1.0 to 0.1.1

Run Details

1188 of 1269 relevant lines covered (93.62%)

5.61 hits per line

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

77.78

/django_pgschemas/postgresql_backend/introspection.py

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

from ..schema import get_current_schema
from . import _constraints
from .settings import base_backend, original_backend

try:
    DatabaseIntrospection = original_backend.DatabaseIntrospection
except AttributeError:
    DatabaseIntrospection = base_backend.DatabaseIntrospection


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

lorinkoz / django-pgschemas / 6618841281

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