ff3d5f0090fae65618706d56ac15c18d63644a6b-PR-41

Committed 25 Jun 2020 - 9:10 coverage: 93.106% (+0.2%) from 92.863%

Build # ff3d5f0090fae65618706d56ac15c18d63644a6b-PR-41

Build Type

github-actions

Committed by GitHub

Commit Message

Merge 45646ebfa into bc3babb2f

Pull Request Pull Request #41: Experiment with distributing schemas across multiple databases

Run Details

123 of 128 new or added lines in 20 files covered. (96.09%)

1 existing line in 1 file now uncovered.

1148 of 1233 relevant lines covered (93.11%)

0.93 hits per line

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

93.94

/django_pgschemas/routers.py

from django.apps import apps
from django.conf import settings

from .schema import schema_handler
from .utils import get_clone_reference


class SyncRouter(object):
    """
    A router to control which applications will be synced depending on the schema we're syncing.
    It also controls database for read/write in a tenant sharding configuration.
    """

    def app_in_list(self, app_label, app_list):
        app_config = apps.get_app_config(app_label)
        app_config_full_name = "{}.{}".format(app_config.__module__, app_config.__class__.__name__)
        return (app_config.name in app_list) or (app_config_full_name in app_list)

    def db_for_read(self, model, **hints):
        if not schema_handler.active or schema_handler.active.schema_name in ["public", get_clone_reference()]:
            return None
        return schema_handler.active.get_database()

    def db_for_write(self, model, **hints):
        if not schema_handler.active or schema_handler.active.schema_name in ["public", get_clone_reference()]:
            return None
        return schema_handler.active.get_database()

    def allow_migrate(self, db, app_label, model_name=None, **hints):
        if not schema_handler.active:
            return False
        app_list = []
        databases = []
        if schema_handler.active.schema_name == "public":
            app_list = settings.TENANTS["public"]["APPS"]
            databases = settings.TENANTS["public"].get("DATABASES") or ["default"]
        elif schema_handler.active.schema_name in settings.TENANTS:
            app_list = settings.TENANTS[schema_handler.active.schema_name]["APPS"]
            databases = settings.TENANTS[schema_handler.active.schema_name].get("DATABASES") or ["default"]
        else:
            app_list = settings.TENANTS["default"]["APPS"]
            databases = settings.TENANTS["default"].get("DATABASES") or ["default"]
        if not app_list or not databases:
            return None
        return db in databases and self.app_in_list(app_label, app_list)

1	from django.apps import apps	1×
2	from django.conf import settings	1×
3
4	from .schema import schema_handler	1×
5	from .utils import get_clone_reference	1×
6
7
8	class SyncRouter(object):	1×
9	"""
10	A router to control which applications will be synced depending on the schema we're syncing.
11	It also controls database for read/write in a tenant sharding configuration.
12	"""
13
14	def app_in_list(self, app_label, app_list):	1×
15	app_config = apps.get_app_config(app_label)	1×
16	app_config_full_name = "{}.{}".format(app_config.__module__, app_config.__class__.__name__)	1×
17	return (app_config.name in app_list) or (app_config_full_name in app_list)	1×
18
19	def db_for_read(self, model, **hints):	1×
20	if not schema_handler.active or schema_handler.active.schema_name in ["public", get_clone_reference()]:	1×
21	return None	1×
22	return schema_handler.active.get_database()	1×
23
24	def db_for_write(self, model, **hints):	1×
25	if not schema_handler.active or schema_handler.active.schema_name in ["public", get_clone_reference()]:	1×
26	return None	1×
27	return schema_handler.active.get_database()	1×
28
29	def allow_migrate(self, db, app_label, model_name=None, **hints):	1×
30	if not schema_handler.active:	1×
31	return False	!
32	app_list = []	1×
33	databases = []	1×
34	if schema_handler.active.schema_name == "public":	1×
35	app_list = settings.TENANTS["public"]["APPS"]	1×
36	databases = settings.TENANTS["public"].get("DATABASES") or ["default"]	1×
37	elif schema_handler.active.schema_name in settings.TENANTS:	1×
38	app_list = settings.TENANTS[schema_handler.active.schema_name]["APPS"]	1×
39	databases = settings.TENANTS[schema_handler.active.schema_name].get("DATABASES") or ["default"]	1×
40	else:
41	app_list = settings.TENANTS["default"]["APPS"]	1×
42	databases = settings.TENANTS["default"].get("DATABASES") or ["default"]	1×
43	if not app_list or not databases:	1×
UNCOV 44	return None	!
45	return db in databases and self.app_in_list(app_label, app_list)	1×

lorinkoz / django-pgschemas / ff3d5f0090fae65618706d56ac15c18d63644a6b-PR-41

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