Skip to main content

Celery integration for django-tenant-schemas and django-tenants

Project description

tenant-schemas-celery

Celery application implementation that allows celery tasks to cooperate with the django-tenants package.

This project might not seem frequently updated, but it just has all the functionality needed. Issues and questions are answered quickly.

Installation

   $ pip install tenant-schemas-celery
   $ pip install django-tenants

Usage

  • Define a celery app using given CeleryApp class.
   import os
   os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'app.settings')

   from django.conf import settings

   from tenant_schemas_celery.app import CeleryApp as TenantAwareCeleryApp

   app = TenantAwareCeleryApp()
   app.config_from_object('django.conf:settings')
   app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)

This assumes a fresh Celery 5.4.0 application. For previous versions, the key is to create a new CeleryApp instance that will be used to access task decorator from.

  • Replace your @task decorator with @app.task
   from celery import shared_task
   from django.db import connection
   from myproject.celery import app
   from tenant_schemas_celery.task import TenantTask

   @app.task
   def my_task():
      print(connection.schema_name)

   @shared_task(base=TenantTask, bind=True)
   def my_shared_task():
      print("foo")
  • Run celery worker (myproject.celery is where you've defined the app variable)
    $ celery worker -A myproject.celery
  • Post registered task. The schema name will get automatically added to the task's arguments.
   from myproject.tasks import my_task
   my_task.delay()

The TenantTask class transparently inserts current connection's schema into the task's kwargs. The schema name is then popped from the task's kwargs in task_prerun signal handler, and the connection's schema is changed accordingly.

Multiple databases support

Inside your celery tasks you might be working with multiple databases. You might want to change the schema for all of the connections, or just a subset of them.

You can now use the CELERY_TASK_TENANT_CACHE_SECONDS django setting, or TASK_TENANT_CACHE_SECONDS celery setting, or the tenant_databases attribute of the TenantTask to a list of database names (the key in the settings.DATABASES dictionary).

If not set, the settings defaults to ["default"].

Tenant objects cache

Every time a celery task is executed, the tenant object of the connection object is being refetched. For some use cases, this can introduce significant performance hit.

In such scenarios, you can pass tenant_cache_seconds argument to the @app.task() decorator. This will cause the tenant objects to be cached for given period of time. 0 turns this off. You can also enable cache globally by setting celery's TASK_TENANT_CACHE_SECONDS (app-specific, usually it's CELERY_TASK_TENANT_CACHE_SECONDS).

@app.task(tenant_cache_seconds=30)
def some_task():
    ...

Celery beat integration

In order to run celery beat tasks in a multi-tenant environment, you've got the following options:

  • Use a dispatching task that will send a task for each tenant
  • Use a custom scheduler
  • Use django-celery-beat package

i.e: Let's say that you would like to run a reset_remaining_jobs tasks periodically, for every tenant that you have.

Dispatcher task pattern

You can schedule one dispatcher task that will iterate over all schemas and send that task within the schema's context:

from django_tenants.utils import get_tenant_model, tenant_context
from django_tenant_schemas.utils import get_tenant_model, tenant_context

@app.task
def reset_remaining_jobs_in_all_schemas():
    for tenant in get_tenant_model().objects.exclude(schema_name='public'):
        with tenant_context(tenant):
            reset_remaining_jobs_in_schema.delay()

@app.task
def reset_remaining_jobs_in_schema():
    <do some logic>

The reset_remaining_jobs_in_all_schemas task (called the dispatch task) should be registered in your celery beat schedule. The reset_remaining_jobs_in_schema task should be called from the dispatch task.

That way you have full control over which schemas the task should be scheduled in.

Note: All tasks will be scheduled immediately across all tenants, possibly creating the thundering herd problem.

Custom scheduler

If you are using the standard Scheduler or PersistentScheduler classes provided by celery, you can transition to using this package's TenantAwareScheduler or TenantAwarePersistentScheduler classes. You should then specify the scheduler you want to use in the celery beat config or your invocation to beat. i.e:

celery -A proj beat --scheduler=tenant_schemas_celery.scheduler.TenantAwareScheduler
Caveats
  • There's a chance that celery beat will try to run a task for a newly created tenant before its migrations are ready, which could potentially lead to deadlocks. This is specially true with big projects with a lot of migrations and very frequent tasks (i.e: every minute). In order to mitigate it, one could do the following:

    1. Subclass any of TenantAwareScheduler or TenantAwarePersistentScheduler and override the get_queryset method to match your definition of "ready" tenants. For example, imagine that you had a ready flag in your tenant model. You could do the following:
    # tenants_app/scheduler.py
    class MyTenantAwareScheduler(TenantAwareScheduler):
        @classmethod
        def get_queryset(cls):
            return super().get_queryset().filter(ready=True)
    
    1. Use the new scheduler in your celery beat config or invocation:
    celery -A proj beat --scheduler=tenants_app.scheduler.MyTenantAwareScheduler
    
  • TenantAwareSchedulerMixin uses a subclass of SchedulerEntry that allows the user to provide specific schemas to run a task on. This might prove useful if you have a task you only want to run in the public schema or to a subset of your tenants. In order to set that, you must configure tenant_schemas in the tasks definition as such:

app.conf.beat_schedule = {
    "my-task": {
        "task": "myapp.tasks.my_task",
        "schedule": schedules.crontab(minute="*/5"),
        "tenant_schemas": ["public"]
    }
}

django-celery-beat integration

You can use the tenant_schemas_celery.db_scheduler.TenantAwareDatabaseScheduler scheduler to integrate the django-celery-beat package with multiple tenants. The scheduler will pull all periodic tasks defined in every tenant schema and public schema, and add them to the scheduler's registry.

Because tasks from schemas are merged into a single registry, tasks' names must be unique. For example, you can add the @<schema_name> suffix to each of the entries:

# In public schema
PeriodicTask.objects.create(name="my_task@public", task="my_app.tasks.my_task")
# In schema "tenant"
PeriodicTask.objects.create(name="my_task@tenant", task="my_app.tasks.my_task")

The package will raise ValueError if entries with conflicting names are detected.

Note: Since every periodic task defined in the database can have different schedule (incl. offset), this allows you to avoid the thundering herd problem.

Compatibility changes

The >=4.0 series drop support for Python 3.8 and Celery<5.

The >=3.0 series drop support for Python 3.7.

The >=2.1 series drop support for tenant-schemas. It hasn't been maintainted for a long time.

The 2.x series support Python>=3.7.

The 1.x series support Python>=3.6. Python 3.6 reached EOL 2021-12.

The 0.x series are the last one to support Python<3.6.

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

tenant_schemas_celery-4.0.1.tar.gz (18.9 kB view details)

Uploaded Source

Built Distribution

tenant_schemas_celery-4.0.1-py3-none-any.whl (21.4 kB view details)

Uploaded Python 3

File details

Details for the file tenant_schemas_celery-4.0.1.tar.gz.

File metadata

  • Download URL: tenant_schemas_celery-4.0.1.tar.gz
  • Upload date:
  • Size: 18.9 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.1.0 CPython/3.9.21

File hashes

Hashes for tenant_schemas_celery-4.0.1.tar.gz
Algorithm Hash digest
SHA256 8b8f055fcd82aa53274c09faf88653a935241518d93b86ab2d43a3df3b70c7f8
MD5 8a29c1f9ec51a5c65d249822585bcb96
BLAKE2b-256 19f8cf055bf171b5d83d6fe96f1840fba90d3d274be2b5c35cd21b873302b128

See more details on using hashes here.

File details

Details for the file tenant_schemas_celery-4.0.1-py3-none-any.whl.

File metadata

File hashes

Hashes for tenant_schemas_celery-4.0.1-py3-none-any.whl
Algorithm Hash digest
SHA256 d06a3ff6956db3a95168ce2051b7bff2765f9ce0d070e14df92f07a2b60ae0a0
MD5 22ee10d68dfeb8eab5cba906641ae87e
BLAKE2b-256 e9a8fd663c461550d6fedfb24e987acc1557ae5b6615ca08fc6c70dbaaa88aa5

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page