2940841586

Committed 27 Aug 2022 - 22:19 coverage: 94.706% (-0.3%) from 94.967%

Build Type

github

Committed by GitHub

Commit Message

Merge 287950c28 into 574651d20

Pull Request Pull Request #120: Major code revamp

Run Details

29 of 30 new or added lines in 5 files covered. (96.67%)

1145 of 1209 relevant lines covered (94.71%)

0.95 hits per line

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

97.5

/django_pgschemas/postgresql_backend/base.py

import psycopg2
from django.db.utils import DatabaseError

from ..schema import get_current_schema, get_default_schema
from ..utils import check_schema_name, get_limit_set_calls
from .introspection import DatabaseSchemaIntrospection
from .settings import EXTRA_SEARCH_PATHS, original_backend

IntegrityError = psycopg2.IntegrityError


def get_search_path(schema=None):
    if schema is None:
        schema = get_default_schema()

    search_path = ["public"] if schema.schema_name == "public" else [schema.schema_name, "public"]
    search_path.extend(EXTRA_SEARCH_PATHS)

    for part in search_path:
        check_schema_name(part)

    return ", ".join(search_path)


class DatabaseWrapper(original_backend.DatabaseWrapper):
    def __init__(self, *args, **kwargs):
        self._search_path = None
        super().__init__(*args, **kwargs)

        # Use a patched version of the DatabaseIntrospection that only returns the table list for the
        # currently selected schema.
        self.introspection = DatabaseSchemaIntrospection(self)

    def close(self):
        self._search_path = None
        super().close()

    def _cursor(self, name=None):
        if name:
            # Only supported and required by Django 1.11 (server-side cursor)
            cursor = super()._cursor(name=name)
        else:
            cursor = super()._cursor()

        search_path_for_current_schema = get_search_path(get_current_schema())

        # optionally limit the number of executions - under load, the execution
        # of `set search_path` can be quite time consuming
        if (not get_limit_set_calls()) or (self._search_path != search_path_for_current_schema):
            # Actual search_path modification for the cursor. Database will
            # search schemas from left to right when looking for the object
            # (table, index, sequence, etc.).

            if name:
                # Named cursor can only be used once
                cursor_for_search_path = self.connection.cursor()
            else:
                # Reuse
                cursor_for_search_path = cursor

            # In the event that an error already happened in this transaction and we are going
            # to rollback we should just ignore database error when setting the search_path
            # if the next instruction is not a rollback it will just fail also, so
            # we do not have to worry that it's not the good one
            try:
                cursor_for_search_path.execute(f"SET search_path = {search_path_for_current_schema}")
            except (DatabaseError, psycopg2.InternalError):
                self._search_path = search_path_for_current_schema
            else:
                self._search_path = None
            if name:
                cursor_for_search_path.close()
        return cursor

1	import psycopg2	12×
2	from django.db.utils import DatabaseError	12×
3
4	from ..schema import get_current_schema, get_default_schema	12×
5	from ..utils import check_schema_name, get_limit_set_calls	12×
6	from .introspection import DatabaseSchemaIntrospection	12×
7	from .settings import EXTRA_SEARCH_PATHS, original_backend	12×
8
9	IntegrityError = psycopg2.IntegrityError	12×
10
11
12	def get_search_path(schema=None):	12×
13	if schema is None:	12×
NEW 14	schema = get_default_schema()	!
15
16	search_path = ["public"] if schema.schema_name == "public" else [schema.schema_name, "public"]	12×
17	search_path.extend(EXTRA_SEARCH_PATHS)	12×
18
19	for part in search_path:	12×
20	check_schema_name(part)	12×
21
22	return ", ".join(search_path)	12×
23
24
25	class DatabaseWrapper(original_backend.DatabaseWrapper):	12×
26	def __init__(self, args, *kwargs):	12×
27	self._search_path = None	12×
28	super().__init__(args, *kwargs)	12×
29
30	# Use a patched version of the DatabaseIntrospection that only returns the table list for the
31	# currently selected schema.
32	self.introspection = DatabaseSchemaIntrospection(self)	12×
33
34	def close(self):	12×
35	self._search_path = None	12×
36	super().close()	12×
37
38	def _cursor(self, name=None):	12×
39	if name:	12×
40	# Only supported and required by Django 1.11 (server-side cursor)
41	cursor = super()._cursor(name=name)	6×
42	else:
43	cursor = super()._cursor()	12×
44
45	search_path_for_current_schema = get_search_path(get_current_schema())	12×
46
47	# optionally limit the number of executions - under load, the execution
48	# of `set search_path` can be quite time consuming
49	if (not get_limit_set_calls()) or (self._search_path != search_path_for_current_schema):	12×
50	# Actual search_path modification for the cursor. Database will
51	# search schemas from left to right when looking for the object
52	# (table, index, sequence, etc.).
53
54	if name:	12×
55	# Named cursor can only be used once
56	cursor_for_search_path = self.connection.cursor()	6×
57	else:
58	# Reuse
59	cursor_for_search_path = cursor	12×
60
61	# In the event that an error already happened in this transaction and we are going
62	# to rollback we should just ignore database error when setting the search_path
63	# if the next instruction is not a rollback it will just fail also, so
64	# we do not have to worry that it's not the good one
65	try:	12×
66	cursor_for_search_path.execute(f"SET search_path = {search_path_for_current_schema}")	12×
67	except (DatabaseError, psycopg2.InternalError):	12×
68	self._search_path = search_path_for_current_schema	12×
69	else:
70	self._search_path = None	12×
71	if name:	12×
72	cursor_for_search_path.close()	6×
73	return cursor	12×

lorinkoz / django-pgschemas / 2940841586

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