Abstract Django DB Models

[blbuford]

Hello,

We encountered an issue in our products where multiple Django services share the same database/schema with Django Celery Beat (DCB) installed. In this setup, Service A attempts to execute Service B’s periodic tasks and vice versa, even though they do not share the same codebase. To address this, we aimed to configure DCB to write to separate tables for each service.

I found an implementation in #534 and have updated it to work with the current mainline DCB. This feature was also requested in #904, and @auvipy mentioned it is on the DCB roadmap. Unit tests pass on my local Python environment, though I have not run the full test matrix.

This change has been very helpful for us, allowing us to do the following:

from django.utils.translation import gettext_lazy as _
from django_celery_beat.models.abstract import AbstractPeriodicTask
from django_celery_beat.querysets import PeriodicTaskQuerySet


class ProductAPeriodicTask(AbstractPeriodicTask):

    objects = PeriodicTaskQuerySet.as_manager()

    class Meta:
        db_table = 'django_celery_beat_periodictask_product_a'
        verbose_name = _('ProductA Periodic Task')
        verbose_name_plural = _('ProductA Periodic Tasks')
        abstract = False

and add the following to our settings.py:

CELERY_BEAT_PERIODICTASK_MODEL = 'core.ProductAPeriodicTask'

With this approach, DCB writes to a dedicated table for Product A.

Please let me know if there’s anything I can do to help move this forward or improve the PR for merging.

Thank you!

Best regards,
Brett

[codecov]

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 89.35%. Comparing base (0bac717) to head (2bf1a84).
Report is 1 commits behind head on main.

Additional details and impacted files

@@            Coverage Diff             @@
##             main     #911      +/-   ##
==========================================
+ Coverage   88.19%   89.35%   +1.16%     
==========================================
  Files          32       35       +3     
  Lines        1008     1118     +110     
  Branches      105      117      +12     
==========================================
+ Hits          889      999     +110     
  Misses        101      101              
  Partials       18       18              

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[blbuford]

Is there any progress or input on this?

[Copilot]

Pull Request Overview

This PR refactors the Django Celery Beat models to be fully abstract and configurable via settings, allowing each service to use its own database tables. Key changes include:

• Introduce helpers.py with functions to dynamically fetch models based on CELERY_BEAT_*_MODEL settings.
• Convert all core models in models/abstract.py to abstract base classes and add concrete implementations in models/generic.py.
• Update schedulers.py to use dynamic models and add tests & documentation for the new custom-model feature.

Reviewed Changes

Copilot reviewed 7 out of 8 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
t/unit/test_helpers.py	Add tests for helper functions fetching custom and default models
django_celery_beat/helpers.py	New dynamic model lookup functions based on project settings
django_celery_beat/schedulers.py	Swap static model imports for dynamic lookup via helpers
django_celery_beat/models/abstract.py	Convert concrete models to abstract classes; update imports
django_celery_beat/models/generic.py	Provide concrete implementations inheriting from abstract classes
django_celery_beat/models/init.py	Expose new generic classes in package API
README.rst	Document how to use custom abstract models in settings

Comments suppressed due to low confidence (2)

django_celery_beat/models/generic.py:31

• [nitpick] The ClockedSchedule docstring is duplicated from IntervalSchedule and mentions intervals. Consider updating it to accurately describe clocked schedules.

    """Schedule with a fixed interval."""

t/unit/test_helpers.py:47

• Using .upper() on the model name will generate incorrect class names (e.g. CRONTABSCHEDULE instead of CrontabSchedule). Use the exact class name casing when constructing app_label.model_name for lookup.

    model_setting_name = f"CELERY_BEAT_{model_name.upper()}_MODEL"

[Copilot]

Fix the typo in the docstring: change Interal to Internal.

Suggested change

	"""Interal task scheduling class."""
	"""Internal task scheduling class."""

[Copilot]

[nitpick] The model‐fetching functions in this file have nearly identical logic. Consider extracting the common pattern into a single helper that takes the setting name and default import path to reduce duplication.

Suggested change

	def crontabschedule_model():
	"""Return the CrontabSchedule model that is active in this project."""
	if not hasattr(settings, "CELERY_BEAT_CRONTABSCHEDULE_MODEL"):
	from .models.generic import CrontabSchedule
	return CrontabSchedule
	try:
	return apps.get_model(settings.CELERY_BEAT_CRONTABSCHEDULE_MODEL)
	except ValueError:
	raise ImproperlyConfigured(
	"CELERY_BEAT_CRONTABSCHEDULE_MODEL must be of the form "
	"'app_label.model_name'"
	)
	except LookupError:
	raise ImproperlyConfigured(
	"CELERY_BEAT_CRONTABSCHEDULE_MODEL refers to model "
	f"'{settings.CELERY_BEAT_CRONTABSCHEDULE_MODEL}' that has not "
	"been installed"
	)
	def intervalschedule_model():
	"""Return the IntervalSchedule model that is active in this project."""
	if not hasattr(settings, "CELERY_BEAT_INTERVALSCHEDULE_MODEL"):
	from .models.generic import IntervalSchedule
	return IntervalSchedule
	try:
	return apps.get_model(settings.CELERY_BEAT_INTERVALSCHEDULE_MODEL)
	except ValueError:
	raise ImproperlyConfigured(
	"CELERY_BEAT_INTERVALSCHEDULE_MODEL must be of the form "
	"'app_label.model_name'"
	)
	except LookupError:
	raise ImproperlyConfigured(
	"CELERY_BEAT_INTERVALSCHEDULE_MODEL refers to model "
	f"'{settings.CELERY_BEAT_INTERVALSCHEDULE_MODEL}' that has not "
	"been installed"
	)
	def get_model_or_default(setting_name, default_model_path):
	"""Fetch a model based on a setting name or return the default model."""
	if not hasattr(settings, setting_name):
	module_path, model_name = default_model_path.rsplit(".", 1)
	module = __import__(module_path, fromlist=[model_name])
	return getattr(module, model_name)
	try:
	return apps.get_model(getattr(settings, setting_name))
	except ValueError:
	raise ImproperlyConfigured(
	f"{setting_name} must be of the form 'app_label.model_name'"
	)
	except LookupError:
	raise ImproperlyConfigured(
	f"{setting_name} refers to a model "
	f"'{getattr(settings, setting_name)}' that has not been installed"
	)
	def crontabschedule_model():
	"""Return the CrontabSchedule model that is active in this project."""
	return get_model_or_default(
	"CELERY_BEAT_CRONTABSCHEDULE_MODEL",
	"django_celery_beat.models.generic.CrontabSchedule"
	)
	def intervalschedule_model():
	"""Return the IntervalSchedule model that is active in this project."""
	return get_model_or_default(
	"CELERY_BEAT_INTERVALSCHEDULE_MODEL",
	"django_celery_beat.models.generic.IntervalSchedule"
	)