5368812806

pending completion

Build # 5368812806

Build Type

Pull #153

github

Committed by

web-flow

Commit Message

Merge c8150986c into 0c3db7ba7

Pull Request Pull Request #153: Update workflows

Run Details

1141 of 1205 relevant lines covered (94.69%)

11.36 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	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✔
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)	×
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()	×
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 = None	12✔
69	else:
70	self._search_path = search_path_for_current_schema	12✔
71	if name:	12✔
72	cursor_for_search_path.close()	×
73	return cursor	12✔

lorinkoz / django-pgschemas / 5368812806

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