diff --git a/django_q/cluster.py b/django_q/cluster.py index f3c1bce..0a0da9a 100644 --- a/django_q/cluster.py +++ b/django_q/cluster.py @@ -9,11 +9,6 @@ from time import sleep from django import core, db from django.apps.registry import apps -from django_q.monitor import monitor -from django_q.pusher import pusher -from django_q.scheduler import scheduler -from django_q.worker import worker - try: apps.check_apps_ready() except core.exceptions.AppRegistryNotReady: @@ -25,16 +20,15 @@ from django.utils import timezone from django.utils.translation import gettext_lazy as _ # Local -import django_q.tasks from django_q.brokers import Broker, get_broker from django_q.conf import Conf, get_ppid, logger, psutil, setproctitle from django_q.humanhash import humanize -from django_q.models import Schedule, Success, Task +from django_q.monitor import monitor +from django_q.pusher import pusher from django_q.queues import Queue -from django_q.signing import BadSignature, SignedPackage +from django_q.scheduler import scheduler from django_q.status import Stat, Status - -from .utils import get_func_repr +from django_q.worker import worker class Cluster: diff --git a/django_q/monitor.py b/django_q/monitor.py index da7e3db..6f9b165 100644 --- a/django_q/monitor.py +++ b/django_q/monitor.py @@ -1,15 +1,23 @@ from multiprocessing.process import current_process from multiprocessing.queues import Queue -from django import db +from django import core, db from django.utils.translation import gettext_lazy as _ +from django.apps.registry import apps + +try: + apps.check_apps_ready() +except core.exceptions.AppRegistryNotReady: + import django + + django.setup() -import django_q.tasks from django_q.brokers import Broker, get_broker from django_q.conf import Conf, logger, setproctitle from django_q.models import Success, Task from django_q.signals import post_execute from django_q.signing import SignedPackage +from django_q.tasks import async_chain from django_q.utils import close_old_django_connections, get_func_repr try: @@ -77,7 +85,7 @@ def save_task(task, broker: Broker): return # enqueues next in a chain if task.get("chain", None): - django_q.tasks.async_chain( + async_chain( task["chain"], group=task["group"], cached=task["cached"], @@ -185,7 +193,7 @@ def save_cached(task, broker: Broker): broker.cache.set(group_key, group_list, timeout) # async_task next in a chain if task.get("chain", None): - django_q.tasks.async_chain( + async_chain( task["chain"], group=group, cached=task["cached"], diff --git a/django_q/pusher.py b/django_q/pusher.py index 503274e..405ca98 100644 --- a/django_q/pusher.py +++ b/django_q/pusher.py @@ -3,7 +3,16 @@ from multiprocessing.process import current_process from multiprocessing.queues import Queue from time import sleep +from django import core from django.utils.translation import gettext_lazy as _ +from django.apps.registry import apps + +try: + apps.check_apps_ready() +except core.exceptions.AppRegistryNotReady: + import django + + django.setup() from django_q.brokers import Broker, get_broker from django_q.conf import Conf, logger diff --git a/django_q/scheduler.py b/django_q/scheduler.py index 10aaa98..88bd6b6 100644 --- a/django_q/scheduler.py +++ b/django_q/scheduler.py @@ -1,15 +1,24 @@ import ast from multiprocessing.process import current_process -from django import db +from django import core, db from django.utils import timezone from django.utils.translation import gettext_lazy as _ -import django_q.tasks +from django.apps.registry import apps + +try: + apps.check_apps_ready() +except core.exceptions.AppRegistryNotReady: + import django + + django.setup() + from django_q.brokers import Broker, get_broker from django_q.conf import Conf, logger from django_q.humanhash import humanize from django_q.models import Schedule +from django_q.tasks import async_task from django_q.utils import close_old_django_connections, localtime @@ -86,7 +95,7 @@ def scheduler(broker: Broker = None): q_options["broker"] = broker q_options["group"] = q_options.get("group", s.name or s.id) kwargs["q_options"] = q_options - s.task = django_q.tasks.async_task(s.func, *args, **kwargs) + s.task = async_task(s.func, *args, **kwargs) # log it if not s.task: logger.error( diff --git a/django_q/worker.py b/django_q/worker.py index fc627fc..e842f19 100644 --- a/django_q/worker.py +++ b/django_q/worker.py @@ -4,8 +4,17 @@ from multiprocessing import Value from multiprocessing.process import current_process from multiprocessing.queues import Queue +from django import core from django.utils import timezone from django.utils.translation import gettext_lazy as _ +from django.apps.registry import apps + +try: + apps.check_apps_ready() +except core.exceptions.AppRegistryNotReady: + import django + + django.setup() from django_q.conf import Conf, error_reporter, logger, resource, setproctitle from django_q.signals import post_spawn, pre_execute