3765774880

Build Type

github

Committed by GitHub

Commit Message

Merge 403e353d1 into 83ce13044

Pull Request Pull Request #132: Project refresh

Run Details

62 of 64 new or added lines in 8 files covered. (96.88%)

3 existing lines in 1 file now uncovered.

1144 of 1210 relevant lines covered (94.55%)

3.78 hits per line

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

90.0

/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 = None
            else:
                self._search_path = search_path_for_current_schema
            if name:
                cursor_for_search_path.close()
        return cursor

1	import psycopg2	4✔
2	from django.db.utils import DatabaseError	4✔
3
4	from ..schema import get_current_schema, get_default_schema	4✔
5	from ..utils import check_schema_name, get_limit_set_calls	4✔
6	from .introspection import DatabaseSchemaIntrospection	4✔
7	from .settings import EXTRA_SEARCH_PATHS, original_backend	4✔
8
9	IntegrityError = psycopg2.IntegrityError	4✔
10
11
12	def get_search_path(schema=None):	4✔
13	if schema is None:	4✔
14	schema = get_default_schema()	×
15
16	search_path = ["public"] if schema.schema_name == "public" else [schema.schema_name, "public"]	4✔
17	search_path.extend(EXTRA_SEARCH_PATHS)	4✔
18
19	for part in search_path:	4✔
20	check_schema_name(part)	4✔
21
22	return ", ".join(search_path)	4✔
23
24
25	class DatabaseWrapper(original_backend.DatabaseWrapper):	4✔
26	def __init__(self, args, *kwargs):	4✔
27	self._search_path = None	4✔
28	super().__init__(args, *kwargs)	4✔
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)	4✔
33
34	def close(self):	4✔
35	self._search_path = None	4✔
36	super().close()	4✔
37
38	def _cursor(self, name=None):	4✔
39	if name:	4✔
40	# Only supported and required by Django 1.11 (server-side cursor)
UNCOV 41	cursor = super()._cursor(name=name)	×
42	else:
43	cursor = super()._cursor()	4✔
44
45	search_path_for_current_schema = get_search_path(get_current_schema())	4✔
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):	4✔
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:	4✔
55	# Named cursor can only be used once
UNCOV 56	cursor_for_search_path = self.connection.cursor()	×
57	else:
58	# Reuse
59	cursor_for_search_path = cursor	4✔
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:	4✔
66	cursor_for_search_path.execute(f"SET search_path = {search_path_for_current_schema}")	4✔
67	except (DatabaseError, psycopg2.InternalError):	4✔
68	self._search_path = None	4✔
69	else:
70	self._search_path = search_path_for_current_schema	4✔
71	if name:	4✔
UNCOV 72	cursor_for_search_path.close()	×
73	return cursor	4✔

lorinkoz / django-pgschemas / 3765774880

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