diff --git a/procrastinate/blueprints.py b/procrastinate/blueprints.py index a8ea2622b..ae306c7b7 100644 --- a/procrastinate/blueprints.py +++ b/procrastinate/blueprints.py @@ -86,10 +86,12 @@ def _check_stack(self): ) def _register_task(self, task: "tasks.Task") -> None: + from procrastinate import tasks + # Each call to _add_task may raise TaskAlreadyRegistered. # We're using an intermediary dict to make sure that if the registration # is interrupted midway though, self.tasks is left unmodified. - to_add = {} + to_add: Dict[str, tasks.Task] = {} self._add_task(task=task, name=task.name, to=to_add) for alias in task.aliases: @@ -150,9 +152,11 @@ def add_tasks_from(self, blueprint: "Blueprint", *, namespace: str) -> None: raise exceptions.TaskAlreadyRegistered( f"A namespace named {namespace} was already registered" ) - # Modify existing tasks and other blueprint to add the namespace + # Modify existing tasks and other blueprint to add the namespace, and + # set the blueprint for task in set(blueprint.tasks.values()): task.add_namespace(namespace) + task.blueprint = self blueprint.tasks = new_tasks # Finally, add the namespaced tasks to this namespace @@ -226,7 +230,7 @@ def _wrap(func: Callable[..., "tasks.Task"]): task = tasks.Task( func, - app=None, + blueprint=self, queue=queue, lock=lock, queueing_lock=queueing_lock, diff --git a/procrastinate/tasks.py b/procrastinate/tasks.py index 9c959d883..803fc6cad 100644 --- a/procrastinate/tasks.py +++ b/procrastinate/tasks.py @@ -2,7 +2,8 @@ import logging from typing import Any, Callable, Dict, List, Optional -from procrastinate import app, exceptions, jobs, manager +from procrastinate import app as app_module +from procrastinate import blueprints, exceptions, jobs, manager from procrastinate import retry as retry_module from procrastinate import types, utils @@ -72,7 +73,7 @@ def __init__( self, func: Callable, *, - app: Optional[app.App], + blueprint: blueprints.Blueprint, # task naming name: Optional[str] = None, aliases: Optional[List[str]] = None, @@ -85,7 +86,7 @@ def __init__( queueing_lock: Optional[str] = None, ): self.queue = queue - self.app = app + self.blueprint = blueprint self.func: Callable = func self.aliases = aliases if aliases else [] self.retry_strategy = retry_module.get_retry_strategy(retry) @@ -177,12 +178,14 @@ def configure( ValueError If you try to define both schedule_at and schedule_in """ - if self.app is None: + if not isinstance(self.blueprint, app_module.App): raise exceptions.UnboundTaskError + app = self.blueprint + return configure_task( name=self.name, - job_manager=self.app.job_manager, + job_manager=app.job_manager, lock=lock if lock is not None else self.lock, queueing_lock=( queueing_lock if queueing_lock is not None else self.queueing_lock