fork/django-q2
1
0
Fork 0
mirror of https://github.com/django-q2/django-q2.git synced 2026-10-06 04:48:12 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' into refactor

Browse Source
This commit is contained in:
Stan Triepels
2023-04-11 15:04:25 +02:00
committed by GitHub
parent e3517bd4c6 322a53f06d
commit ac83602a76
24 changed files with 525 additions and 359 deletions
Download Patch File Download Diff File Expand all files Collapse all files
django_q/__init__.py
+1 -1
View File
@@ -1,6 +1,6 @@
import django
VERSION = (1, 4, 11)
VERSION = (1, 5, 1)
if django.VERSION < (3, 2):
default_app_config = "django_q.apps.DjangoQConfig"
django_q/admin.py
+8 -5
View File
@@ -13,7 +13,7 @@ from django_q.tasks import async_task
class TaskAdmin(admin.ModelAdmin):
"""model admin for success tasks."""
list_display = ("name", "group", "func", "started", "stopped", "time_taken")
list_display = ("name", "group", "func", "cluster", "started", "stopped", "time_taken")
def has_add_permission(self, request):
"""Don't allow adds."""
@@ -26,7 +26,7 @@ class TaskAdmin(admin.ModelAdmin):
search_fields = ("name", "func", "group")
readonly_fields = []
list_filter = ("group",)
list_filter = ("group", "cluster")
def get_readonly_fields(self, request, obj=None):
"""Set all fields readonly."""
@@ -36,7 +36,8 @@ class TaskAdmin(admin.ModelAdmin):
def retry_failed(FailAdmin, request, queryset):
"""Submit selected tasks back to the queue."""
for task in queryset:
async_task(task.func, *task.args or (), hook=task.hook, **task.kwargs or {})
async_task(task.func, *task.args or (), hook=task.hook,
group=task.group, cluster=task.cluster, **task.kwargs or {})
task.delete()
@@ -46,7 +47,7 @@ retry_failed.short_description = _("Resubmit selected tasks to queue")
class FailAdmin(admin.ModelAdmin):
"""model admin for failed tasks."""
list_display = ("name", "group", "func", "started", "stopped", "short_result")
list_display = ("name", "group", "func", "cluster", "started", "stopped", "short_result")
def has_add_permission(self, request):
"""Don't allow adds."""
@@ -54,7 +55,7 @@ class FailAdmin(admin.ModelAdmin):
actions = [retry_failed]
search_fields = ("name", "func", "group")
list_filter = ("group",)
list_filter = ("group", "cluster")
readonly_fields = []
def get_readonly_fields(self, request, obj=None):
@@ -123,6 +124,8 @@ class QueueAdmin(admin.ModelAdmin):
"""queue admin for ORM broker"""
list_display = ("id", "key", "name", "group", "func", "lock", "task_id")
fields = ("key", "lock", "task_id", "name", "group", "func", "args", "kwargs", "q_options")
readonly_fields = fields[2:]
def save_model(self, request, obj, form, change):
obj.save(using=Conf.ORM)
django_q/brokers/__init__.py
+6 -3
View File
@@ -7,7 +7,9 @@ from django_q.conf import Conf
class Broker:
def __init__(self, list_key: str = Conf.PREFIX):
def __init__(self, list_key: str = None):
# With same BROKER_CLASS, `list_key` is just a synonym for `queue_name` except for RedisBroker
list_key = list_key or Conf.CLUSTER_NAME
self.connection = self.get_connection(list_key)
self.list_key = list_key
self.cache = self.get_cache()
@@ -151,7 +153,7 @@ class Broker:
return None
@staticmethod
def get_connection(list_key: str = Conf.PREFIX):
def get_connection(list_key: str = None):
"""
Gets a connection to the broker
:param list_key: Optional queue name
@@ -160,13 +162,14 @@ class Broker:
return 0
def get_broker(list_key: str = Conf.PREFIX) -> Broker:
def get_broker(list_key: str = None) -> Broker:
"""
Gets the configured broker type
:param list_key: optional queue name
:type list_key: str
:return: a broker instance
"""
list_key = list_key or Conf.CLUSTER_NAME
# custom
if Conf.BROKER_CLASS:
module, func = Conf.BROKER_CLASS.rsplit(".", 1)
django_q/brokers/aws_sqs.py
+2 -2
View File
@@ -8,7 +8,7 @@ QUEUE_DOES_NOT_EXIST = "AWS.SimpleQueueService.NonExistentQueue"
class Sqs(Broker):
def __init__(self, list_key: str = Conf.PREFIX):
def __init__(self, list_key: str = None):
self.sqs = None
super(Sqs, self).__init__(list_key)
self.queue = self.get_queue()
@@ -77,7 +77,7 @@ class Sqs(Broker):
return "AWS SQS"
@staticmethod
def get_connection(list_key: str = Conf.PREFIX) -> Session:
def get_connection(list_key: str = None) -> Session:
config = Conf.SQS
if "aws_region" in config:
config["region_name"] = config["aws_region"]
django_q/brokers/ironmq.py
+2 -1
View File
@@ -46,6 +46,7 @@ class IronMQBroker(Broker):
return self.delete(task_id)
@staticmethod
def get_connection(list_key: str = Conf.PREFIX) -> Queue:
def get_connection(list_key: str = None) -> Queue:
list_key = list_key or Conf.CLUSTER_NAME
ironmq = IronMQ(name=None, **Conf.IRON_MQ)
return ironmq.queue(queue_name=list_key)
django_q/brokers/mongo.py
+2 -2
View File
@@ -15,7 +15,7 @@ def _timeout():
class Mongo(Broker):
def __init__(self, list_key=Conf.PREFIX):
def __init__(self, list_key: str = None):
super(Mongo, self).__init__(list_key)
self.collection = self.get_collection()
@@ -24,7 +24,7 @@ class Mongo(Broker):
self.collection = self.get_collection()
@staticmethod
def get_connection(list_key: str = Conf.PREFIX) -> MongoClient:
def get_connection(list_key: str = None) -> MongoClient:
return MongoClient(**Conf.MONGO)
def get_collection(self):
django_q/brokers/orm.py
+3 -2
View File
@@ -16,7 +16,7 @@ def _timeout():
class ORM(Broker):
@staticmethod
def get_connection(list_key: str = Conf.PREFIX):
def get_connection(list_key: str = None):
if transaction.get_autocommit(
using=Conf.ORM
): # Only True when not in an atomic block
@@ -55,8 +55,9 @@ class ORM(Broker):
self.delete(task_id)
def enqueue(self, task):
# list_key might be null (e.g. in a test setup) but OrmQ.key has not-null constraint
package = self.get_connection().create(
key=self.list_key, payload=task, lock=timezone.now()
key=self.list_key or Conf.CLUSTER_NAME, payload=task, lock=timezone.now()
)
return package.pk
django_q/brokers/redis_broker.py
+3 -2
View File
@@ -11,7 +11,8 @@ except ImportError:
class Redis(Broker):
def __init__(self, list_key: str = Conf.PREFIX):
def __init__(self, list_key: str = None):
list_key = list_key or Conf.CLUSTER_NAME
super(Redis, self).__init__(list_key=f"django_q:{list_key}:q")
def enqueue(self, task):
@@ -57,7 +58,7 @@ class Redis(Broker):
return self.connection.mget(keys)
@staticmethod
def get_connection(list_key: str = Conf.PREFIX) -> Redis:
def get_connection(list_key: str = None) -> Redis:
if django_redis and Conf.DJANGO_REDIS:
return django_redis.get_redis_connection(Conf.DJANGO_REDIS)
if isinstance(Conf.REDIS, str):
django_q/cluster.py
+12 -7
View File
@@ -38,14 +38,16 @@ from django_q.status import Stat, Status
class Cluster:
def __init__(self, broker: Broker = None):
self.broker = broker or get_broker()
# Cluster do not need an init or default broker except for testing,
# The sentinel will create a broker for cluster and utilize ALT_CLUSTERS config in Conf.
self.broker = broker # DON'T USE get_broker() to set a default broker here.
self.sentinel = None
self.stop_event = None
self.start_event = None
self.pid = current_process().pid
self.cluster_id = uuid.uuid4()
self.host = socket.gethostname()
self.timeout = Conf.TIMEOUT
self.timeout = None
signal.signal(signal.SIGTERM, self.sig_handler)
signal.signal(signal.SIGINT, self.sig_handler)
@@ -131,7 +133,7 @@ class Sentinel:
start_event,
cluster_id,
broker=None,
timeout=Conf.TIMEOUT,
timeout=None,
start=True,
):
# Make sure we catch signals for the pool
@@ -146,7 +148,7 @@ class Sentinel:
self.tob = timezone.now()
self.stop_event = stop_event
self.start_event = start_event
self.timeout = timeout
self.timeout = timeout or Conf.TIMEOUT
self.event_out = Event()
logger.info(
_("%(name)s main at %(id)s") % {"name": self.name, "id": current_process().pid}
@@ -154,6 +156,10 @@ class Sentinel:
if start:
self.start()
def queue_name(self):
# multi-queue: cluster name is (broker's) queue_name
return self.broker.list_key if self.broker else '--'
def start(self):
self.broker.ping()
self.spawn_cluster()
@@ -191,13 +197,13 @@ class Sentinel:
_("%(name)s guarding cluster %(cluster_name)s")
% {
"name": current_process().name,
"cluster_name": humanize(self.cluster_id.hex),
"cluster_name": humanize(self.cluster_id.hex) + f" [{self.queue_name()}]",
}
)
self.start_event.set()
logger.info(
_("Q Cluster %(cluster_name)s running.")
% {"cluster_name": humanize(self.cluster_id.hex)}
% {"cluster_name": humanize(self.cluster_id.hex) + f" [{self.queue_name()}]"}
)
counter = 0
# Guard loop. Runs at least once
@@ -274,7 +280,6 @@ class Sentinel:
for worker in self.pool.workers:
worker.process.kill()
logger.debug(_("All tasks were processed and workers where stopped"))
self.monitor.add_task("STOP")
django_q/conf.py
+19 -1
View File
@@ -39,10 +39,22 @@ class Conf:
"""
try:
conf = settings.Q_CLUSTER
conf = settings.Q_CLUSTER.copy()
except AttributeError:
conf = {}
_Q_CLUSTER_NAME = os.getenv("Q_CLUSTER_NAME")
if _Q_CLUSTER_NAME and _Q_CLUSTER_NAME != conf.get("name") and \
_Q_CLUSTER_NAME != conf.get("cluster_name"):
conf["cluster_name"] = _Q_CLUSTER_NAME
alt_conf = conf.pop("ALT_CLUSTERS")
if isinstance(alt_conf, dict):
alt_conf = alt_conf.get(_Q_CLUSTER_NAME)
if isinstance(alt_conf, dict):
alt_conf.pop('name', None)
alt_conf.pop('cluster_name', None)
conf.update(alt_conf)
# Redis server configuration . Follows standard redis keywords
REDIS = conf.get("redis", {})
@@ -70,8 +82,14 @@ class Conf:
MONGO_DB = conf.get("mongo_db", None)
# Name of the cluster or site. For when you run multiple sites on one redis server
# It's also the `salt` for signing OrmQ, and part of the Redis stats caching key
# For all clusters in one site, PREFIX should be the same value to be able to decrypt payloads
PREFIX = conf.get("name", "default")
# Support alternative cluster name to use multiple queues in one site.
# cluster name and queue name are interchangeable, same thing.
CLUSTER_NAME = conf.get("cluster_name", PREFIX)
# Log output level
LOG_LEVEL = conf.get("log_level", "DEBUG")
django_q/management/commands/qcluster.py
+14
View File
@@ -2,6 +2,7 @@ from django.core.management.base import BaseCommand
from django.utils.translation import gettext as _
from django_q.cluster import Cluster
import os
class Command(BaseCommand):
@@ -16,8 +17,21 @@ class Command(BaseCommand):
default=False,
help="Run once and then stop.",
)
parser.add_argument(
"-n",
"--name",
dest="cluster_name",
default=None,
help="Set alternative cluster name instead of the name in Q_CLUSTER settings (for multi-queue setup). "
"On Linux you should set name through `Q_CLUSTER_NAME=cluster_name python manage.py qcluster` instead."
)
def handle(self, *args, **options):
# Set alternative cluster_name before creating the cluster (cluster_name is broker's queue_name, too)
cluster_name = options.get("cluster_name")
if cluster_name:
os.environ["Q_CLUSTER_NAME"] = cluster_name
q = Cluster()
q.start()
if options.get("run_once", False):
django_q/migrations/0017_task_cluster_alter.py
+43
View File
@@ -0,0 +1,43 @@
# Generated by Django 4.1.5 on 2023-03-07 12:18
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("django_q", "0016_schedule_intended_date_kwarg"),
]
operations = [
migrations.AddField(
model_name="task",
name="cluster",
field=models.CharField(blank=True, default=None, max_length=100, null=True),
),
migrations.AlterField(
model_name="ormq",
name="key",
field=models.CharField(
help_text="Name of the target cluster", max_length=100
),
),
migrations.AlterField(
model_name="ormq",
name="lock",
field=models.DateTimeField(
help_text="Prevent any cluster from pulling until", null=True
),
),
migrations.AlterField(
model_name="schedule",
name="cluster",
field=models.CharField(
blank=True,
default=None,
help_text="Name of the target cluster",
max_length=100,
null=True,
),
),
]
django_q/models.py
+25 -6
View File
@@ -12,6 +12,7 @@ from django.utils import timezone
from django.utils.timezone import is_aware
from django.utils.html import format_html
from django.utils.translation import gettext_lazy as _
from django.utils.functional import cached_property
# External
from picklefield import PickledObjectField
@@ -32,6 +33,7 @@ class Task(models.Model):
kwargs = PickledObjectField(null=True, protocol=-1)
result = PickledObjectField(null=True, protocol=-1)
group = models.CharField(max_length=100, editable=False, null=True)
cluster = models.CharField(max_length=100, default=None, null=True, blank=True)
started = models.DateTimeField(editable=False)
stopped = models.DateTimeField(editable=False)
success = models.BooleanField(default=True, editable=False)
@@ -214,7 +216,10 @@ class Schedule(models.Model):
help_text=_("Cron expression"),
)
task = models.CharField(max_length=100, null=True, editable=False)
cluster = models.CharField(max_length=100, default=None, null=True, blank=True)
cluster = models.CharField(
max_length=100, default=None, null=True, blank=True,
help_text=_("Name of the target cluster")
)
intended_date_kwarg = models.CharField(
max_length=100,
null=True,
@@ -329,13 +334,17 @@ class Schedule(models.Model):
class OrmQ(models.Model):
key = models.CharField(max_length=100)
key = models.CharField(max_length=100, help_text=_("Name of the target cluster"))
payload = models.TextField()
lock = models.DateTimeField(null=True)
lock = models.DateTimeField(null=True, help_text=_("Prevent any cluster from pulling until"))
@property
@cached_property
def task(self):
return SignedPackage.loads(self.payload)
try:
return SignedPackage.loads(self.payload)
except Exception as e:
return {"id": "*" + e.__class__.__name__}
def func(self):
if isinstance(self.task, dict):
@@ -355,7 +364,17 @@ class OrmQ(models.Model):
def group(self):
if isinstance(self.task, dict):
return self.task.get("group", "")
return self.task.group
return self.task.group)
def args(self):
return self.task.get("args")
def kwargs(self):
return self.task.get("kwargs")
def q_options(self):
exclude = {"id", "name", "group", "func", "args", "kwargs"}
return {k: v for k, v in self.task.items() if k not in exclude}
class Meta:
app_label = "django_q"
django_q/tasks.py
+5 -2
View File
@@ -32,6 +32,7 @@ def async_task(func, *args, **kwargs):
"iter_cached",
"chain",
"broker",
"cluster",
"timeout",
)
q_options = given_kwargs.pop("q_options", {})
@@ -46,7 +47,7 @@ def async_task(func, *args, **kwargs):
)
# don't serialize the broker
broker = given_kwargs.pop("broker", None) or q_options.pop("broker", None) or get_broker()
broker = given_kwargs.pop("broker", None) or q_options.pop("broker", None) or get_broker(task.get("cluster")) or get_broker()
print(broker.list_key)
@@ -65,7 +66,7 @@ def async_task(func, *args, **kwargs):
return _sync(pack)
# push it
enqueue_id = broker.enqueue(pack)
logger.info(f"Enqueued {enqueue_id}")
logger.info(f"Enqueued [{broker.list_key}] {enqueue_id}")
logger.debug(f"Pushed {tag}")
return task.id
@@ -265,6 +266,7 @@ def fetch_cached(task_id, wait=0, broker=None):
hook=task.hook,
args=task.args,
kwargs=task.kwargs,
cluster=task.get("cluster"),
started=task.started_at,
stopped=task.finished_at,
result=task.result,
@@ -335,6 +337,7 @@ def fetch_group_cached(group_id, failures=True, wait=0, count=None, broker=None)
hook=task.hook,
args=task.args,
kwargs=task.kwargs,
cluster=task.get("cluster"),
started=task.started_at,
stopped=task.finished_at,
result=task.result,
django_q/tests/test_scheduler.py
+2 -2
View File
@@ -346,7 +346,7 @@ def test_scheduler(broker, monkeypatch):
assert schedule.next_run.date() == (timezone.now() + timedelta(weeks=2)).date()
broker.delete_queue()
monkeypatch.setattr(Conf, "PREFIX", "some_cluster_name")
monkeypatch.setattr(Conf, "CLUSTER_NAME", "some_cluster_name")
# create a schedule on another cluster
schedule = create_schedule(
"math.copysign",
@@ -366,7 +366,7 @@ def test_scheduler(broker, monkeypatch):
# queue must be empty
assert len(tasks) == 0
monkeypatch.setattr(Conf, "PREFIX", "default")
monkeypatch.setattr(Conf, "CLUSTER_NAME", "default")
# create a schedule on the same cluster
schedule = create_schedule(
"math.copysign",
django_q/utils.py
-2
View File
@@ -45,8 +45,6 @@ def add_years(d, years):
return d.replace(year=new_date.year, month=new_date.month, day=new_date.day)
def localtime(value=None) -> datetime:
"""Override for timezone.localtime to deal with naive times and local times"""
if settings.USE_TZ: