Experiment with distributing schemas across multiple databases

.github/workflows/code.yaml

@@ -14,8 +14,8 @@ jobs:
     strategy:
       max-parallel: 4
       matrix:
-        python-version: [3.6, 3.7, 3.8]
-        django-version: ["~=2.0.0", "~=2.1.0", "~=2.2.0", "~=3.0.0"]
+        python-version: [3.6, 3.7, 3.8, 3.9]
+        django-version: ["~=2.0.0", "~=2.1.0", "~=2.2.0", "~=3.0.0", "~=3.1.0"]
     services:
       postgres:
         image: postgres:latest

README.rst

@@ -50,6 +50,39 @@ Django project. It is a fork of `django-tenants`_ with some conceptual changes:
 
 .. _django-tenants: https://github.com/tomturner/django-tenants
 
+Which package to use?
+---------------------
+
+There are currently multiple packages to handle multi-tenancy via PostgreSQL schemas.
+This table should help you make an informed decision on which one to choose.
+
+.. list-table::
+   :widths: 50 50
+   :header-rows: 1
+
+   * - Package
+     - Features
+   * - `django-tenant-schemas`_
+     - Original project.
+       Now active and maintained by `@goodtune`_.
+   * - `django-tenants`_
+     - Active and maintained by `@tomturner`_.
+       Built on top of `django-tenant-schemas`_.
+       Uses a ``Domain`` model for allowing multiple domains per tenant.
+       Allows for parallel migrations with custom migration executor.
+       Other multiple improvements.
+   * - `django-pgschemas`_
+     - Active and maintained by `@lorinkoz`_.
+       Built on top of `django-tenants`_.
+       Different philosphy for tenants.
+       Other improvements listed above.
+
+.. _django-tenants-schemas: https://github.com/bernardopires/django-tenant-schemas
+.. _@goodtune: https://github.com/goodtune
+.. _django-tenants: https://github.com/tomturner/django-tenants
+.. _@tomturner: https://github.com/tomturner
+.. _django-pgschemas: https://github.com/lorinkoz/django-pgschemas
+.. _@lorinkoz: https://github.com/lorinkoz
 
 Documentation
 -------------

django_pgschemas/__init__.py

@@ -1 +1,3 @@
+from .schema import schema_handler
+
 default_app_config = "django_pgschemas.apps.DjangoPGSchemasConfig"

django_pgschemas/apps.py

@@ -17,10 +17,6 @@ def _check_tenant_dict(self):
     def _check_public_schema(self):
         if not isinstance(settings.TENANTS.get("public"), dict):
             raise ImproperlyConfigured("TENANTS must contain a 'public' dict.")
-        if "TENANT_MODEL" not in settings.TENANTS["public"]:
-            raise ImproperlyConfigured("TENANTS['public'] must contain a 'TENANT_MODEL' key.")
-        if "DOMAIN_MODEL" not in settings.TENANTS["public"]:
-            raise ImproperlyConfigured("TENANTS['public'] must contain a 'DOMAIN_MODEL' key.")
         if "URLCONF" in settings.TENANTS["public"]:
             raise ImproperlyConfigured("TENANTS['public'] cannot contain a 'URLCONF' key.")
         if "WS_URLCONF" in settings.TENANTS["public"]:
@@ -33,6 +29,10 @@ def _check_public_schema(self):
     def _check_default_schemas(self):
         if not isinstance(settings.TENANTS.get("default"), dict):
             raise ImproperlyConfigured("TENANTS must contain a 'default' dict.")
+        if "TENANT_MODEL" not in settings.TENANTS["default"]:
+            raise ImproperlyConfigured("TENANTS['default'] must contain a 'TENANT_MODEL' key.")
+        if "DOMAIN_MODEL" not in settings.TENANTS["default"]:
+            raise ImproperlyConfigured("TENANTS['default'] must contain a 'DOMAIN_MODEL' key.")
         if "URLCONF" not in settings.TENANTS["default"]:
             raise ImproperlyConfigured("TENANTS['default'] must contain a 'URLCONF' key.")
         if "DOMAINS" in settings.TENANTS["default"]:
@@ -43,7 +43,7 @@ def _check_default_schemas(self):
             "CLONE_REFERENCE" in settings.TENANTS["default"]
             and settings.TENANTS["default"]["CLONE_REFERENCE"] in settings.TENANTS
         ):
-            raise ImproperlyConfigured("TENANTS['default']['CLONE_REFERENCE'] must be a unique schema name")
+            raise ImproperlyConfigured("TENANTS['default']['CLONE_REFERENCE'] must be a unique schema name.")
 
     def _check_overall_schemas(self):
         for schema in settings.TENANTS:

django_pgschemas/cache.py → django_pgschemas/contrib/cache.py

@@ -1,4 +1,4 @@
-from django.db import connection
+from ..schema import schema_handler
 
 
 def make_key(key, key_prefix, version):
@@ -8,7 +8,7 @@ def make_key(key, key_prefix, version):
     Constructs the key used by all other methods. Prepends the tenant
     `schema_name` and `key_prefix'.
     """
-    return "%s:%s:%s:%s" % (connection.schema.schema_name, key_prefix, version, key)
+    return "%s:%s:%s:%s" % (schema_handler.active.schema_name, key_prefix, version, key)
 
 
 def reverse_key(key):

django_pgschemas/contrib/channels/auth.py

@@ -1,11 +1,10 @@
+from channels.auth import AuthMiddleware, CookieMiddleware, SessionMiddleware, _get_user_session_key, login, logout
+from channels.db import database_sync_to_async
 from django.conf import settings
 from django.contrib.auth import BACKEND_SESSION_KEY, HASH_SESSION_KEY, load_backend
 from django.contrib.auth.models import AnonymousUser
 from django.utils.crypto import constant_time_compare
 
-from channels.auth import login, logout, CookieMiddleware, SessionMiddleware, AuthMiddleware, _get_user_session_key
-from channels.db import database_sync_to_async
-
 
 @database_sync_to_async
 def get_user(scope):

django_pgschemas/contrib/channels/router.py

@@ -1,12 +1,11 @@
+from channels.routing import ProtocolTypeRouter, URLRouter
 from django.conf import settings
 from django.core.exceptions import ImproperlyConfigured
 from django.utils.encoding import force_text
 from django.utils.module_loading import import_string
 
-from channels.routing import ProtocolTypeRouter, URLRouter
-
 from ...schema import SchemaDescriptor
-from ...utils import remove_www, get_tenant_model, get_domain_model
+from ...utils import get_domain_model, get_tenant_model, remove_www
 from .auth import TenantAuthMiddlewareStack
 
 

django_pgschemas/contrib/files/storage.py

@@ -2,7 +2,8 @@
 
 from django.conf import settings
 from django.core.files.storage import FileSystemStorage
-from django.db import connection
+
+from ...schema import schema_handler
 
 
 class TenantFileSystemStorage(FileSystemStorage):
@@ -12,13 +13,13 @@ class TenantFileSystemStorage(FileSystemStorage):
     """
 
     def get_schema_path_identifier(self):
-        if not connection.schema:
+        if not schema_handler.active:
             return ""
-        path_identifier = connection.schema.schema_name
-        if hasattr(connection.schema, "schema_pathname"):
-            path_identifier = connection.schema.schema_pathname()
+        path_identifier = schema_handler.active.schema_name
+        if hasattr(schema_handler.active, "schema_pathname"):
+            path_identifier = schema_handler.active.schema_pathname()
         elif hasattr(settings, "PGSCHEMAS_PATHNAME_FUNCTION"):
-            path_identifier = settings.PGSCHEMAS_PATHNAME_FUNCTION(connection.schema)
+            path_identifier = settings.PGSCHEMAS_PATHNAME_FUNCTION(schema_handler.active)
         return path_identifier
 
     @property  # To avoid caching of tenant
@@ -44,7 +45,7 @@ def base_url(self):
         appended.
         """
         url_folder = self.get_schema_path_identifier()
-        if url_folder and connection.schema and connection.schema.folder:
+        if url_folder and schema_handler.active and schema_handler.active.folder:
             # Since we're already prepending all URLs with schema, there is no
             # need to make the differentiation here
             url_folder = ""

django_pgschemas/log.py

@@ -1,6 +1,6 @@
 import logging
 
-from django.db import connection
+from .schema import schema_handler
 
 
 class SchemaContextFilter(logging.Filter):
@@ -9,6 +9,6 @@ class SchemaContextFilter(logging.Filter):
     """
 
     def filter(self, record):
-        record.schema_name = connection.schema.schema_name
-        record.domain_url = connection.schema.domain_url
+        record.schema_name = schema_handler.active.schema_name
+        record.domain_url = schema_handler.active.domain_url
         return True

django_pgschemas/management/commands/__init__.py

@@ -1,11 +1,10 @@
 from django.conf import settings
 from django.core.management.base import BaseCommand, CommandError
-from django.db.models import CharField, Q
-from django.db.models import Value as V
+from django.db.models import CharField, Q, Value as V
 from django.db.models.functions import Concat
 from django.db.utils import ProgrammingError
 
-from ...schema import SchemaDescriptor
+from ...schema import schema_handler
 from ...utils import create_schema, dynamic_models_exist, get_clone_reference, get_tenant_model
 from ._executors import parallel, sequential
 
@@ -67,11 +66,13 @@ def add_arguments(self, parser):
             help="Schema(s) to exclude when executing the current command",
         )
         parser.add_argument(
-            "--executor",
-            dest="executor",
-            default="sequential",
-            choices=EXECUTORS,
-            help="Executor to be used for running command on schemas",
+            "--sdb", nargs="?", dest="schema_database", default="default", help="Database to operate with the schema(s)"
+        )
+        parser.add_argument(
+            "--parallel",
+            dest="parallel",
+            action="store_true",
+            help="Run command in parallel mode",
         )
         parser.add_argument(
             "--no-create-schemas",
@@ -81,6 +82,7 @@ def add_arguments(self, parser):
         )
 
     def get_schemas_from_options(self, **options):
+        database = options.get("database") or options.get("schema_database")
         skip_schema_creation = options.get("skip_schema_creation", False)
         try:
             schemas = self._get_schemas_from_options(**options)
@@ -97,16 +99,17 @@ def get_schemas_from_options(self, **options):
                 raise CommandError("This command can only run in %s" % self.specific_schemas)
         if not skip_schema_creation:
             for schema in schemas:
-                create_schema(schema, check_if_exists=True, sync_schema=False, verbosity=0)
+                create_schema(schema, database, check_if_exists=True, sync_schema=False, verbosity=0)
         return schemas
 
     def get_executor_from_options(self, **options):
-        return EXECUTORS[options.get("executor")]
+        return EXECUTORS["parallel"] if options.get("parallel") else EXECUTORS["sequential"]
 
     def get_scope_display(self):
         return "|".join(self.specific_schemas or []) or self.scope
 
     def _get_schemas_from_options(self, **options):
+        database = options.get("database") or options.get("schema_database")
         schemas = options.get("schemas") or []
         excluded_schemas = options.get("excluded_schemas") or []
         include_all_schemas = options.get("all_schemas") or False
@@ -140,9 +143,19 @@ def _get_schemas_from_options(self, **options):
                 raise CommandError("No schema provided")
 
         TenantModel = get_tenant_model()
-        static_schemas = [x for x in settings.TENANTS.keys() if x != "default"] if allow_static else []
+        static_schemas = (
+            [
+                x
+                for x in settings.TENANTS.keys()
+                if x != "default" and database in (settings.TENANTS[x].get("DATABASES") or ["default"])
+            ]
+            if allow_static
+            else []
+        )
         dynamic_schemas = (
-            TenantModel.objects.values_list("schema_name", flat=True) if dynamic_ready and allow_dynamic else []
+            [x.schema_name for x in TenantModel.objects.all() if x.get_database() == database]
+            if dynamic_ready and allow_dynamic
+            else []
         )
         if clone_reference and allow_static:
             static_schemas.append(clone_reference)
@@ -173,8 +186,10 @@ def _get_schemas_from_options(self, **options):
                 schemas_to_return.add(schema)
             elif schema == clone_reference:
                 schemas_to_return.add(schema)
-            elif dynamic_ready and TenantModel.objects.filter(schema_name=schema).exists() and allow_dynamic:
-                schemas_to_return.add(schema)
+            elif dynamic_ready and allow_dynamic:
+                tenant = TenantModel.objects.filter(schema_name=schema).first()
+                if tenant and tenant.get_database() == database:
+                    schemas_to_return.add(schema)
 
         schemas = list(set(schemas) - schemas_to_return)
 
@@ -188,13 +203,18 @@ def _get_schemas_from_options(self, **options):
                     and any([x for x in data["DOMAINS"] if x.startswith(schema)])
                 ]
             if dynamic_ready and allow_dynamic:
-                local += (
-                    TenantModel.objects.annotate(
-                        route=Concat("domains__domain", V("/"), "domains__folder", output_field=CharField())
-                    )
-                    .filter(Q(schema_name=schema) | Q(domains__domain__istartswith=schema) | Q(route=schema))
-                    .distinct()
-                    .values_list("schema_name", flat=True)
+                local += list(
+                    {
+                        x.schema_name
+                        for x in (
+                            TenantModel.objects.annotate(
+                                route=Concat("domains__domain", V("/"), "domains__folder", output_field=CharField())
+                            )
+                            .filter(Q(schema_name=schema) | Q(domains__domain__istartswith=schema) | Q(route=schema))
+                            .distinct()
+                        )
+                        if x.get_database() == database
+                    }
                 )
             if not local:
                 raise CommandError("No schema found for '%s'" % schema)
@@ -216,13 +236,18 @@ def _get_schemas_from_options(self, **options):
                 if schema_name not in ["public", "default", clone_reference]
                 and any([x for x in data["DOMAINS"] if x.startswith(schema)])
             ]
-            local += (
-                TenantModel.objects.annotate(
-                    route=Concat("domains__domain", V("/"), "domains__folder", output_field=CharField())
-                )
-                .filter(Q(schema_name=schema) | Q(domains__domain__istartswith=schema) | Q(route=schema))
-                .distinct()
-                .values_list("schema_name", flat=True)
+            local += list(
+                {
+                    x.schema_name
+                    for x in (
+                        TenantModel.objects.annotate(
+                            route=Concat("domains__domain", V("/"), "domains__folder", output_field=CharField())
+                        )
+                        .filter(Q(schema_name=schema) | Q(domains__domain__istartswith=schema) | Q(route=schema))
+                        .distinct()
+                    )
+                    if x.get_database() == database
+                }
             )
             if not local:
                 raise CommandError("No schema found for '%s' (excluded)" % schema)
@@ -248,18 +273,8 @@ def handle(self, *args, **options):
         executor(schemas, self, "_raw_handle_tenant", args, options, pass_schema_in_kwargs=True)
 
     def _raw_handle_tenant(self, *args, **kwargs):
-        schema_name = kwargs.pop("schema_name")
-        if schema_name in settings.TENANTS:
-            domains = settings.TENANTS[schema_name].get("DOMAINS", [])
-            tenant = SchemaDescriptor.create(schema_name=schema_name, domain_url=domains[0] if domains else None)
-            self.handle_tenant(tenant, *args, **kwargs)
-        elif schema_name == get_clone_reference():
-            tenant = SchemaDescriptor.create(schema_name=schema_name)
-            self.handle_tenant(tenant, *args, **kwargs)
-        else:
-            TenantModel = get_tenant_model()
-            tenant = TenantModel.objects.get(schema_name=schema_name)
-            self.handle_tenant(tenant, *args, **kwargs)
+        kwargs.pop("schema_name")
+        self.handle_tenant(schema_handler.active, *args, **kwargs)
 
     def handle_tenant(self, tenant, *args, **options):
         pass

django_pgschemas/management/commands/_executors.py

@@ -3,8 +3,11 @@
 
 from django.conf import settings
 from django.core.management import call_command
-from django.core.management.base import BaseCommand, OutputWrapper, CommandError
-from django.db import connection, transaction, connections
+from django.core.management.base import BaseCommand, OutputWrapper
+from django.db import connection, connections, transaction
+
+from ...schema import SchemaDescriptor, schema_handler
+from ...utils import get_clone_reference, get_tenant_model
 
 
 def run_on_schema(
@@ -52,7 +55,17 @@ def __call__(self, message):
 
     if fork_db:
         connections.close_all()
-    connection.set_schema_to(schema_name)
+
+    if schema_name in settings.TENANTS:
+        domains = settings.TENANTS[schema_name].get("DOMAINS", [])
+        schema = SchemaDescriptor.create(schema_name=schema_name, domain_url=domains[0] if domains else None)
+    elif schema_name == get_clone_reference():
+        schema = SchemaDescriptor.create(schema_name=schema_name)
+    else:
+        TenantModel = get_tenant_model()
+        schema = TenantModel.objects.get(schema_name=schema_name)
+
+    schema_handler.set_schema(schema)
 
     if pass_schema_in_kwargs:
         kwargs.update({"schema_name": schema_name})