Merge 4a58beb into ec7829e

.github/workflows/code.yaml

@@ -49,6 +49,7 @@ jobs:
       - name: Run Tests
         run: |
           poetry run coverage run dpgs_sandbox/manage.py test tests -r
+          poetry run dpgs_sandbox/manage.py test tests --settings settings.static_only -r
           poetry run coverage lcov -o ./coverage/lcov.info
       - name: Upload coverage to Coveralls in parallel
         uses: coverallsapp/github-action@master

django_pgschemas/apps.py

@@ -27,7 +27,10 @@ def _check_public_schema(self):
             raise ImproperlyConfigured("TENANTS['public'] cannot contain a 'FALLBACK_DOMAINS' key.")
 
     def _check_default_schemas(self):
-        if not isinstance(settings.TENANTS.get("default"), dict):
+        if "default" not in settings.TENANTS:
+            return  # Escape hatch for static only configs
+
+        if not isinstance(settings.TENANTS["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.")
@@ -60,6 +63,9 @@ def _check_complementary_settings(self):
     def _check_extra_search_paths(self):
         if hasattr(settings, "PGSCHEMAS_EXTRA_SEARCH_PATHS"):
             TenantModel = get_tenant_model()
+            if TenantModel is None:
+                return
+
             cursor = connection.cursor()
             cursor.execute(
                 "SELECT 1 FROM information_schema.tables WHERE table_name = %s;", [TenantModel._meta.db_table]

django_pgschemas/checks.py

@@ -1,3 +1,5 @@
+from typing import Any, Optional
+
 from django.conf import settings
 from django.contrib.auth import get_user_model
 from django.contrib.sessions.base_session import AbstractBaseSession
@@ -9,22 +11,28 @@
 from .utils import get_clone_reference, get_domain_model, get_tenant_model
 
 
-def get_tenant_app():
-    return get_tenant_model(require_ready=False)._meta.app_config.name
+def get_tenant_app() -> Optional[str]:
+    TenantModel = get_tenant_model(require_ready=False)
+    if TenantModel is None:
+        return None
+    return TenantModel._meta.app_config.name
 
 
-def get_domain_app():
-    return get_domain_model(require_ready=False)._meta.app_config.name
+def get_domain_app() -> Optional[str]:
+    DomainModel = get_domain_model(require_ready=False)
+    if DomainModel is None:
+        return None
+    return DomainModel._meta.app_config.name
 
 
-def get_user_app():
+def get_user_app() -> Optional[str]:
     try:
         return get_user_model()._meta.app_config.name
     except ImproperlyConfigured:
         return None
 
 
-def get_session_app():
+def get_session_app() -> Optional[str]:
     engine = import_module(settings.SESSION_ENGINE)
     store = engine.SessionStore
     if hasattr(store, "get_model_class"):
@@ -35,10 +43,14 @@ def get_session_app():
 
 
 @checks.register()
-def check_principal_apps(app_configs, **kwargs):
+def check_principal_apps(app_configs: Any, **kwargs: Any) -> None:
     errors = []
     tenant_app = get_tenant_app()
     domain_app = get_domain_app()
+
+    if tenant_app is None or domain_app is None:
+        return []
+
     if tenant_app not in settings.TENANTS["public"].get("APPS", []):
         errors.append(
             checks.Error("Your tenant app '%s' must be on the 'public' schema." % tenant_app, id="pgschemas.W001")
@@ -47,6 +59,7 @@ def check_principal_apps(app_configs, **kwargs):
         errors.append(
             checks.Error("Your domain app '%s' must be on the 'public' schema." % domain_app, id="pgschemas.W001")
         )
+
     for schema in settings.TENANTS:
         schema_apps = settings.TENANTS[schema].get("APPS", [])
         if schema == "public":
@@ -67,21 +80,24 @@ def check_principal_apps(app_configs, **kwargs):
                     id="pgschemas.W001",
                 )
             )
+
     return errors
 
 
 @checks.register()
-def check_other_apps(app_configs, **kwargs):
+def check_other_apps(app_configs: Any, **kwargs: Any) -> None:
     errors = []
     user_app = get_user_app()
     session_app = get_session_app()
-    if "django.contrib.contenttypes" in settings.TENANTS["default"].get("APPS", []):
+
+    if "django.contrib.contenttypes" in settings.TENANTS.get("default", {}).get("APPS", []):
         errors.append(
             checks.Warning(
                 "'django.contrib.contenttypes' in TENANTS['default']['APPS'] must be on 'public' schema only.",
                 id="pgschemas.W002",
             )
         )
+
     for schema in settings.TENANTS:
         schema_apps = settings.TENANTS[schema].get("APPS", [])
         if schema not in ["public", "default"]:
@@ -108,28 +124,37 @@ def check_other_apps(app_configs, **kwargs):
                         id="pgschemas.W003",
                     )
                 )
+
     return errors
 
 
 @checks.register(checks.Tags.database)
-def check_schema_names(app_configs, **kwargs):
+def check_schema_names(app_configs: Any, **kwargs: Any) -> None:
     errors = []
     static_names = set(settings.TENANTS.keys())
     clone_reference = get_clone_reference()
+    TenantModel = get_tenant_model()
+
+    if TenantModel is None:
+        return []
+
     if clone_reference:
         static_names.add(clone_reference)
     try:
-        dynamic_names = set(get_tenant_model().objects.values_list("schema_name", flat=True))
+        dynamic_names = set(TenantModel.objects.values_list("schema_name", flat=True))
     except ProgrammingError:
         # This happens on the first run of migrate, with empty database.
         # It can also happen when the tenant model contains unapplied migrations that break.
         dynamic_names = set()
+
     intersection = static_names & dynamic_names
+
     if intersection:
         errors.append(
             checks.Critical(
                 "Name clash found between static and dynamic tenants: %s" % intersection,
                 id="pgschemas.W004",
             )
         )
+
     return errors

django_pgschemas/management/commands/__init__.py

@@ -164,7 +164,9 @@ def _get_schemas_from_options(self, **options):
         TenantModel = get_tenant_model()
         static_schemas = [x for x in settings.TENANTS.keys() if x != "default"] if allow_static else []
         dynamic_schemas = (
-            TenantModel.objects.values_list("schema_name", flat=True) if dynamic_ready and allow_dynamic else []
+            TenantModel.objects.values_list("schema_name", flat=True)
+            if TenantModel is not None and dynamic_ready and allow_dynamic
+            else []
         )
         if clone_reference and allow_static:
             static_schemas.append(clone_reference)
@@ -195,7 +197,12 @@ def find_schema_by_reference(reference, as_excluded=False):
                 return reference
             elif reference == clone_reference:
                 return reference
-            elif dynamic_ready and TenantModel.objects.filter(schema_name=reference).exists() and allow_dynamic:
+            elif (
+                TenantModel is not None
+                and dynamic_ready
+                and TenantModel.objects.filter(schema_name=reference).exists()
+                and allow_dynamic
+            ):
                 return reference
             else:
                 local = []
@@ -206,7 +213,7 @@ def find_schema_by_reference(reference, as_excluded=False):
                         if schema_name not in ["public", "default"]
                         and any(x for x in data["DOMAINS"] if x.startswith(reference))
                     ]
-                if dynamic_ready and allow_dynamic:
+                if TenantModel is not None and dynamic_ready and allow_dynamic:
                     local += (
                         TenantModel.objects.annotate(
                             route=Concat("domains__domain", V("/"), "domains__folder", output_field=CharField())

django_pgschemas/management/commands/_executors.py

@@ -3,7 +3,7 @@
 
 from django.conf import settings
 from django.core.management import call_command
-from django.core.management.base import BaseCommand, OutputWrapper
+from django.core.management.base import BaseCommand, CommandError, OutputWrapper
 from django.db import connection, connections, transaction
 
 from ...schema import SchemaDescriptor, activate
@@ -66,9 +66,10 @@ def __call__(self, message):
         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()
+    elif (TenantModel := get_tenant_model()) is not None:
         schema = TenantModel.objects.get(schema_name=schema_name)
+    else:
+        raise CommandError(f"Unable to find schema {schema_name}!")
 
     activate(schema)
 

django_pgschemas/management/commands/cloneschema.py

@@ -100,7 +100,7 @@ def handle(self, *args, **options):
         dry_run = options.get("dry_run")
         if options.get("interactive", True):
             TenantModel = get_tenant_model()
-            if TenantModel.objects.filter(schema_name=options["source"]).exists():
+            if TenantModel is not None and TenantModel.objects.filter(schema_name=options["source"]).exists():
                 tenant, domain = self.get_dynamic_tenant(**options)
         try:
             clone_schema(options["source"], options["destination"], dry_run)

django_pgschemas/middleware.py

@@ -41,14 +41,18 @@ def middleware(request):
         else:
             DomainModel = get_domain_model()
             prefix = request.path.split("/")[1]
-            try:
-                domain = DomainModel.objects.select_related("tenant").get(domain=hostname, folder=prefix)
-            except DomainModel.DoesNotExist:
+            domain = None
+
+            if DomainModel is not None:
                 try:
-                    domain = DomainModel.objects.select_related("tenant").get(domain=hostname, folder="")
+                    domain = DomainModel.objects.select_related("tenant").get(domain=hostname, folder=prefix)
                 except DomainModel.DoesNotExist:
-                    domain = None
-            if domain:
+                    try:
+                        domain = DomainModel.objects.select_related("tenant").get(domain=hostname, folder="")
+                    except DomainModel.DoesNotExist:
+                        pass
+
+            if domain is not None:
                 tenant = domain.tenant
                 tenant.domain_url = hostname
                 tenant.folder = None

django_pgschemas/models.py

@@ -104,8 +104,12 @@ class DomainMixin(models.Model):
     All models that store the domains must inherit this class.
     """
 
-    tenant = models.ForeignKey(
-        settings.TENANTS["default"]["TENANT_MODEL"], db_index=True, related_name="domains", on_delete=models.CASCADE
+    tenant = (
+        models.ForeignKey(
+            settings.TENANTS["default"]["TENANT_MODEL"], db_index=True, related_name="domains", on_delete=models.CASCADE
+        )
+        if "default" in settings.TENANTS
+        else None
     )
 
     domain = models.CharField(max_length=253, db_index=True)

django_pgschemas/signals.py

@@ -18,8 +18,11 @@
 
 @receiver(pre_delete)
 def tenant_delete_callback(sender, instance, **kwargs):
-    if not isinstance(instance, get_tenant_model()):
+    TenantModel = get_tenant_model()
+    if TenantModel is None:
+        return
+    if not isinstance(instance, TenantModel):
         return
     if instance.auto_drop_schema and schema_exists(instance.schema_name):
-        dynamic_tenant_pre_drop.send(sender=get_tenant_model(), tenant=instance.serializable_fields())
+        dynamic_tenant_pre_drop.send(sender=TenantModel, tenant=instance.serializable_fields())
         instance.drop_schema()

django_pgschemas/utils.py

@@ -10,13 +10,17 @@
 from django.db.models import Model
 
 
-def get_tenant_model(require_ready: bool = True) -> Model:
+def get_tenant_model(require_ready: bool = True) -> Optional[Model]:
     "Returns the tenant model."
+    if "default" not in settings.TENANTS:
+        return None
     return apps.get_model(settings.TENANTS["default"]["TENANT_MODEL"], require_ready=require_ready)
 
 
-def get_domain_model(require_ready: bool = True) -> Model:
+def get_domain_model(require_ready: bool = True) -> Optional[Model]:
     "Returns the domain model."
+    if "default" not in settings.TENANTS:
+        return None
     return apps.get_model(settings.TENANTS["default"]["DOMAIN_MODEL"], require_ready=require_ready)
 
 
@@ -29,6 +33,8 @@ def get_limit_set_calls() -> bool:
 
 
 def get_clone_reference() -> Optional[str]:
+    if "default" not in settings.TENANTS:
+        return None
     return settings.TENANTS["default"].get("CLONE_REFERENCE", None)
 
 
@@ -115,8 +121,14 @@ def dynamic_models_exist() -> bool:
     WHERE  table_schema = 'public'
     AND    table_name in ('%s', '%s');
     """
+    TenantModel = get_tenant_model()
+    DomainModel = get_domain_model()
+
+    if TenantModel is None or DomainModel is None:
+        return False
+
     cursor = connection.cursor()
-    cursor.execute(sql % (get_tenant_model()._meta.db_table, get_domain_model()._meta.db_table))
+    cursor.execute(sql % (TenantModel._meta.db_table, DomainModel._meta.db_table))
     value = cursor.fetchone() == (2,)
     cursor.close()
     return value

docs/advanced.rst

@@ -131,6 +131,50 @@ hand, an incoming request for ``mydomain.com/some/url/`` will fail for all
 static tenants, then fail for all dynamic tenants, and will finally match
 against the fallback domains of the main tenant.
 
+Static-only tenants
+-------------------
+
+It's also possible to have only static tenants. For this, the default key must
+be omitted:
+
+.. code-block:: python
+
+    TENANTS = {
+        "public": {
+            "APPS": [
+                "django.contrib.contenttypes",
+                "django.contrib.staticfiles",
+                # ...
+                "django_pgschemas",
+                "shared_app",
+                # ...
+            ],
+        },
+        "www": {
+            "APPS": [
+                "django.contrib.auth",
+                "django.contrib.sessions",
+                # ...
+                "main_app",
+            ],
+            "DOMAINS": ["mydomain.com"],
+            "URLCONF": "main_app.urls",
+        },
+        "blog": {
+            "APPS": [
+                "django.contrib.auth",
+                "django.contrib.sessions",
+                # ...
+                "blog_app",
+            ],
+            "DOMAINS": ["blog.mydomain.com", "help.mydomain.com"],
+            "URLCONF": "blog_app.urls",
+        }
+    }
+
+In this case, no model is expected to inherit from ``TenantMixin`` and
+``DomainMixin``, and no clone reference schema can be created.
+
 Running management commands
 ---------------------------
 

dpgs_sandbox/manage.py

@@ -3,7 +3,7 @@
 import sys
 
 if __name__ == "__main__":
-    os.environ.setdefault("DJANGO_SETTINGS_MODULE", "settings")
+    os.environ.setdefault("DJANGO_SETTINGS_MODULE", "settings.full")
     try:
         from django.core.management import execute_from_command_line
     except ImportError: