Minimum required celery version is 5.0

local-compose.yml

@@ -0,0 +1,39 @@
+version: "3.6"
+services:
+  postgres:
+    image: postgres:13
+    environment:
+      - POSTGRES_PASSWORD=qwe123
+      - POSTGRES_USER=tenant_celery
+      - POSTGRES_DB=tenant_celery
+    network_mode: host
+
+  rabbitmq:
+    image: rabbitmq
+    network_mode: host
+
+  redis:
+    image: redis
+    network_mode: host
+
+  app:
+    image: "python:${PYTHON_VERSION:-3.8}-slim"
+    depends_on:
+      - postgres
+      - redis
+    environment:
+      - DATABASE_HOST=postgres
+      - BROKER_URL=${BROKER_URL:-amqp://guest:guest@rabbitmq:5672/}
+      - ADDITIONAL_REQUIREMENTS=${ADDITIONAL_REQUIREMENTS}
+      - TASK_TENANT_CACHE_SECONDS=10
+
+    volumes:
+      - ./tenant_schemas_celery:/app/tenant_schemas_celery
+      - ./test_app:/app/test_app
+      - ./requirements.txt:/app/requirements.txt
+      - ./__app_run_tests:/app/__app_run_tests
+      - ./setup.py:/app/setup.py
+      - ./VERSION:/app/VERSION
+      - ./README.md:/app/README.md
+
+    command: ["bash", "-c", "cd /app && ./__app_run_tests"]

run-celery-worker

@@ -0,0 +1,6 @@
+#!/bin/bash
+. .env/bin/activate
+cd test_app
+export DJANGO_SETTINGS_MODULE=test_app.settings
+
+celery -A tenant_schemas_celery.test_app:app worker -l INFO

setup.py

@@ -24,7 +24,7 @@
     ],
     description='Celery integration for django-tenant-schemas and django-tenants',
     install_requires=[
-        'celery',
+        'celery>=5',
     ],
     packages=find_packages(),
     python_requires=">=3.8",

tenant_schemas_celery/app.py

@@ -1,5 +1,3 @@
-from __future__ import absolute_import
-
 try:
     from celery import Celery
 except ImportError:
@@ -80,7 +78,7 @@ class CeleryApp(Celery):
 
     def __init__(self, *args, **kwargs):
         kwargs.setdefault("task_cls", self.task_cls)
-        super(CeleryApp, self).__init__(*args, **kwargs)
+        super().__init__(*args, **kwargs)
 
     def create_task_cls(self):
         return self.subclass_with_self(
@@ -99,4 +97,4 @@ def _add_current_schema(self, kwds):
 
     def send_task(self, name, args=None, kwargs=None, **options):
         self._update_headers(options)
-        return super(CeleryApp, self).send_task(name, args=args, kwargs=kwargs, **options)
+        return super().send_task(name, args=args, kwargs=kwargs, **options)

tenant_schemas_celery/integration_test.py

@@ -1,5 +1,3 @@
-from __future__ import absolute_import
-
 import time
 
 import pytest

tenant_schemas_celery/registry_test.py

@@ -4,8 +4,8 @@
 from celery import Task
 
 from tenant_schemas_celery.task import TenantTask
-from .test_app import app
-from .test_tasks import get_schema_name, get_schema_from_class_task, SchemaClassTask
+from tenant_schemas_celery.test_app import app
+from tenant_schemas_celery.test_tasks import get_schema_name, get_schema_from_class_task, SchemaClassTask
 
 
 def test_get_schema_name_registration(transactional_db):

tenant_schemas_celery/task.py

@@ -1,5 +1,4 @@
-import celery
-from celery.app.task import Task
+from celery import Task
 from django.db import connection
 
 from tenant_schemas_celery.cache import SimpleCache
@@ -9,7 +8,7 @@
 
 class SharedTenantCache(SimpleCache):
     def __init__(self):
-        super(SharedTenantCache, self).__init__(storage=_shared_storage)
+        super().__init__(storage=_shared_storage)
 
 
 class TenantTask(Task):
@@ -37,7 +36,7 @@ def tenant_cache(cls):
 
     @classmethod
     def get_tenant_for_schema(cls, schema_name):
-        from .compat import get_tenant_model
+        from tenant_schemas_celery.compat import get_tenant_model
 
         missing = object()
         cache = cls.tenant_cache()
@@ -65,11 +64,5 @@ def _add_current_schema(self, kwds):
         kwds["_schema_name"] = kwds.get("_schema_name", connection.schema_name)
 
     def apply(self, args=None, kwargs=None, *arg, **kw):
-        if celery.VERSION[0] < 4:
-            kwargs = kwargs or {}
-            self._add_current_schema(kwargs)
-
-        else:
-            # Celery 4.0 introduced strong typing and the `headers` meta dict.
-            self._update_headers(kw)
-        return super(TenantTask, self).apply(args, kwargs, *arg, **kw)
+        self._update_headers(kw)
+        return super().apply(args, kwargs, *arg, **kw)

tenant_schemas_celery/test_app.py

@@ -1,7 +1,7 @@
 import os
 
 try:
-    from .app import CeleryApp
+    from tenant_schemas_celery.app import CeleryApp
 except ImportError:
     app = None
 else:

tenant_schemas_celery/test_scheduler.py

@@ -8,7 +8,7 @@
 from pytest import fixture, mark
 from tenant_schemas_celery.app import CeleryApp
 
-from .scheduler import (
+from tenant_schemas_celery.scheduler import (
     TenantAwarePersistentScheduler,
     TenantAwareSchedulerMixin,
 )

tenant_schemas_celery/test_utils.py

@@ -1,5 +1,3 @@
-from __future__ import absolute_import
-
 from django.core.exceptions import FieldDoesNotExist
 
 from test_app.shared.models import Client

test-compose.yml

@@ -6,18 +6,12 @@ services:
       - POSTGRES_PASSWORD=qwe123
       - POSTGRES_USER=tenant_celery
       - POSTGRES_DB=tenant_celery
-    logging:
-      driver: "none"
 
   rabbitmq:
     image: rabbitmq
-    logging:
-      driver: "none"
 
   redis:
     image: redis
-    logging:
-      driver: "none"
 
   app:
     image: "python:${PYTHON_VERSION:-3.8}-slim"

test_app/test_app/settings.py

@@ -160,7 +160,7 @@
 
 STATIC_URL = '/static/'
 
-CELERY_BROKER_URL = os.environ.get("BROKER_URL", 'amqp://tenants:tenants@localhost:5672/')
+CELERY_BROKER_URL = os.environ.get("BROKER_URL", 'amqp://guest:guest@localhost:5672/')
 CELERYBEAT_SCHEDULE = {
     'test-periodic-task': {
         'task': 'test_app.tenant.tasks.periodic_print_schema',